Proposal for TCP server at all layers (core, transport & rxjava1)

Based on various discussions around different issues, I am proposing a fresh design for a TCP server in this PR.
I have implemented a TCP server at all layers in reactive-ipc, i.e. core, transport(netty) and rxjava1. I have also removed all code which isn't related to the implementation. We can discuss those utilities in isolation as and when required to be added. Although, not used, I have kept the Buffer and related classes as that aspect is not yet discussed to conclusion.

There are two samples each for netty transport and rxjava1 layer available to see how will a typical interaction look like in modules ripc-transport-netty4-examples and ripc-rxjava1-examples

Backpressure has not been implemented yet.

Author: Nitesh Kant

build.gradle

@@ -48,11 +48,6 @@ allprojects {
 			"-Xlint:-rawtypes"     // TODO enable and fix warnings
 	]
 
-	compileJava {
-		sourceCompatibility = 1.7
-		targetCompatibility = 1.7
-	}
-
 	compileTestJava {
 		sourceCompatibility = 1.8
 		targetCompatibility = 1.8
@@ -87,9 +82,29 @@ project('ripc-protocol-tcp') {
 	dependencies {
 		// ripc-core
 		compile project(":ripc-core")
+        compile "org.slf4j:slf4j-api:1.7.6"
+		testCompile 'io.reactivex:rxjava:1.0.8'
+        testCompile 'io.reactivex:rxjava-reactive-streams:0.3.0'
 	}
 }
 
+project('ripc-transport-netty4-examples') {
+	description = 'Reactive IPC TCP Component examples'
+
+    compileTestJava {
+        sourceCompatibility = 1.8
+        targetCompatibility = 1.8
+    }
+
+    dependencies {
+		// ripc-core
+		compile project(":ripc-transport-netty4")
+        compile 'io.reactivex:rxjava:1.0.8'
+        compile 'io.reactivex:rxjava-reactive-streams:0.3.0'
+        compile 'org.slf4j:slf4j-simple:1.7.6'
+    }
+}
+
 project('ripc-transport-netty4') {
 	description = 'Reactive IPC Netty 4.x Transport Implementation'
 	dependencies {
@@ -98,17 +113,32 @@ project('ripc-transport-netty4') {
 
 		// Netty
 		compile "io.netty:netty-all:$nettyVersion"
+        compile "org.slf4j:slf4j-api:1.7.6"
 	}
 }
 
-project('ripc-composition-rxjava1') {
+project('ripc-rxjava1') {
 	description = 'Reactive IPC Composition Layer Implementation'
 	dependencies {
 		// ripc-tcp
-		compile project(":ripc-transport-netty4")
+		compile project(":ripc-protocol-tcp")
+        compile 'io.reactivex:rxjava:1.0.8'
+        compile 'io.reactivex:rxjava-reactive-streams:0.3.0'
+	}
+}
 
-		// RxJava 1.0
-		compile "io.reactivex:rxjava:$rxjava1Version"
+project('ripc-rxjava1-examples') {
+	description = 'Reactive IPC Composition Layer examples'
+
+    compileTestJava {
+        sourceCompatibility = 1.8
+        targetCompatibility = 1.8
+    }
+
+	dependencies {
+		// ripc-tcp
+		compile project(":ripc-rxjava1")
+		compile project(":ripc-transport-netty4-examples")
 	}
 }
 
@@ -125,7 +155,7 @@ configure(rootProject) {
 				it.tasks.getByName("jar")
 			}
 		}
-		options.memberLevel = org.gradle.external.javadoc.JavadocMemberLevel.PROTECTED
+		options.memberLevel = JavadocMemberLevel.PROTECTED
 		options.author = true
 		options.header = rootProject.description
 		//options.overview = "src/api/overview.html"

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

@@ -0,0 +1,48 @@
+package io.ripc.internal;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+/**
+ * Temporary utility class for creating and transforming {@link Publisher}s.
+ */
+public class Publishers {
+
+    public static <T> Publisher<T> just(final T value) {
+        return new Publisher<T>() {
+            @Override
+            public void subscribe(final Subscriber<? super T> s) {
+                s.onSubscribe(new Subscription() {
+                    @Override
+                    public void request(long n) {
+                        s.onNext(value);
+                        s.onComplete();
+                    }
+
+                    @Override
+                    public void cancel() {
+                    }
+                });
+            }
+        };
+    }
+
+    public static <T> Publisher<T> error(final Throwable t) {
+        return new Publisher<T>() {
+            @Override
+            public void subscribe(final Subscriber<? super T> s) {
+                s.onSubscribe(new Subscription() {
+                    @Override
+                    public void request(long n) {
+                        s.onError(t);
+                    }
+
+                    @Override
+                    public void cancel() {
+                    }
+                });
+            }
+        };
+    }
+}

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

@@ -1,4 +1,4 @@
-package io.ripc.core.io;
+package io.ripc.io;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.CharsetDecoder;

ripc-core/src/main/java/io/ripc/core/package-info.java

@@ -1,13 +1,53 @@
 package io.ripc.protocol.tcp;
 
-import io.ripc.core.io.Buffer;
 import org.reactivestreams.Publisher;
 
 /**
- * Created by jbrisbin on 3/10/15.
+ * An abstraction for a TCP connection.
+ *
+ * @param <R> The type of objects read from this connection.
+ * @param <W> The type of objects written to this connection.
  */
-public interface Connection<B> extends Publisher<Buffer<B>> {
+public interface Connection<R, W> extends Publisher<R> {
 
-	void write(Publisher<Buffer<B>> data);
+    /**
+     * Writes the passed stream of {@code data} and returns the result as a {@link Publisher}. All items emitted by
+     * this stream are flushed on completion of the stream.
+     *
+     * @param data Data stream to write.
+     *
+     * @return Result of write.
+     */
+	Publisher<Void> write(Publisher<W> data);
 
+    /**
+     * Writes the passed stream of {@code data} and returns the result as a {@link Publisher}. All written items are
+     * flushed whenever the passed {@code flushSelector} returns {@code true}
+     *
+     * @param data Data stream to write.
+     * @param flushSelector Selector that is invoked after every emitted item is written. If this selector returns
+     * {@code true} then all items written till now are flushed.
+     *
+     * @return Result of write.
+     */
+    Publisher<Void> write(Publisher<W> data, FlushSelector<W> flushSelector);
+
+    /**
+     * A function that is used for determining when a flush has to be invoked on the underlying channel.
+     *
+     * @param <W> Type of items emitted by the stream using this selector.
+     */
+    interface FlushSelector<W> {
+
+        /**
+         * Selects whether flush should be invoked on the channel.
+         *
+         * @param count The index of this item. Starts with 1.
+         * @param lastWrittenItem Item which was last written before calling this selector.
+         *
+         * @return {@code true} if flush should be invoked.
+         */
+        boolean select(long count, W lastWrittenItem);
+
+    }
 }

ripc-core/src/main/java/io/ripc/internal/Publishers.java

@@ -0,0 +1,8 @@
+package io.ripc.protocol.tcp;
+
+import org.reactivestreams.Publisher;
+
+public interface TcpHandler<R, W> {
+
+    Publisher<Void> handle(Connection<R, W> connection);
+}

ripc-core/src/main/java/io/ripc/core/io/Buffer.java renamed to ripc-core/src/main/java/io/ripc/io/Buffer.java

@@ -0,0 +1,7 @@
+package io.ripc.protocol.tcp;
+
+public interface TcpInterceptor<I, O, II, OO> {
+
+    TcpHandler<II, OO> intercept(TcpHandler<I, O> handler);
+
+}

ripc-core/src/test/java/io/ripc/core/package-info.java

@@ -0,0 +1,88 @@
+package io.ripc.protocol.tcp;
+
+import io.ripc.internal.Publishers;
+import org.reactivestreams.Publisher;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public abstract class TcpServer<R, W> {
+
+    @SuppressWarnings("rawtypes")
+    private LazyTcpHandler lazyHandler;
+    protected final TcpHandler<R, W> thisHandler;
+    protected final AtomicBoolean started;
+
+    protected TcpServer() {
+        lazyHandler = new LazyTcpHandler<>();
+        thisHandler = null;
+        this.started = new AtomicBoolean();
+    }
+
+    protected TcpServer(TcpHandler<R, W> handler) {
+        thisHandler = handler;
+        this.started = new AtomicBoolean();
+    }
+
+    public <RR, WW> TcpServer<RR, WW> intercept(TcpInterceptor<R, W, RR, WW> interceptor) {
+        if (null == thisHandler) {
+            @SuppressWarnings("unchecked")
+            TcpServer<RR, WW> toReturn = newServer(interceptor.intercept(lazyHandler));
+            toReturn.lazyHandler = lazyHandler;
+            return toReturn;
+        } else {
+            TcpServer<RR, WW> toReturn = newServer(interceptor.intercept(thisHandler));
+            toReturn.lazyHandler = lazyHandler;
+            return toReturn;
+        }
+    }
+
+    public final TcpServer<R, W> start(TcpHandler<R, W> handler) {
+        if (!started.compareAndSet(false, true)) {
+            throw new IllegalStateException("Server already started");
+        }
+
+        if (null == thisHandler) {
+            doStart(handler);
+        } else {
+            lazyHandler.start(handler);
+            doStart(thisHandler);
+        }
+        return this;
+    }
+
+    public final void startAndAwait(TcpHandler<R, W> handler) {
+        start(handler);
+        awaitShutdown();
+    }
+
+    public final boolean shutdown() {
+        return !started.compareAndSet(true, false) || doShutdown();
+    }
+
+    public abstract void awaitShutdown();
+
+    public abstract boolean doShutdown();
+
+    protected abstract <RR, WW> TcpServer<RR, WW> newServer(TcpHandler<RR, WW> handler);
+
+    protected abstract TcpServer<R, W> doStart(TcpHandler<R, W> handler);
+
+    private static class LazyTcpHandler<R, W> implements TcpHandler<R, W> {
+
+        private TcpHandler<R, W> delegate;
+
+        @Override
+        public Publisher<Void> handle(Connection<R, W> connection) {
+            if (null == delegate) {
+                return Publishers.error(new IllegalStateException("Handler not initialized."));
+            } else {
+                return delegate.handle(connection);
+            }
+        }
+
+        @SuppressWarnings("unchecked")
+        private void start(TcpHandler handler) {
+            delegate = handler;
+        }
+    }
+}