Update javadoc to document what components do and what their intended role is.

Author: jbrisbin

ripc-core/src/main/java/io/ripc/core/DemandCalculator.java

@@ -1,10 +1,19 @@
 package io.ripc.core;
 
 /**
- * Created by jbrisbin on 3/26/15.
+ * A {@code DemandCalculator} implementation is responsible for calculating what the demand value should be to send to
+ * {@link org.reactivestreams.Subscription#request(long)}.
  */
 public interface DemandCalculator {
 
+	/**
+	 * Calculate demand based on current pending backlog. A value {@code < 1} means "don't make any new requests" since
+	 * values less than {@code 1} are illegal according to the Reactive Streams spec. A value of {@code >= 1} means "use
+	 * this value as the demand".
+	 *
+	 * @param pending outstanding backlog of previous demand accumulations
+	 * @return &lt 1 to indicate no requests should be performed, &gt= 1 to indicate positive demand
+	 */
 	long calculateDemand(long pending);
 
 }

ripc-core/src/main/java/io/ripc/core/EventListener.java

@@ -0,0 +1,8 @@
+package io.ripc.core;
+
+/**
+ * An {@code EventListener} is a generic component that has no default behavior. Components are meant to use specific
+ * subclasses of this interface to intercept events and perform appropriate behavior.
+ */
+public interface EventListener {
+}

ripc-core/src/main/java/io/ripc/core/Handler.java

@@ -0,0 +1,12 @@
+package io.ripc.core;
+
+/**
+ * A {@code Handler} takes an arbitrary object as input and "handles" it.
+ */
+public interface Handler<T> {
+	/**
+	 * Handle the given object (do something useful with it).
+	 * @param obj
+	 */
+	void handle(T obj);
+}

ripc-core/src/main/java/io/ripc/core/SingletonPublisher.java

