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.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/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/Connection.java

@@ -1,19 +1,53 @@
 package io.ripc.protocol.tcp;
 
-import io.ripc.core.io.Buffer;
 import org.reactivestreams.Publisher;
 
 /**
- * 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.
+ * 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> {
 
-	/**
-	 * Send outbound data using the Reactive Streams {@code Publisher} contract.
-	 *
-	 * @param data publisher of outbound data
-	 */
-	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-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(Connection<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);
+
+}