[streaming] BatchForward Refactor

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/batch/MyBatchStreamSink.java → flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/batch/BachForwardSink.java

@@ -19,7 +19,7 @@ import eu.stratosphere.streaming.api.StreamRecord;
 import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
 import eu.stratosphere.types.StringValue;
 
-public class MyBatchStreamSink implements UserSinkInvokable {
+public class BachForwardSink implements UserSinkInvokable {
 
 	private StringValue word = new StringValue("");
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/batch/MyBatchStream.java → flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/batch/BatchForward.java

@@ -19,13 +19,13 @@ import eu.stratosphere.nephele.jobgraph.JobGraph;
 import eu.stratosphere.streaming.api.JobGraphBuilder;
 import eu.stratosphere.test.util.TestBase2;
 
-public class MyBatchStream extends TestBase2{
+public class BatchForward extends TestBase2{
 
 	@Override
 	public JobGraph getJobGraph() {
 		JobGraphBuilder graphBuilder = new JobGraphBuilder("testGraph");
-		graphBuilder.setSource("StreamSource", MyBatchStreamSource.class);
-		graphBuilder.setSink("StreamSink", MyBatchStreamSink.class);
+		graphBuilder.setSource("StreamSource", BatchForwardSource.class);
+		graphBuilder.setSink("StreamSink", BachForwardSink.class);
 
 		graphBuilder.broadcastConnect("StreamSource", "StreamSink");
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/batch/MyBatchStreamSource.java → flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/batch/BatchForwardSource.java

@@ -15,22 +15,17 @@
 
 package eu.stratosphere.streaming.test.batch;
 
-import eu.stratosphere.streaming.api.AtomRecord;
 import eu.stratosphere.streaming.api.StreamRecord;
 import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
 import eu.stratosphere.types.StringValue;
 
-public class MyBatchStreamSource  extends UserSourceInvokable {
-//	private final String motto = "Stratosphere Big Data looks tiny from here";
-	private final String motto = "Gyuszi Gabor Big Marci Gyuszi";
-	private final AtomRecord record=new AtomRecord();
-	private final StreamRecord mottoRecord=new StreamRecord();
-	
+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);
+
 	@Override
 	public void invoke() throws Exception {
-		record.setField(0, new StringValue(motto));
-		mottoRecord.addRecord(record);
-		for (int i = 0; i < 100; i++) {
+		for (int i = 0; i < 10; i++) {
 			emit(mottoRecord);
 		}
 	}