Based on a discussion today with @rstoyanchev and others, I updated the sketch here to remove the server as a Publisher of connections and provided a simple callback in its place. I also implemented the write functionality in a very naive way. It's a batch publisher, though, in that it won't flush the connection until the onComplete signal is received. This is another discussion (whether flush should be tied to onComplete or not) so this code should be looked at only as a point of discussion.

Author: jbrisbin

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

@@ -4,10 +4,16 @@
 import org.reactivestreams.Publisher;
 
 /**
- * Created by jbrisbin on 3/10/15.
+ * A {@code Connection} is a Reactive Streams {@link org.reactivestreams.Publisher} that provides subscribers with
+ * inbound data and exposes the {@link #write(org.reactivestreams.Publisher)} method for sending outbound data.
  */
 public interface Connection<B> extends Publisher<Buffer<B>> {
 
+	/**
+	 * Send outbound data using the Reactive Streams {@code Publisher} contract.
+	 *
+	 * @param data publisher of outbound data
+	 */
 	void write(Publisher<Buffer<B>> data);
 
 }

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

@@ -0,0 +1,16 @@
+package io.ripc.protocol.tcp;
+
+/**
+ * A {@code ConnectionHandler} is responsible for composing a Reactive Stream pipeline(s) when a new connection is
+ * received by the server.
+ */
+public interface ConnectionHandler<B> {
+
+	/**
+	 * Implementations will compose an appropriate pipeline based on capabilities and server configuration.
+	 *
+	 * @param connection connection around which to compose the pipeline
+	 */
+	void handle(Connection<B> connection);
+
+}

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

@@ -10,11 +10,16 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
 /**
  * Created by jbrisbin on 3/10/15.
  */
 public class ChannelInboundHandlerSubscription extends ChannelInboundHandlerAdapter implements Subscription {
 
+	private static final AtomicLongFieldUpdater<ChannelInboundHandlerSubscription> PEND_UPD
+			= AtomicLongFieldUpdater.newUpdater(ChannelInboundHandlerSubscription.class, "pending");
+
 	private final Channel                             channel;
 	private final Subscriber<? super Buffer<ByteBuf>> subscriber;
 
@@ -38,16 +43,16 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
 	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 		if (pending < 1) {
 			super.channelInactive(ctx);
-			return;
 		}
-		subscriber.onComplete();
 	}
 
 	@Override
 	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
 		if (pending < 1) {
 			super.channelReadComplete(ctx);
+			return;
 		}
+		subscriber.onComplete();
 	}
 
 	@Override
@@ -60,10 +65,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
 		ByteBuf buf = (ByteBuf) msg;
 		try {
 			subscriber.onNext(new ByteBufBuffer(buf, false));
-			synchronized (this) {
-				pending--;
-			}
-			channel.read();
+			PEND_UPD.decrementAndGet(this);
+			//channel.read();
 		} catch (Throwable t) {
 			subscriber.onError(t);
 		}
@@ -75,13 +78,12 @@ public void request(long demand) {
 			return;
 		}
 
-		synchronized (this) {
-			if (demand < Long.MAX_VALUE) {
-				pending += demand;
-			} else {
-				pending = Long.MAX_VALUE;
-			}
+		if (demand < Long.MAX_VALUE) {
+			PEND_UPD.addAndGet(this, demand);
+		} else {
+			PEND_UPD.set(this, Long.MAX_VALUE);
 		}
+
 		channel.read();
 	}
 

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

@@ -8,29 +8,25 @@
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.logging.LoggingHandler;
 import io.ripc.core.NamedDaemonThreadFactory;
-import io.ripc.protocol.tcp.Connection;
-import io.ripc.protocol.tcp.ConnectionPublisher;
-import io.ripc.transport.netty4.tcp.ChannelInitializerSubscription;
-import org.reactivestreams.Subscriber;
-
-import java.util.ArrayList;
-import java.util.List;
+import io.ripc.protocol.tcp.ConnectionHandler;
 
 /**
  * Created by jbrisbin on 3/10/15.
  */
