[streaming] Updated input checking for next record at invoke

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

@@ -17,6 +17,7 @@ package eu.stratosphere.streaming.api.streamcomponent;
 
 import java.io.IOException;
 import java.util.ConcurrentModificationException;
+import java.util.LinkedList;
 import java.util.List;
 
 import org.apache.commons.logging.Log;
@@ -26,7 +27,6 @@ import eu.stratosphere.configuration.Configuration;
 import eu.stratosphere.nephele.event.task.AbstractTaskEvent;
 import eu.stratosphere.nephele.event.task.EventListener;
 import eu.stratosphere.nephele.io.ChannelSelector;
-import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.nephele.io.RecordWriter;
 import eu.stratosphere.nephele.template.AbstractInvokable;
 import eu.stratosphere.streaming.api.invokable.DefaultSinkInvokable;
@@ -75,15 +75,15 @@ public final class StreamComponentHelper<T extends AbstractInvokable> {
 
 	}
 
-	public void setConfigInputs(T taskBase, Configuration taskConfiguration, List<RecordReader<StreamRecord>> inputs)
+	public void setConfigInputs(T taskBase, Configuration taskConfiguration, List<StreamRecordReader<StreamRecord>> inputs)
 			throws StreamComponentException {
 		int numberOfInputs = taskConfiguration.getInteger("numberOfInputs", 0);
 		for (int i = 0; i < numberOfInputs; i++) {
 
 			if (taskBase instanceof StreamTask) {
-				inputs.add(new RecordReader<StreamRecord>((StreamTask) taskBase, StreamRecord.class));
+				inputs.add(new StreamRecordReader<StreamRecord>((StreamTask) taskBase, StreamRecord.class));
 			} else if (taskBase instanceof StreamSink) {
-				inputs.add(new RecordReader<StreamRecord>((StreamSink) taskBase, StreamRecord.class));
+				inputs.add(new StreamRecordReader<StreamRecord>((StreamSink) taskBase, StreamRecord.class));
 			} else {
 				throw new StreamComponentException("Nonsupported object passed to setConfigInputs");
 			}
@@ -142,7 +142,7 @@ public final class StreamComponentHelper<T extends AbstractInvokable> {
 	}
 
 	// TODO find a better solution for this
-	public void threadSafePublish(AbstractTaskEvent event, RecordReader<StreamRecord> input)
+	public void threadSafePublish(AbstractTaskEvent event, StreamRecordReader<StreamRecord> input)
 			throws InterruptedException, IOException {
 
 		boolean concurrentModificationOccured = false;
@@ -177,12 +177,13 @@ public final class StreamComponentHelper<T extends AbstractInvokable> {
 		}
 	}
 
-	public void invokeRecords(RecordInvokable userFunction, List<RecordReader<StreamRecord>> inputs, String name) throws Exception
-			{
+	public void invokeRecords(RecordInvokable userFunction, List<StreamRecordReader<StreamRecord>> inputs, String name)
+			throws Exception {
+		List<StreamRecordReader<StreamRecord>> closedInputs = new LinkedList<StreamRecordReader<StreamRecord>>();
 		boolean hasInput = true;
 		while (hasInput) {
 			hasInput = false;
-			for (RecordReader<StreamRecord> input : inputs) {
+			for (StreamRecordReader<StreamRecord> input : inputs) {
 				if (input.hasNext()) {
 					hasInput = true;
 					StreamRecord record = input.next();
@@ -191,7 +192,11 @@ public final class StreamComponentHelper<T extends AbstractInvokable> {
 					threadSafePublish(new AckEvent(id), input);
 					log.debug("ACK: " + id + " -- " + name);
 				}
+				else if (input.isInputClosed()) {
+					closedInputs.add(input);
+				}
 			}
+			inputs.removeAll(closedInputs);
 		}
 	}
 

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

+/***********************************************************************************************************************
+ * Copyright (C) 2010-2013 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ **********************************************************************************************************************/
+
+package eu.stratosphere.streaming.api.streamcomponent;
+
+import java.io.IOException;
+
+import eu.stratosphere.core.io.IOReadableWritable;
+import eu.stratosphere.nephele.io.AbstractSingleGateRecordReader;
+import eu.stratosphere.nephele.io.InputChannelResult;
+import eu.stratosphere.nephele.io.MutableRecordDeserializerFactory;
+import eu.stratosphere.nephele.io.Reader;
+import eu.stratosphere.nephele.template.AbstractOutputTask;
+import eu.stratosphere.nephele.template.AbstractTask;
+
+/**
+ * A record writer connects an input gate to an application. It allows the application
+ * query for incoming records and read them from input gate.
+ * 
+ * @param <T> The type of the record that can be read from this record reader.
+ */
+public class StreamRecordReader<T extends IOReadableWritable> extends AbstractSingleGateRecordReader<T> implements Reader<T> {
+	
+	private final Class<T> recordType;
+	
+	/**
+	 * Stores the last read record.
+	 */
+	private T lookahead;
+
+	/**
+	 * Stores if more no more records will be received from the assigned input gate.
+	 */
+	private boolean noMoreRecordsWillFollow;
+
+	// --------------------------------------------------------------------------------------------
+	
+	/**
+	 * Constructs a new record reader and registers a new input gate with the application's environment.
+	 * 
+	 * @param taskBase
+	 *        The application that instantiated the record reader.
+	 * @param recordType
+	 *        The class of records that can be read from the record reader.
+	 */
+	public StreamRecordReader(AbstractTask taskBase, Class<T> recordType) {
+		super(taskBase, MutableRecordDeserializerFactory.<T>get(), 0);
+		this.recordType = recordType;
+	}
+
+	/**
+	 * Constructs a new record reader and registers a new input gate with the application's environment.
+	 * 
+	 * @param outputBase
+	 *        The application that instantiated the record reader.
+	 * @param recordType
+	 *        The class of records that can be read from the record reader.
+	 */
+	public StreamRecordReader(AbstractOutputTask outputBase, Class<T> recordType) {
+		super(outputBase, MutableRecordDeserializerFactory.<T>get(), 0);
+		this.recordType = recordType;
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	
+	/**
+	 * Checks if at least one more record can be read from the associated input gate. This method may block
+	 * until the associated input gate is able to read the record from one of its input channels.
+	 * 
+	 * @return <code>true</code>it at least one more record can be read from the associated input gate, otherwise
+	 *         <code>false</code>
+	 */
+	@Override
+	public boolean hasNext() throws IOException, InterruptedException{
+		if (this.lookahead != null) {
+			return true;
+		} else {
+			if (this.noMoreRecordsWillFollow) {
+				return false;
+			}
+			
+			T record = instantiateRecordType();
+			
+			while (true) {
+				InputChannelResult result = this.inputGate.readRecord(record);
+				switch (result) {
+					case INTERMEDIATE_RECORD_FROM_BUFFER:
+					case LAST_RECORD_FROM_BUFFER:
+						this.lookahead = record;
+						return true;
+						
+					case END_OF_SUPERSTEP:
+						if (incrementEndOfSuperstepEventAndCheck())
+							return false;
+						else
+							return false;
+					case TASK_EVENT:
+						handleEvent(this.inputGate.getCurrentEvent());
+						return false;	
+					case END_OF_STREAM:
+						this.noMoreRecordsWillFollow = true;
+						return false;
+					case NONE: // internal event or an incomplete record that needs further chunks
+						// the current unit is exhausted
+						break;
+					default:
+						return false;
+				}
+			}
+		}
+	}
+	
+	/**
+	 * Reads the current record from the associated input gate.
+	 * 
+	 * @return the current record from the associated input gate.
+	 * @throws IOException
+	 *         thrown if any error occurs while reading the record from the input gate
+	 */
+	@Override
+	public T next() throws IOException, InterruptedException {
+		if (hasNext()) {
+			T tmp = this.lookahead;
+			this.lookahead = null;
+			return tmp;
+		} else {
+			return null;
+		}
+	}
+	
+	@Override
+	public boolean isInputClosed() {
+		return this.noMoreRecordsWillFollow;
+	}
+	
+	private T instantiateRecordType() {
+		try {
+			return this.recordType.newInstance();
+		} catch (InstantiationException e) {
+			throw new RuntimeException("Cannot instantiate class '" + this.recordType.getName() + "'.", e);
+		} catch (IllegalAccessException e) {
+			throw new RuntimeException("Cannot instantiate class '" + this.recordType.getName() + "'.", e);
+		}
+	}
+}

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

@@ -22,7 +22,6 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import eu.stratosphere.configuration.Configuration;
-import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.nephele.template.AbstractOutputTask;
 import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
@@ -31,14 +30,14 @@ public class StreamSink extends AbstractOutputTask {
 
 	private static final Log log = LogFactory.getLog(StreamSink.class);
 
-	private List<RecordReader<StreamRecord>> inputs;
+	private List<StreamRecordReader<StreamRecord>> inputs;
 	private UserSinkInvokable userFunction;
 	private StreamComponentHelper<StreamSink> streamSinkHelper;
 	private String name;
 
 	public StreamSink() {
 		// TODO: Make configuration file visible and call setClassInputs() here
-		inputs = new LinkedList<RecordReader<StreamRecord>>();
+		inputs = new LinkedList<StreamRecordReader<StreamRecord>>();
 		userFunction = null;
 		streamSinkHelper = new StreamComponentHelper<StreamSink>();
 	}

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

@@ -23,7 +23,6 @@ import org.apache.commons.logging.LogFactory;
 
 import eu.stratosphere.configuration.Configuration;
 import eu.stratosphere.nephele.io.ChannelSelector;
-import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.nephele.io.RecordWriter;
 import eu.stratosphere.nephele.template.AbstractTask;
 import eu.stratosphere.streaming.api.invokable.UserTaskInvokable;
@@ -34,7 +33,7 @@ public class StreamTask extends AbstractTask {
 
 	private static final Log log = LogFactory.getLog(StreamTask.class);
 
-	private List<RecordReader<StreamRecord>> inputs;
+	private List<StreamRecordReader<StreamRecord>> inputs;
 	private List<RecordWriter<StreamRecord>> outputs;
 	private List<ChannelSelector<StreamRecord>> partitioners;
 	private UserTaskInvokable userFunction;
@@ -48,7 +47,7 @@ public class StreamTask extends AbstractTask {
 
 	public StreamTask() {
 		// TODO: Make configuration file visible and call setClassInputs() here
-		inputs = new LinkedList<RecordReader<StreamRecord>>();
+		inputs = new LinkedList<StreamRecordReader<StreamRecord>>();
 		outputs = new LinkedList<RecordWriter<StreamRecord>>();
 		partitioners = new LinkedList<ChannelSelector<StreamRecord>>();
 		userFunction = null;

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/wordcount/WordCountLocal.java

@@ -31,8 +31,8 @@ public class WordCountLocal {
 	private static JobGraph getJobGraph() throws Exception {
 		JobGraphBuilder graphBuilder = new JobGraphBuilder("testGraph");
 		graphBuilder.setSource("WordCountSource", WordCountDummySource.class);
-		graphBuilder.setTask("WordCountSplitter", WordCountSplitter.class, 2, 1);
-		graphBuilder.setTask("WordCountCounter", WordCountCounter.class, 2, 1);
+		graphBuilder.setTask("WordCountSplitter", WordCountSplitter.class, 2, 2);
+		graphBuilder.setTask("WordCountCounter", WordCountCounter.class, 2, 2);
 		graphBuilder.setSink("WordCountSink", WordCountSink.class);
 
 		graphBuilder.shuffleConnect("WordCountSource", "WordCountSplitter");