[streaming] Started implementing FlatStreamTask

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/FlatStreamRecord.java

+package eu.stratosphere.streaming.api;
+
+import eu.stratosphere.types.LongValue;
+import eu.stratosphere.types.Record;
+import eu.stratosphere.types.Value;
+
+public class FlatStreamRecord {
+
+	private long id;
+	private Record record;
+	private int numberOfFields;
+	
+	public FlatStreamRecord() {
+		id = 94;
+	}
+	
+	public FlatStreamRecord(int numberOfFields) {
+		this();
+		
+		this.numberOfFields = numberOfFields;
+		record = new Record(numberOfFields + 1);
+		LongValue idValue = new LongValue(id);
+		record.addField(idValue);
+	}
+	
+	public FlatStreamRecord(Record record) {
+		this();
+		
+		this.numberOfFields = record.getNumFields();
+		this.record = record;
+		LongValue idValue = new LongValue(id);
+		record.addField(idValue);
+	}
+	
+	public int getNumFields() {
+		return this.numberOfFields;
+	}
+	
+	public boolean getFieldInto(int fieldNum, Value target) {
+		return record.getFieldInto(fieldNum, target);
+	}
+	
+	public <T extends Value> T getField(final int fieldNum, final Class<T> type) {
+		return record.getField(fieldNum, type);
+	}
+
+	//TODO: set to private or package private, this cannot be seen from outside!
+	public Record getRecord() {
+		return this.record;
+	}
+	
+}

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamSink.java

@@ -68,7 +68,7 @@ public class StreamSink extends AbstractOutputTask {
       for (RecordReader<Record> input : inputs) {
         if (input.hasNext()) {
           hasInput = true;
-					userFunction.invoke(input.next(), input);
+          userFunction.invoke(input.next());
         }
       }
     }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamSource.java

