[streaming] BatchForwardLocal updated with Tuple

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/batch/BatchForwardSink.java

@@ -17,16 +17,15 @@ package eu.stratosphere.streaming.examples.batch;
 
 import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.types.StringValue;
 
 public class BatchForwardSink extends UserSinkInvokable {
 
-	private StringValue word = new StringValue("");
+	private String word = new String();
 
 	@Override
 	public void invoke(StreamRecord record) throws Exception {
-		word = (StringValue) record.getField(0, 0);
-		System.out.println("========" + word.getValue() + "=========");
+		word = record.getString(0, 0);
+		System.out.println("========" + word + "=========");
 
 	}
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/batch/BatchForwardSource.java

@@ -15,13 +15,13 @@
 
 package eu.stratosphere.streaming.examples.batch;
 
+import eu.stratosphere.api.java.tuple.Tuple1;
 import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.types.StringValue;
 
 public class BatchForwardSource extends UserSourceInvokable {
-	private final StringValue motto = new StringValue("Stratosphere Big Data looks tiny from here");
-	private final StreamRecord mottoRecord = new StreamRecord(motto);
+	private final StreamRecord mottoRecord = new StreamRecord(
+			new Tuple1<String>("Stratosphere Big Data looks tiny from here"));
 
 	@Override
 	public void invoke() throws Exception {