Revamp code to support dynamic backpressure and to reflect discussions around changing the role of the TcpConnection.

Author: jbrisbin

build.gradle

@@ -6,13 +6,13 @@ ext {
 
 	// Logging
 	slf4jVersion = '1.7.10'
-	logbackVersion = '1.1.2'
+	logbackVersion = '1.1.3'
 
 	// Network Transports
 	nettyVersion = '4.0.26.Final'
 
 	// Composition Libraries
-	rxjava1Version = '1.0.7'
+	rxjava1Version = '1.0.8'
 
 	// Testing
 	mockitoVersion = '1.10.19'
@@ -69,6 +69,7 @@ subprojects { subproject ->
 		// Testing
 		testCompile "junit:junit:$junitVersion",
 				"org.hamcrest:hamcrest-library:1.3",
+				"org.mockito:mockito-core:$mockitoVersion",
 				"org.slf4j:slf4j-api:$slf4jVersion"
 		testRuntime "ch.qos.logback:logback-classic:$logbackVersion"
 	}

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

@@ -0,0 +1,10 @@
+package io.ripc.core;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public interface DemandCalculator {
+
+	long calculateDemand(long pending);
+
+}

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

@@ -0,0 +1,17 @@
+package io.ripc.core;
+
+import org.reactivestreams.Publisher;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public final class Publishers {
+
+	private Publishers() {
+	}
+
+	public static <T> Publisher<?> just(final T obj) {
+		return new SingletonPublisher<>(obj);
+	}
+
+}

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

@@ -0,0 +1,50 @@
+package io.ripc.core;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public class SingletonPublisher<T> implements Publisher<T>, DemandCalculator {
+
+	private final AtomicBoolean requested = new AtomicBoolean(false);
+
+	private final T value;
+
+	public SingletonPublisher(T value) {
+		this.value = value;
+	}
+
+	@Override
+	public long calculateDemand(long pending) {
+		return (requested.get() ? -1 : 1);
+	}
+
+	@Override
+	public void subscribe(final Subscriber<? super T> subscriber) {
+		subscriber.onSubscribe(new Subscription() {
+			@Override
+			public void request(long n) {
+				if (!Specification.spec_3_9_verifyPositiveDemand(n, subscriber)) {
+					return;
+				}
+				if (requested.compareAndSet(false, true)) {
+					if (null != value) {
+						subscriber.onNext(value);
+					}
+					subscriber.onComplete();
+				}
+			}
+
+			@Override
+			public void cancel() {
+				requested.set(true);
+			}
+		});
+	}
+
+}

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

@@ -0,0 +1,48 @@
+package io.ripc.protocol.tcp;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public abstract class AbstractTcpConnectionEventHandler implements TcpConnectionEventHandler, CompleteEventHandler {
+
+	@Override
+	public void onOpen(TcpConnection connection) {
+		// NO-OP
+	}
+
+	@Override
+	public void onClose(TcpConnection connection) {
+		// NO-OP
+	}
+
+	@Override
+	public void onAbort(TcpConnection connection) {
+		// NO-OP
+	}
+
+	@Override
+	public void onError(TcpConnection connection, Throwable cause) {
+		// NO-OP
+	}
+
+	@Override
+	public void onReadable(TcpConnection connection) {
+		// NO-OP
+	}
+
+	@Override
+	public void onWritable(TcpConnection connection) {
+		// NO-OP
+	}
+
+	@Override
+	public boolean onReadComplete(TcpConnection connection) {
+		return true;
+	}
+
+	@Override
+	public boolean onWriteComplete(TcpConnection connection, Object msg) {
+		return false;
+	}
+
+}

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

@@ -0,0 +1,12 @@
+package io.ripc.protocol.tcp;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public interface CompleteEventHandler {
+
+	boolean onReadComplete(TcpConnection connection);
+
+	boolean onWriteComplete(TcpConnection connection, Object msg);
+
+}

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

@@ -0,0 +1,16 @@
+package io.ripc.protocol.tcp;
+
+import org.reactivestreams.Publisher;
+
+/**
+ * A {@code Connection} provides a reader for inbound data and a writer for outbound.
+ */
+public interface TcpConnection {
+
+	TcpConnection eventHandler(TcpConnectionEventHandler eventHandler);
+
+	Publisher<?> reader();
+
+	TcpConnection writer(Publisher<?> sink);
+
+}

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

@@ -0,0 +1,20 @@
+package io.ripc.protocol.tcp;
+
+/**
+ * Created by jbrisbin on 3/26/15.
+ */
+public interface TcpConnectionEventHandler {
+
+	void onOpen(TcpConnection connection);
+
+	void onClose(TcpConnection connection);
+
+	void onAbort(TcpConnection connection);
+
+	void onError(TcpConnection connection, Throwable cause);
+
+	void onReadable(TcpConnection connection);
+
+	void onWritable(TcpConnection connection);
+
+}