@@ -56,8 +56,7 @@ public class StreamSource extends AbstractInputTask<RandIS> {
 
   private void setConfigInputs() {
 
-		numberOfOutputs = getTaskConfiguration().getInteger("numberOfOutputs",
-				0);
+    numberOfOutputs = getTaskConfiguration().getInteger("numberOfOutputs", 0);
     Class<? extends ChannelSelector<Record>> partitioner;
 
     for (int i = 1; i <= numberOfOutputs; i++) {
@@ -73,12 +72,18 @@ public class StreamSource extends AbstractInputTask<RandIS> {
       }
     }
 
+    for (ChannelSelector<Record> outputPartitioner : partitioners) {
+      outputs.add(new RecordWriter<Record>(this, Record.class,
+          outputPartitioner));
+    }
+
     Class<? extends UserSourceInvokable> userFunctionClass;
     userFunctionClass = getTaskConfiguration().getClass("userfunction",
         DefaultSourceInvokable.class, UserSourceInvokable.class);
 
     try {
       userFunction = userFunctionClass.newInstance();
+      userFunction.declareOutputs(outputs);
     } catch (Exception e) {
 
     }
@@ -88,18 +93,14 @@ public class StreamSource extends AbstractInputTask<RandIS> {
   @Override
   public void registerInputOutput() {
     setConfigInputs();
-		for (ChannelSelector<Record> partitioner : partitioners) {
-			outputs.add(new RecordWriter<Record>(this, Record.class,
-					partitioner));
-		}
+
   }
 
   @Override
   public void invoke() throws Exception {
 
-		for (RecordWriter<Record> output : outputs) {
-			userFunction.invoke(output);
-		}
+    userFunction.invoke();
+
   }
 
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/StreamTask.java

@@ -26,6 +26,7 @@ import eu.stratosphere.streaming.api.invokable.DefaultTaskInvokable;
 import eu.stratosphere.streaming.api.invokable.UserTaskInvokable;
 import eu.stratosphere.streaming.partitioner.DefaultPartitioner;
 import eu.stratosphere.types.Record;
+import eu.stratosphere.types.StringValue;
 
 public class StreamTask extends AbstractTask {
 
@@ -67,11 +68,17 @@ public class StreamTask extends AbstractTask {
       }
     }
 
+    for (ChannelSelector<Record> outputPartitioner : partitioners) {
+      outputs.add(new RecordWriter<Record>(this, Record.class,
+          outputPartitioner));
+    }
+
     Class<? extends UserTaskInvokable> userFunctionClass;
     userFunctionClass = getTaskConfiguration().getClass("userfunction",
         DefaultTaskInvokable.class, UserTaskInvokable.class);
     try {
       userFunction = userFunctionClass.newInstance();
+      userFunction.declareOutputs(outputs);
     } catch (Exception e) {
 
     }
@@ -80,12 +87,10 @@ public class StreamTask extends AbstractTask {
   @Override
   public void registerInputOutput() {
     setConfigInputs();
-    for (ChannelSelector<Record> partitioner : partitioners) {
-      outputs.add(new RecordWriter<Record>(this, Record.class, partitioner));
-    }
+    Record r = new Record();
+    r.addField(new StringValue(""));
   }
 
-  // TODO: Performance with multiple outputs
   @Override
   public void invoke() throws Exception {
     boolean hasInput = true;
@@ -94,9 +99,7 @@ public class StreamTask extends AbstractTask {
       for (RecordReader<Record> input : inputs) {
         if (input.hasNext()) {
           hasInput = true;
-          for (RecordWriter<Record> output : outputs) {
-            userFunction.invoke(input.next(), output);
-          }
+          userFunction.invoke(new FlatStreamRecord(input.next()));
         }
       }
     }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/DefaultSinkInvokable.java

@@ -15,15 +15,13 @@
 
 package eu.stratosphere.streaming.api.invokable;
 
-import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.types.Record;
 import eu.stratosphere.types.StringValue;
 
 public class DefaultSinkInvokable implements UserSinkInvokable {
 
   @Override
-	public void invoke(Record record, RecordReader<Record> input)
-			throws Exception {
+  public void invoke(Record record) throws Exception {
     StringValue value = new StringValue("");
     record.getFieldInto(0, value);
     System.out.println(value.getValue());

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/DefaultSourceInvokable.java

@@ -15,19 +15,19 @@
 
 package eu.stratosphere.streaming.api.invokable;
 
-import eu.stratosphere.nephele.io.RecordWriter;
+import eu.stratosphere.streaming.api.FlatStreamRecord;
 import eu.stratosphere.types.Record;
 import eu.stratosphere.types.StringValue;
 
-public class DefaultSourceInvokable implements UserSourceInvokable {
+public class DefaultSourceInvokable extends UserSourceInvokable {
 
   private String motto = "Stratosphere -- Big Data looks tiny from here";
   private String[] mottoArray = motto.split(" ");
 
   @Override
-	public void invoke(RecordWriter<Record> output) throws Exception {
+  public void invoke() throws Exception {
     for (CharSequence word : mottoArray) {
-			output.emit(new Record(new StringValue(word)));
+      emit(new FlatStreamRecord(new Record(new StringValue(word))));
     }
   }
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/DefaultTaskInvokable.java

@@ -15,15 +15,12 @@
 
 package eu.stratosphere.streaming.api.invokable;
 
-import eu.stratosphere.nephele.io.RecordWriter;
-import eu.stratosphere.types.Record;
+import eu.stratosphere.streaming.api.FlatStreamRecord;
 
-public class DefaultTaskInvokable implements UserTaskInvokable {
+public class DefaultTaskInvokable extends UserTaskInvokable {
 
 	@Override
-	public void invoke(Record record, RecordWriter<Record> output)
-			throws Exception {
-		output.emit(record);
+	public void invoke(FlatStreamRecord streamRecord) throws Exception {
+		emit(streamRecord);
 	}
-
 }
\ No newline at end of file

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/StreamInvokable.java

+package eu.stratosphere.streaming.api.invokable;
+
+import java.util.List;
+
+import eu.stratosphere.nephele.io.RecordWriter;
+import eu.stratosphere.streaming.api.FlatStreamRecord;
+import eu.stratosphere.types.Record;
+
+public abstract class StreamInvokable {
+
+  private List<RecordWriter<Record>> outputs;
+
+  public final void declareOutputs(List<RecordWriter<Record>> outputs) {
+    this.outputs = outputs;
+  }
+
+  public final void emit(FlatStreamRecord streamRecord) {
+    for (RecordWriter<Record> output : outputs) {
+      try {
+        output.emit(streamRecord.getRecord());
+      } catch (Exception e) {
+        System.out.println("Emit error");
+      }
+    }
+  }
+}

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/UserSinkInvokable.java

@@ -15,11 +15,9 @@
 
 package eu.stratosphere.streaming.api.invokable;
 
-import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.types.Record;
 
 public interface UserSinkInvokable {
 
-  public void invoke(Record record,
-      RecordReader<Record> input) throws Exception;
+  public void invoke(Record record) throws Exception;
 }
\ No newline at end of file

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/UserSourceInvokable.java

@@ -15,9 +15,9 @@
 
 package eu.stratosphere.streaming.api.invokable;
 
-import eu.stratosphere.nephele.io.RecordWriter;
-import eu.stratosphere.types.Record;
+public abstract class UserSourceInvokable extends StreamInvokable {
 
-public interface UserSourceInvokable {
-  public void invoke(RecordWriter<Record> output) throws Exception ;
+  public void invoke() throws Exception {
+
+  }
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/UserTaskInvokable.java

@@ -15,11 +15,11 @@
 
 package eu.stratosphere.streaming.api.invokable;
 
-import eu.stratosphere.nephele.io.RecordWriter;
-import eu.stratosphere.types.Record;
+import eu.stratosphere.streaming.api.FlatStreamRecord;
 
-public interface UserTaskInvokable {
+public abstract class UserTaskInvokable extends StreamInvokable {
 
-  public void invoke(Record record,
-      RecordWriter<Record> output) throws Exception;
+  public void invoke(FlatStreamRecord record) throws Exception {
+
+  }
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/QuerySourceInvokable.java

@@ -15,16 +15,16 @@
 
 package eu.stratosphere.streaming.test;
 
-import eu.stratosphere.nephele.io.RecordWriter;
+import eu.stratosphere.streaming.api.FlatStreamRecord;
 import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
 import eu.stratosphere.types.IntValue;
 import eu.stratosphere.types.LongValue;
 import eu.stratosphere.types.Record;
 
-public class QuerySourceInvokable implements UserSourceInvokable {
+public class QuerySourceInvokable extends UserSourceInvokable {
 
   @Override
-  public void invoke(RecordWriter<Record> output) throws Exception {
+  public void invoke() throws Exception {
     for (int i = 0; i < 5; i++) {
       Record record1 = new Record(3);
       record1.setField(0, new IntValue(5));
@@ -34,10 +34,10 @@ public class QuerySourceInvokable implements UserSourceInvokable {
       Record record2 = new Record(3);
       record2.setField(0, new IntValue(4));
       record2.setField(1, new LongValue(510));
-      record1.setField(2, new LongValue(100));
+      record2.setField(2, new LongValue(100));
 
-      output.emit(record1);
-      output.emit(record2);
+      emit(new FlatStreamRecord(record1));
+      emit(new FlatStreamRecord(record2));
     }
   }
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/TestSinkInvokable.java

@@ -15,7 +15,6 @@
 
 package eu.stratosphere.streaming.test;
 
-import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
 import eu.stratosphere.types.Record;
 import eu.stratosphere.types.StringValue;
@@ -23,8 +22,7 @@ import eu.stratosphere.types.StringValue;
 public class TestSinkInvokable implements UserSinkInvokable {
 
   @Override
-	public void invoke(Record record, RecordReader<Record> input)
-			throws Exception {
+  public void invoke(Record record) throws Exception {
 
     StringValue value = new StringValue("");
     record.getFieldInto(0, value);

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/TestSourceInvokable.java

@@ -15,17 +15,16 @@
 
 package eu.stratosphere.streaming.test;
 
-
-import eu.stratosphere.nephele.io.RecordWriter;
+import eu.stratosphere.streaming.api.FlatStreamRecord;
 import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
 import eu.stratosphere.types.IntValue;
 import eu.stratosphere.types.LongValue;
 import eu.stratosphere.types.Record;
 
-public class TestSourceInvokable implements UserSourceInvokable {
+public class TestSourceInvokable extends UserSourceInvokable {
 
   @Override
-  public void invoke(RecordWriter<Record> output) throws Exception {
+  public void invoke() throws Exception {
     for (int i = 0; i < 10; i++) {
       Record record1 = new Record(2);
       record1.setField(0, new IntValue(5));
@@ -33,8 +32,8 @@ public class TestSourceInvokable implements UserSourceInvokable {
       Record record2 = new Record(2);
       record2.setField(0, new IntValue(4));
       record2.setField(1, new LongValue(500));
-    	output.emit(record1);
-    	output.emit(record2);
+      emit(new FlatStreamRecord(record1));
+      emit(new FlatStreamRecord(record2));
     }
   }
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/TestTaskInvokable.java

@@ -15,7 +15,7 @@
 
 package eu.stratosphere.streaming.test;
 
-import eu.stratosphere.nephele.io.RecordWriter;
+import eu.stratosphere.streaming.api.FlatStreamRecord;
 import eu.stratosphere.streaming.api.invokable.UserTaskInvokable;
 import eu.stratosphere.streaming.test.cellinfo.WorkerEngineExact;
 import eu.stratosphere.types.IntValue;
@@ -23,13 +23,12 @@ import eu.stratosphere.types.LongValue;
 import eu.stratosphere.types.Record;
 import eu.stratosphere.types.StringValue;
 
-public class TestTaskInvokable implements UserTaskInvokable {
+public class TestTaskInvokable extends UserTaskInvokable {
 
   private WorkerEngineExact engine = new WorkerEngineExact(10, 1000, 0);
 
   @Override
-	public void invoke(Record record, RecordWriter<Record> output)
-			throws Exception {
+  public void invoke(FlatStreamRecord record) throws Exception {
     IntValue value1 = new IntValue(0);
     record.getFieldInto(0, value1);
     LongValue value2 = new LongValue(0);
@@ -38,15 +37,14 @@ public class TestTaskInvokable implements UserTaskInvokable {
     // INFO
     if (record.getNumFields() == 2) {
       engine.put(value1.getValue(), value2.getValue());
-			output.emit(new Record(new StringValue(value1 + " " + value2)));
+      emit(new FlatStreamRecord(new Record(new StringValue(value1 + " " + value2))));
     }
     // QUERY
     else if (record.getNumFields() == 3) {
       LongValue value3 = new LongValue(0);
       record.getFieldInto(2, value3);
-
-			output.emit(new Record(new StringValue(String.valueOf(engine.get(
-					value2.getValue(), value3.getValue(), value1.getValue())))));
+      emit(new FlatStreamRecord(new Record(new StringValue(String.valueOf(engine.get(
+          value2.getValue(), value3.getValue(), value1.getValue()))))));
     }
   }
 }