@@ -7,7 +7,9 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
- * Created by jbrisbin on 3/26/15.
+ * A {@code SingletonPublisher} provides a single value only once and then calls {@code onComplete}. If the value is
+ * {@code null}, then the {@link org.reactivestreams.Subscriber#onNext(Object)} is not called but {@link
+ * org.reactivestreams.Subscriber#onComplete()} is.
  */
 public class SingletonPublisher<T> implements Publisher<T>, DemandCalculator {
 

ripc-core/src/main/java/io/ripc/core/Specification.java

@@ -3,7 +3,7 @@
 import org.reactivestreams.Subscriber;
 
 /**
- * Created by jbrisbin on 3/10/15.
+ * Helper class to encapsulate various checks required by the Reactive Streams Specification.
  */
 public abstract class Specification {
 

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/connection/TcpConnection.java

@@ -1,16 +1,37 @@
 package io.ripc.protocol.tcp.connection;
 
+import io.ripc.core.EventListener;
 import org.reactivestreams.Publisher;
 
 /**
  * A {@code Connection} provides a reader for inbound data and a writer for outbound.
  */
 public interface TcpConnection {
 
+	/**
+	 * Receive the {@link org.reactivestreams.Publisher} which will be producing inbound data. This is intended to be
+	 * composed into a more sophisticated processing pipeline by a higher layer of composition helpers.
+	 *
+	 * @return {@code this}
+	 */
 	Publisher<?> reader();
 
+	/**
+	 * Set the write {@link org.reactivestreams.Publisher} that will be used to send outbound data to the peer. Can only
+	 * be invoked once.
+	 *
+	 * @param writer the {@link org.reactivestreams.Publisher} which will produce outbound data
+	 * @return {@code this}
+	 */
 	TcpConnection writer(Publisher<?> writer);
 
-	TcpConnection addListener(TcpConnectionEventListener listener);
+	/**
+	 * Add a type of {@link io.ripc.core.EventListener} to the connection which will respond to the events for which that
+	 * specific listener type is responsible. What's available varies by protocol and server implementation.
+	 *
+	 * @param listener
+	 * @return {@code this}
+	 */
+	TcpConnection addListener(EventListener listener);
 
 }

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/connection/TcpConnectionEventListener.java

@@ -0,0 +1,23 @@
+package io.ripc.protocol.tcp.connection.listener;
+
+import io.ripc.core.EventListener;
+import io.ripc.protocol.tcp.connection.TcpConnection;
+
+/**
+ * An {@code EventListener} implementation that is used to receive notifications that an IO channel has completed the
+ * read of all available data.
+ */
+public interface ReadCompleteListener extends EventListener {
+
+	/**
+	 * Invoked when all data has been read from the given {@link io.ripc.protocol.tcp.connection.TcpConnection}.
+	 * Returning
+	 * {@code true} here means "I'm finished with the connection you may close it now" and returning {@code false} means
+	 * "leave the connection open".
+	 *
+	 * @param connection the connection on which reads are complete
+	 * @return {@code true} to close the connection, {@code false} to leave it open
+	 */
+	boolean readComplete(TcpConnection connection);
+
+}

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/connection/TcpConnectionHandler.java

@@ -1,13 +1,23 @@
 package io.ripc.protocol.tcp.connection.listener;
 
+import io.ripc.core.EventListener;
 import io.ripc.protocol.tcp.connection.TcpConnection;
-import io.ripc.protocol.tcp.connection.TcpConnectionEventListener;
 
 /**
- * Created by jbrisbin on 3/26/15.
+ * An {@code EventListener} implementation that is used to receive notifications that writes have been successfully
+ * completed on the underlying IO channel.
  */
-public interface WriteCompleteListener extends TcpConnectionEventListener {
+public interface WriteCompleteListener extends EventListener {
 
+	/**
+	 * Invoked every time a write to the IO channel is complete. Returning {@code false} here means "don't do any
+	 * flushing" and returning {@code true} means "flush the underlying IO channel".
+	 *
+	 * @param connection the connection on which the write was made
+	 * @param count      the number of items that have been written since the last flush
+	 * @param msg        the last value written
+	 * @return {@code true} to perform a flush of the channel, {@code false} otherwise
+	 */
 	boolean writeComplete(TcpConnection connection, long count, Object msg);
 
 }

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/connection/listener/ReadCompleteListener.java

@@ -1,6 +1,5 @@
 package io.ripc.transport.netty4.tcp;
 
-import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -11,7 +10,8 @@
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 /**
- * Created by jbrisbin on 3/10/15.
+ * A {@code ChannelInboundHandlerAdapter} that is responsible for propagating data from the IO channel to the read
+ * {@link org.reactivestreams.Subscriber}.
  */
 public class ChannelInboundHandlerSubscription extends ChannelInboundHandlerAdapter implements Subscription {
 
@@ -60,11 +60,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
 			return;
 		}
 
-		ByteBuf buf = (ByteBuf) msg;
 		try {
-			subscriber.onNext(buf);
+			subscriber.onNext(msg);
 			PEND_UPD.decrementAndGet(this);
-			//channel.read();
 		} catch (Throwable t) {
 			subscriber.onError(t);
 		}

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/connection/listener/WriteCompleteListener.java

@@ -6,28 +6,39 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.ripc.protocol.tcp.connection.TcpConnection;
+import io.ripc.protocol.tcp.connection.listener.ReadCompleteListener;
 import io.ripc.protocol.tcp.connection.listener.WriteCompleteListener;
 
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 /**
- * Created by jbrisbin on 3/26/15.
+ * Listens for events on the Netty IO channel and invokes the appropriate {@link io.ripc.core.EventListener} for that
+ * type of event.
  */
-public class TcpConnectionEventListenerChannelHandler extends ChannelDuplexHandler {
+public class EventListenerChannelHandler extends ChannelDuplexHandler {
 
-	private static final AtomicLongFieldUpdater<TcpConnectionEventListenerChannelHandler> WRITTEN_UPD
-			= AtomicLongFieldUpdater.newUpdater(TcpConnectionEventListenerChannelHandler.class, "writtenSinceLastFlush");
+	private static final AtomicLongFieldUpdater<EventListenerChannelHandler> WRITTEN_UPD
+			= AtomicLongFieldUpdater.newUpdater(EventListenerChannelHandler.class, "writtenSinceLastFlush");
 
 	private final TcpConnection connection;
 
 	private volatile long writtenSinceLastFlush = 0L;
 
+	private ReadCompleteListener  readCompleteListener;
 	private WriteCompleteListener writeCompleteListener;
 
-	public TcpConnectionEventListenerChannelHandler(TcpConnection connection) {
+	public EventListenerChannelHandler(TcpConnection connection) {
 		this.connection = connection;
 	}
 
+	public ReadCompleteListener getReadCompleteListener() {
+		return readCompleteListener;
+	}
+
+	public void setReadCompleteListener(ReadCompleteListener readCompleteListener) {
+		this.readCompleteListener = readCompleteListener;
+	}
+
 	public WriteCompleteListener getWriteCompleteListener() {
 		return writeCompleteListener;
 	}
@@ -58,4 +69,12 @@ public void operationComplete(ChannelFuture future) throws Exception {
 		WRITTEN_UPD.incrementAndGet(this);
 	}
 
+	@Override
+	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+		super.channelReadComplete(ctx);
+		if (null != readCompleteListener && readCompleteListener.readComplete(connection)) {
+			ctx.close();
+		}
+	}
+
 }

ripc-transport-netty4/src/main/java/io/ripc/transport/netty4/tcp/ChannelInboundHandlerSubscription.java

@@ -8,7 +8,8 @@
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.logging.LoggingHandler;
-import io.ripc.protocol.tcp.connection.TcpConnectionHandler;
+import io.ripc.core.Handler;
+import io.ripc.protocol.tcp.connection.TcpConnection;
 import io.ripc.transport.netty4.NamedDaemonThreadFactory;
 
 /**
@@ -22,7 +23,7 @@ public NettyTcpServer(ServerBootstrap bootstrap) {
 		this.bootstrap = bootstrap;
 	}
 
-	public static NettyTcpServer listen(int port, TcpConnectionHandler handler) {
+	public static NettyTcpServer listen(int port, Handler<TcpConnection> handler) {
 		ServerBootstrap b = new ServerBootstrap();
 
 		int threads = Runtime.getRuntime().availableProcessors();

ripc-transport-netty4/src/main/java/io/ripc/transport/netty4/tcp/TcpConnectionEventListenerChannelHandler.java renamed to ripc-transport-netty4/src/main/java/io/ripc/transport/netty4/tcp/EventListenerChannelHandler.java

@@ -2,19 +2,20 @@
 
 import io.netty.channel.Channel;
 import io.ripc.core.DemandCalculator;
+import io.ripc.core.EventListener;
 import io.ripc.protocol.tcp.connection.TcpConnection;
-import io.ripc.protocol.tcp.connection.TcpConnectionEventListener;
+import io.ripc.protocol.tcp.connection.listener.ReadCompleteListener;
 import io.ripc.protocol.tcp.connection.listener.WriteCompleteListener;
 import io.ripc.transport.netty4.tcp.ChannelInboundHandlerSubscription;
-import io.ripc.transport.netty4.tcp.TcpConnectionEventListenerChannelHandler;
+import io.ripc.transport.netty4.tcp.EventListenerChannelHandler;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 /**
- * Represents a Netty Channel.
+ * Represents a Reactive IPC {@code TcpConnection}.
  */
 public class NettyTcpServerConnection implements TcpConnection {
 
@@ -24,24 +25,34 @@ public class NettyTcpServerConnection implements TcpConnection {
 	private final Channel       channel;
 	private final ReadPublisher readPublisher;
 
+	private WriteSubscriber writeSubscriber;
+
 	public NettyTcpServerConnection(Channel channel) {
 		this.channel = channel;
 		this.readPublisher = new ReadPublisher(channel);
 	}
 
 	@Override
-	public TcpConnection addListener(TcpConnectionEventListener listener) {
-		TcpConnectionEventListenerChannelHandler handler =
-				channel.pipeline().get(TcpConnectionEventListenerChannelHandler.class);
+	public TcpConnection addListener(EventListener listener) {
+		EventListenerChannelHandler handler =
+				channel.pipeline().get(EventListenerChannelHandler.class);
 		if (null == handler) {
-			handler = new TcpConnectionEventListenerChannelHandler(this);
+			handler = new EventListenerChannelHandler(this);
 			channel.pipeline().addLast(handler);
 		}
 
-		if (WriteCompleteListener.class.isAssignableFrom(listener.getClass())) {
+		Class<?> listenerType = listener.getClass();
+
+		// Assign WriteCompleteListener that will be notified of successful writes.
+		if (WriteCompleteListener.class.isAssignableFrom(listenerType)) {
 			handler.setWriteCompleteListener((WriteCompleteListener) listener);
 		}
 
+		// Assign a ReadCompleteListener that will be notified when all data has been read.
+		if (ReadCompleteListener.class.isAssignableFrom(listenerType)) {
+			handler.setReadCompleteListener((ReadCompleteListener) listener);
+		}
+
 		return this;
 	}
 
@@ -52,10 +63,15 @@ public Publisher<?> reader() {
 
 	@Override
 	public TcpConnection writer(Publisher<?> writer) {
+		if (null != writeSubscriber) {
+			throw new IllegalStateException("A writer has already been set on this connection");
+		}
+
 		DemandCalculator demandCalculator = DemandCalculator.class.isAssignableFrom(writer.getClass())
 		                                    ? (DemandCalculator) writer
 		                                    : null;
-		writer.subscribe(new WriteSubscriber(demandCalculator));
+		this.writeSubscriber = new WriteSubscriber(demandCalculator);
+		writer.subscribe(writeSubscriber);
 		return this;
 	}
 
@@ -99,7 +115,7 @@ public void run() {
 						toRequest = 1L;
 					}
 
-					if (toRequest == Long.MAX_VALUE) {
+					if (toRequest == Long.MAX_VALUE && pending != Long.MAX_VALUE) {
 						PENDING_UPD.set(WriteSubscriber.this, Long.MAX_VALUE);
 						subscription.request(Long.MAX_VALUE);
 					} else if (toRequest > 0) {
@@ -118,13 +134,17 @@ public void onSubscribe(Subscription subscription) {
 			}
 
 			this.subscription = subscription;
+			// Request for any writes right away.
 			subscriptionRequest.run();
 		}
 
 		@Override
 		public void onNext(Object msg) {
+			// Write to the channel for every onNext signal.
 			channel.write(msg);
+			// Decrement pending counter to show we've handled this write.
 			PENDING_UPD.decrementAndGet(this);
+			// We'll get a StackOverflowError if we don't schedule this request.
 			channel.eventLoop().execute(subscriptionRequest);
 		}
 
@@ -135,7 +155,9 @@ public void onError(Throwable t) {
 
 		@Override
 		public void onComplete() {
+			// Write completes always result in a flush.
 			channel.flush();
+			// This is a terminal state, so close the underlying channel.
 			channel.close();
 		}
 	}