Change event handling to be "listener" based by recognizing specific listener types that are subclasses of a marker interface.

Author: jbrisbin

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/AbstractTcpConnectionEventHandler.java

@@ -1,4 +1,4 @@
-package io.ripc.protocol.tcp;
+package io.ripc.protocol.tcp.connection;
 
 import org.reactivestreams.Publisher;
 
@@ -7,10 +7,10 @@
  */
 public interface TcpConnection {
 
-	TcpConnection eventHandler(TcpConnectionEventHandler eventHandler);
-
 	Publisher<?> reader();
 
-	TcpConnection writer(Publisher<?> sink);
+	TcpConnection writer(Publisher<?> writer);
+
+	TcpConnection addListener(TcpConnectionEventListener listener);
 
 }

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/CompleteEventHandler.java

@@ -0,0 +1,7 @@
+package io.ripc.protocol.tcp.connection;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public interface TcpConnectionEventListener {
+}

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/TcpConnectionEventHandler.java

@@ -1,11 +1,10 @@
-package io.ripc.protocol.tcp;
+package io.ripc.protocol.tcp.connection;
 
 /**
  * A {@code ConnectionHandler} is responsible for composing a Reactive Streams pipeline(s) when a new connection is
  * received by the server. Implementations will compose an appropriate pipeline based on capabilities and server
  * configuration.
  */
-@FunctionalInterface
 public interface TcpConnectionHandler {
 
 	void handle(TcpConnection connection);

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

@@ -0,0 +1,13 @@
+package io.ripc.protocol.tcp.connection.listener;
+
+import io.ripc.protocol.tcp.connection.TcpConnection;
+import io.ripc.protocol.tcp.connection.TcpConnectionEventListener;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public interface WriteCompleteListener extends TcpConnectionEventListener {
+
+	boolean writeComplete(TcpConnection connection, long count, Object msg);
+
+}

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

@@ -0,0 +1,61 @@
+package io.ripc.transport.netty4.tcp;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.ripc.protocol.tcp.connection.TcpConnection;
+import io.ripc.protocol.tcp.connection.listener.WriteCompleteListener;
+
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public class TcpConnectionEventListenerChannelHandler extends ChannelDuplexHandler {
+
+	private static final AtomicLongFieldUpdater<TcpConnectionEventListenerChannelHandler> WRITTEN_UPD
+			= AtomicLongFieldUpdater.newUpdater(TcpConnectionEventListenerChannelHandler.class, "writtenSinceLastFlush");
+
+	private final TcpConnection connection;
+
+	private volatile long writtenSinceLastFlush = 0L;
+
+	private WriteCompleteListener writeCompleteListener;
+
+	public TcpConnectionEventListenerChannelHandler(TcpConnection connection) {
+		this.connection = connection;
+	}
+
+	public WriteCompleteListener getWriteCompleteListener() {
+		return writeCompleteListener;
+	}
+
+	public void setWriteCompleteListener(WriteCompleteListener writeCompleteListener) {
+		this.writeCompleteListener = writeCompleteListener;
+	}
+
+	@Override
+	public void flush(ChannelHandlerContext ctx) throws Exception {
+		super.flush(ctx);
+		WRITTEN_UPD.set(this, 0L);
+	}
+
+	@Override
+	public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+		if (null != writeCompleteListener) {
+			promise.addListener(new ChannelFutureListener() {
+				@Override
+				public void operationComplete(ChannelFuture future) throws Exception {
+					if (future.isSuccess() && writeCompleteListener.writeComplete(connection, writtenSinceLastFlush, msg)) {
+						ctx.flush();
+					}
+				}
+			});
+		}
+		super.write(ctx, msg, promise);
+		WRITTEN_UPD.incrementAndGet(this);
+	}
+
+}

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

@@ -8,7 +8,7 @@
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.logging.LoggingHandler;
-import io.ripc.protocol.tcp.TcpConnectionHandler;
+import io.ripc.protocol.tcp.connection.TcpConnectionHandler;
 import io.ripc.transport.netty4.NamedDaemonThreadFactory;
 
 /**

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

@@ -2,10 +2,11 @@
 
 import io.netty.channel.Channel;
 import io.ripc.core.DemandCalculator;
-import io.ripc.protocol.tcp.TcpConnection;
-import io.ripc.protocol.tcp.TcpConnectionEventHandler;
+import io.ripc.protocol.tcp.connection.TcpConnection;
+import io.ripc.protocol.tcp.connection.TcpConnectionEventListener;
+import io.ripc.protocol.tcp.connection.listener.WriteCompleteListener;
 import io.ripc.transport.netty4.tcp.ChannelInboundHandlerSubscription;
-import io.ripc.transport.netty4.tcp.NettyChannelTcpConnectionEventHandler;
+import io.ripc.transport.netty4.tcp.TcpConnectionEventListenerChannelHandler;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
@@ -23,21 +24,24 @@ public class NettyTcpServerConnection implements TcpConnection {
 	private final Channel       channel;
 	private final ReadPublisher readPublisher;
 
-	private TcpConnectionEventHandler eventHandler;
-
 	public NettyTcpServerConnection(Channel channel) {
 		this.channel = channel;
 		this.readPublisher = new ReadPublisher(channel);
 	}
 
 	@Override
-	public TcpConnection eventHandler(TcpConnectionEventHandler eventHandler) {
-		if (null != this.eventHandler) {
-			throw new IllegalArgumentException(TcpConnectionEventHandler.class.getSimpleName()
-			                                   + " already set on this connection");
+	public TcpConnection addListener(TcpConnectionEventListener listener) {
+		TcpConnectionEventListenerChannelHandler handler =
+				channel.pipeline().get(TcpConnectionEventListenerChannelHandler.class);
+		if (null == handler) {
+			handler = new TcpConnectionEventListenerChannelHandler(this);
+			channel.pipeline().addLast(handler);
+		}
+
+		if (WriteCompleteListener.class.isAssignableFrom(listener.getClass())) {
+			handler.setWriteCompleteListener((WriteCompleteListener) listener);
 		}
-		this.eventHandler = eventHandler;
-		channel.pipeline().addLast(new NettyChannelTcpConnectionEventHandler(eventHandler, this));
+
 		return this;
 	}
 
@@ -47,11 +51,11 @@ public Publisher<?> reader() {
 	}
 
 	@Override
-	public TcpConnection writer(Publisher<?> sink) {
-		DemandCalculator demandCalculator = DemandCalculator.class.isAssignableFrom(sink.getClass())
-		                                    ? (DemandCalculator) sink
+	public TcpConnection writer(Publisher<?> writer) {
+		DemandCalculator demandCalculator = DemandCalculator.class.isAssignableFrom(writer.getClass())
+		                                    ? (DemandCalculator) writer
 		                                    : null;
-		sink.subscribe(new WriteSubscriber(demandCalculator));
+		writer.subscribe(new WriteSubscriber(demandCalculator));
 		return this;
 	}
 
@@ -126,9 +130,7 @@ public void onNext(Object msg) {
 
 		@Override
 		public void onError(Throwable t) {
-			if (null != eventHandler) {
-				eventHandler.onError(NettyTcpServerConnection.this, t);
-			}
+			channel.close();
 		}
 
 		@Override