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

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.5.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.5.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.5.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/internal/Publishers.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/io/Buffer.java → ripc-core/src/main/java/io/ripc/io/Buffer.java

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

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

@@ -0,0 +1,23 @@
+package io.ripc.protocol.tcp;
+
+import org.reactivestreams.Publisher;
+
+/**
+ * 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 TcpConnection<R, W> extends Publisher<R> {
+
+    /**
+     * 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);
+
+}

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

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

ripc-protocol-tcp/src/main/java/io/ripc/protocol/tcp/TcpInterceptor.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-protocol-tcp/src/main/java/io/ripc/protocol/tcp/TcpServer.java

@@ -0,0 +1,39 @@
+package io.ripc.protocol.tcp;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public abstract class TcpServer<R, W> {
+
+    protected final TcpHandler<R, W> thisHandler;
+    protected final AtomicBoolean started;
+
+    protected TcpServer() {
+        thisHandler = null;
+        this.started = new AtomicBoolean();
+    }
+
+    public final TcpServer<R, W> start(TcpHandler<R, W> handler) {
+        if (!started.compareAndSet(false, true)) {
+            throw new IllegalStateException("Server already started");
+        }
+
+        doStart(handler);
+        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 TcpServer<R, W> doStart(TcpHandler<R, W> handler);
+
+}

ripc-rxjava1-examples/src/main/java/io/rpc/rx/protocol/tcp/RxTcpServerSample.java

@@ -0,0 +1,25 @@
+package io.rpc.rx.protocol.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.ripc.protocol.tcp.TcpServer;
+import io.ripc.rx.protocol.tcp.RxTcpServer;
+import io.ripc.transport.netty4.tcp.Netty4TcpServer;
+import rx.Observable;
+
+import static java.nio.charset.Charset.*;
+
+public class RxTcpServerSample {
+
+    public static void main(String[] args) throws InterruptedException {
+
+        TcpServer<ByteBuf, ByteBuf> transport = Netty4TcpServer.<ByteBuf, ByteBuf>create(0);
+
+        RxTcpServer.create(transport)
+                   .startAndAwait(connection -> connection.flatMap(bb -> {
+                       String msgStr = "Hello " + bb.toString(defaultCharset());
+                       ByteBuf msg = Unpooled.buffer().writeBytes(msgStr.getBytes());
+                       return connection.write(Observable.just(msg).doOnCompleted(() -> System.out.println("Done!")));
+                   }));
+    }
+}