-public class NettyTcpServer extends ChannelInitializer<SocketChannel> implements ConnectionPublisher<ByteBuf> {
-
-	private final List<ChannelInitializerSubscription> subscriptions = new ArrayList<>();
+public class NettyTcpServer extends ChannelInitializer<SocketChannel> {
 
-	private final ServerBootstrap bootstrap;
+	private final ServerBootstrap            bootstrap;
+	private final ConnectionHandler<ByteBuf> handler;
 
-	public NettyTcpServer(ServerBootstrap bootstrap) {
+	public NettyTcpServer(ServerBootstrap bootstrap,
+	                      ConnectionHandler<ByteBuf> handler) {
 		this.bootstrap = bootstrap;
+		this.handler = handler;
 	}
 
-	public static ConnectionPublisher<ByteBuf> listen(int port) {
+	public static NettyTcpServer listen(int port, ConnectionHandler<ByteBuf> handler) {
 		ServerBootstrap b = new ServerBootstrap();
 
 		int threads = Runtime.getRuntime().availableProcessors();
@@ -40,32 +36,27 @@ public static ConnectionPublisher<ByteBuf> listen(int port) {
 
 		b.channel(NioServerSocketChannel.class);
 
-		NettyTcpServer server = new NettyTcpServer(b);
+		NettyTcpServer server = new NettyTcpServer(b, handler);
 		b.childHandler(server);
 
 		b.bind(port);
 
 		return server;
 	}
 
+	public void shutdown() {
+		bootstrap.group().shutdownGracefully();
+	}
+
 	@Override
 	protected void initChannel(SocketChannel ch) throws Exception {
 		ch.config().setAutoRead(false);
 		ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);
 
-		synchronized (subscriptions) {
-			for (ChannelInitializerSubscription sub : subscriptions) {
-				ch.pipeline().addLast(sub);
-			}
-		}
-	}
+		ch.pipeline().addLast(new LoggingHandler());
 
-	@Override
-	public void subscribe(Subscriber<? super Connection<ByteBuf>> s) {
-		ChannelInitializerSubscription sub = new ChannelInitializerSubscription(s);
-		synchronized (subscriptions) {
-			subscriptions.add(sub);
-		}
-		s.onSubscribe(sub);
+		NettyTcpServerConnection conn = new NettyTcpServerConnection(ch);
+		handler.handle(conn);
 	}
+
 }

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

@@ -2,24 +2,18 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
-import io.netty.handler.logging.LogLevel;
-import io.netty.handler.logging.LoggingHandler;
 import io.ripc.core.io.Buffer;
 import io.ripc.protocol.tcp.Connection;
 import io.ripc.transport.netty4.tcp.ChannelInboundHandlerSubscription;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
-
-import java.util.ArrayList;
-import java.util.List;
+import org.reactivestreams.Subscription;
 
 /**
- * Created by jbrisbin on 3/10/15.
+ * Represents a Netty Channel.
  */
 public class NettyTcpServerConnection implements Connection<ByteBuf> {
 
-	private final List<Subscriber<? super Buffer<ByteBuf>>> subscribers = new ArrayList<>();
-
 	private final Channel channel;
 
 	public NettyTcpServerConnection(Channel channel) {
@@ -28,25 +22,54 @@ public NettyTcpServerConnection(Channel channel) {
 
 	@Override
 	public void write(Publisher<Buffer<ByteBuf>> data) {
-
+		data.subscribe(new WriteSubscriber());
 	}
 
 	@Override
 	public void subscribe(Subscriber<? super Buffer<ByteBuf>> s) {
 		ChannelInboundHandlerSubscription sub = new ChannelInboundHandlerSubscription(channel, s);
-		synchronized (this) {
-			subscribers.add(s);
-		}
 		s.onSubscribe(sub);
 		channel.pipeline()
-		       .addLast(new LoggingHandler(LogLevel.DEBUG), sub);
+		       .addLast(sub);
 	}
 
 	@Override
 	public String toString() {
 		return "NettyTcpServerConnection{" +
-				"channel=" + channel +
-				'}';
+		       "channel=" + channel +
+		       '}';
+	}
+
+	private final class WriteSubscriber implements Subscriber<Buffer<ByteBuf>> {
+		Subscription subscription;
+
+		@Override
+		public void onSubscribe(Subscription s) {
+			if (null != subscription) {
+				s.cancel();
+				return;
+			}
+			(subscription = s).request(1);
+		}
+
+		@Override
+		public void onNext(Buffer<ByteBuf> buffer) {
+			channel.write(buffer.get());
+			// This causes a StackOverflowError
+			//subscription.request(1);
+			// This doesn't
+			channel.eventLoop().execute(() -> subscription.request(1));
+		}
+
+		@Override
+		public void onError(Throwable t) {
+
+		}
+
+		@Override
+		public void onComplete() {
+			channel.flush();
+		}
 	}
 
 }