TCP POC

- read/write back-pressure
- interception as a chain of handlers
- netty transport parameterized by ByteBuf (for now)
- echo sample

Author: rstoyanchev

build.gradle

@@ -11,6 +11,9 @@ ext {
 	// Network Transports
 	nettyVersion = '4.0.26.Final'
 
+	// Composition Libraries
+	rxjava1Version = '1.0.7'
+
 	// Testing
 	mockitoVersion = '1.10.19'
 	junitVersion = '4.12'
@@ -44,11 +47,6 @@ allprojects {
 			"-Xlint:-fallthrough", // intentionally disabled
 			"-Xlint:-rawtypes"     // TODO enable and fix warnings
 	]
-	
-	compileJava {
-		sourceCompatibility = 1.7
-		targetCompatibility = 1.7
-	}
 
 	compileTestJava {
 		sourceCompatibility = 1.8
@@ -71,35 +69,82 @@ subprojects { subproject ->
 	}
 }
 
-project('reactive-ipc-core') {
+project('ripc-core') {
 	description = 'Reactive IPC Core Components'
 	dependencies {
 		// Reactive Streams
 		compile "org.reactivestreams:reactive-streams:$reactiveStreamsVersion"
 	}
 }
 
-project('reactive-ipc-tcp') {
+project('ripc-protocol-tcp') {
 	description = 'Reactive IPC TCP Components'
 	dependencies {
 		// ripc-core
-		compile project(":reactive-ipc-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('reactive-ipc-netty4') {
+project('ripc-transport-netty4') {
 	description = 'Reactive IPC Netty 4.x Transport Implementation'
 	dependencies {
 		// ripc-tcp
-		compile project(":reactive-ipc-tcp")
+		compile project(":ripc-protocol-tcp")
 
 		// Netty
 		compile "io.netty:netty-all:$nettyVersion"
+		compile "org.slf4j:slf4j-api:1.7.6"
+	}
+}
+
+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-api:$slf4jVersion"
+		compile "ch.qos.logback:logback-classic:$logbackVersion"
+	}
+}
+
+project('ripc-rxjava1') {
+	description = 'Reactive IPC Composition Layer Implementation'
+	dependencies {
+		// ripc-tcp
+		compile project(":ripc-protocol-tcp")
+		compile 'io.reactivex:rxjava:1.0.8'
+		compile 'io.reactivex:rxjava-reactive-streams:0.3.0'
+	}
+}
+
+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")
 	}
 }
 
 configure(rootProject) {
-	description = "Reactive Streams IPC"
+	description = "Reactive IPC"
 
 	task api(type: Javadoc) {
 		group = "Documentation"
@@ -111,7 +156,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"
@@ -129,4 +174,4 @@ configure(rootProject) {
 			classpath = files(subprojects.collect { it.sourceSets.main.compileClasspath })
 		}
 	}
-}
+}

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

@@ -0,0 +1,21 @@
+package io.ripc.core;
+
+
+/**
+ * A contract to request flush operations.
+ *
+ * <p>Intended to be implemented by a {@link org.reactivestreams.Subscriber}
+ * performing write I/O and to be used within the context of the
+ * {@link org.reactivestreams.Publisher} producing items to be written.
+ * A {@link #flush()} may be invoked in between calls to
+ * {@link org.reactivestreams.Subscriber#onNext(Object) onNext} to request
+ * flushing all items that may have been buffered so far.
+ */
+public interface Flushable {
+
+	/**
+	 * Request a flush operation.
+	 */
+	public void flush();
+
+}

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

@@ -0,0 +1,29 @@
+package io.ripc.protocol.tcp;
+
+/**
+ * Base class for TcpInterceptor implementations.
+ *
+ * <p>Implementations that do not transform the connection input or output should
+ * consider extending {@link io.ripc.protocol.tcp.SimpleTcpInterceptor}.
+ */
+public abstract class AbstractTcpInterceptor<I, O, II, OO> implements TcpInterceptor<I, O, II, OO> {
+
+	private TcpHandler<II, OO> nextHandler;
+
+
+	public TcpHandler<II, OO> getNext() {
+		return this.nextHandler;
+	}
+
+	@Override
+	public <III, OOO> TcpInterceptor<II, OO, III, OOO> next(TcpInterceptor<II, OO, III, OOO> handler) {
+		this.nextHandler = handler;
+		return handler;
+	}
+
+	@Override
+	public void last(TcpHandler<II, OO> handler) {
+		this.nextHandler = handler;
+	}
+
+}

reactive-ipc-netty4/src/main/java/io/ripc/transport/netty4/package-info.java

@@ -0,0 +1,35 @@
+package io.ripc.protocol.tcp;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractTcpServer<I, O> implements TcpServer {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	private TcpHandler<I, O> handler;
+
+
+	/**
+	 * Set up an interception chain of handlers vs a single {@link #handler}.
+	 */
+	public <II, OO> TcpInterceptor<I, O, II, OO> interceptor(TcpInterceptor<I, O, II, OO> handler) {
+		this.handler = handler;
+		return handler;
+	}
+
+	/**
+	 * Set a single handler vs an interception chain of {@link #interceptor}.
+	 */
+	public void handler(TcpHandler<I, O> handler) {
+		this.handler = handler;
+	}
+
+	/**
+	 * Return the configured TcpHandler.
+	 */
+	protected TcpHandler<I, O> getHandler() {
+		return this.handler;
+	}
+
+}

reactive-ipc-netty4/src/test/java/io/ripc/transport/netty4/package-info.java

@@ -0,0 +1,9 @@
+package io.ripc.protocol.tcp;
+
+
+/**
+ * A {@link TcpInterceptor} that does not change the connection input and output.
+ */
+public abstract class SimpleTcpInterceptor<I, O> extends AbstractTcpInterceptor<I, O, I, O> {
+
+}

reactive-ipc-tcp/src/main/java/io/ripc/protocol/tcp/package-info.java

@@ -0,0 +1,15 @@
+package io.ripc.protocol.tcp;
+
+import org.reactivestreams.Publisher;
+
+
+public interface TcpConnection<I, O> {
+
+	Publisher<I> reader();
+
+	/**
+	 * @see io.ripc.core.Flushable
+	 */
+	Publisher<Void> writer(Publisher<O> publisher);
+
+}

reactive-ipc-tcp/src/test/java/io/ripc/protocol/tcp/package-info.java

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

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

@@ -0,0 +1,25 @@
+package io.ripc.protocol.tcp;
+
+
+/**
+ * A {@code TcpHandler} that can be chained to another handler to form an
+ * interception chain and may transform the connection input and output.
+ *
+ * @param <I> Input (this handler)
+ * @param <O> Output (this handler)
+ * @param <II> Input (next handler)
+ * @param <OO> Output (next handler)
+ */
+public interface TcpInterceptor<I, O, II, OO> extends TcpHandler<I, O> {
+
+	/**
+	 * Configure the next interceptor.
+	 */
+	<III, OOO> TcpInterceptor<II, OO, III, OOO> next(TcpInterceptor<II, OO, III, OOO> handler);
+
+	/**
+	 * Configure the last handler.
+	 */
+	void last(TcpHandler<II, OO> handler);
+
+}

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

@@ -0,0 +1,10 @@
+package io.ripc.protocol.tcp;
+
+
+public interface TcpServer {
+
+	void start();
+
+	void shutdown();
+
+}