Attempt to make it more functional by not forcing the ConnectionHandler to be a singleton (it **can** be, it just doesn't have to be). Introduced some functional interfaces to leverage like Consumer and Supplier but extended those into specific subinterfaces that have strongly-typed parameters.

Author: jbrisbin

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

@@ -0,0 +1,11 @@
+package io.ripc.core;
+
+/**
+ * Simple functional interface for accepting objects via callback.
+ */
+@FunctionalInterface
+public interface Consumer<T> {
+
+	void accept(T obj);
+
+}

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

@@ -0,0 +1,9 @@
+package io.ripc.core;
+
+/**
+ * Simple functional interface for applying transformations to objects.
+ */
+@FunctionalInterface
+public interface Function<T, V> {
+	V apply(T obj);
+}

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

@@ -0,0 +1,9 @@
+package io.ripc.core;
+
+/**
+ * A simple functional interface to provide instances of a given object.
+ */
+@FunctionalInterface
+public interface Supplier<T> {
+	T get();
+}

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

@@ -0,0 +1,9 @@
+package io.ripc.core.io;
+
+import io.ripc.core.Supplier;
+
+/**
+ * Created by jbrisbin on 3/18/15.
+ */
+public interface BufferSupplier<B> extends Supplier<Buffer<B>> {
+}

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

@@ -1,16 +1,12 @@
 package io.ripc.protocol.tcp;
 
+import io.ripc.core.Consumer;
+import io.ripc.core.Supplier;
+
 /**
- * A {@code ConnectionHandler} is responsible for composing a Reactive Stream pipeline(s) when a new connection is
- * received by the server.
+ * 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.
  */
-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);
-
+public interface ConnectionHandler<B> extends Supplier<Consumer<Connection<B>>> {
 }

ripc-transport-netty4/src/main/java/io/ripc/transport/netty4/tcp/server/NettyTcpServer.java

@@ -15,15 +15,12 @@
 /**
  * Created by jbrisbin on 3/10/15.
  */
-public class NettyTcpServer extends ChannelInitializer<SocketChannel> {
+public class NettyTcpServer {
 
-	private final ServerBootstrap            bootstrap;
-	private final ConnectionHandler<ByteBuf> handler;
+	private final ServerBootstrap bootstrap;
 
-	public NettyTcpServer(ServerBootstrap bootstrap,
-	                      ConnectionHandler<ByteBuf> handler) {
+	public NettyTcpServer(ServerBootstrap bootstrap) {
 		this.bootstrap = bootstrap;
-		this.handler = handler;
 	}
 
 	public static NettyTcpServer listen(int port, ConnectionHandler<ByteBuf> handler) {
@@ -36,8 +33,19 @@ public static NettyTcpServer listen(int port, ConnectionHandler<ByteBuf> handler
 
 		b.channel(NioServerSocketChannel.class);
 
-		NettyTcpServer server = new NettyTcpServer(b, handler);
-		b.childHandler(server);
+		NettyTcpServer server = new NettyTcpServer(b);
+		b.childHandler(new ChannelInitializer<SocketChannel>() {
+			@Override
+			protected void initChannel(SocketChannel ch) throws Exception {
+				ch.config().setAutoRead(false);
+				ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);
+
+				ch.pipeline().addLast(new LoggingHandler());
+
+				NettyTcpServerConnection conn = new NettyTcpServerConnection(ch);
+				handler.get().accept(conn);
+			}
+		});
 
 		b.bind(port);
 
@@ -48,15 +56,4 @@ public void shutdown() {
 		bootstrap.group().shutdownGracefully();
 	}
 
-	@Override
-	protected void initChannel(SocketChannel ch) throws Exception {
-		ch.config().setAutoRead(false);
-		ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);
-
-		ch.pipeline().addLast(new LoggingHandler());
-
-		NettyTcpServerConnection conn = new NettyTcpServerConnection(ch);
-		handler.handle(conn);
-	}
-
 }

ripc-transport-netty4/src/test/java/io/ripc/transport/netty4/tcp/server/NettyTcpServerIntegrationTests.java

@@ -25,7 +25,7 @@ public class NettyTcpServerIntegrationTests {
 	public void canStartNettyTcpServer() throws InterruptedException {
 		CountDownLatch latch = new CountDownLatch(1);
 
-		NettyTcpServer server = NettyTcpServer.listen(3000, conn -> {
+		NettyTcpServer server = NettyTcpServer.listen(3000, () -> conn -> {
 			conn.subscribe(new Subscriber<Buffer<ByteBuf>>() {
 				private Subscription subscription;
 
@@ -78,7 +78,6 @@ public void cancel() {
 					});
 				}
 			});
-
 		});
 
 		while (!latch.await(1, TimeUnit.SECONDS)) {