Add WritePublisher

Author: rstoyanchev

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

@@ -0,0 +1,77 @@
+package io.ripc.core;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+/**
+ * A {@code Publisher} that adapts another for write I/O, for example applying
+ * requests to flush based on some strategy.
+ */
+public class WritePublisher<T> implements Publisher<T> {
+
+	private final Publisher<T> adaptee;
+
+	private int flushCount;
+
+
+	private WritePublisher(Publisher<T> adaptee) {
+		this.adaptee = adaptee;
+	}
+
+	/**
+	 * Create a WritePublisher by wrapping the actual Publisher.
+	 */
+	public static <S> WritePublisher<S> adapt(Publisher<S> publisher) {
+		return new WritePublisher<>(publisher);
+	}
+
+	/**
+	 * Configure how many items should be written before requesting flush.
+	 */
+	public WritePublisher<T> withFlushCount(int count) {
+		this.flushCount = count;
+		return this;
+	}
+
+	@Override
+	public void subscribe(Subscriber<? super T> subscriber) {
+
+		if (this.flushCount > 0 && subscriber instanceof Flushable) {
+			Flushable flushable = (Flushable) subscriber;
+
+			this.adaptee.subscribe(new Subscriber<T>() {
+
+				private volatile int count;
+
+				@Override
+				public void onSubscribe(Subscription s) {
+					subscriber.onSubscribe(s);
+				}
+
+				@Override
+				public void onNext(T t) {
+					subscriber.onNext(t);
+					if (++this.count == WritePublisher.this.flushCount) {
+						flushable.flush();
+						WritePublisher.this.flushCount = 0;
+					}
+				}
+
+				@Override
+				public void onError(Throwable t) {
+					subscriber.onError(t);
+				}
+
+				@Override
+				public void onComplete() {
+					subscriber.onComplete();
+				}
+			});
+		}
+		else {
+			this.adaptee.subscribe(subscriber);
+		}
+	}
+
+}

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

@@ -8,7 +8,9 @@ public interface TcpConnection<I, O> {
 	Publisher<I> reader();
 
 	/**
+	 *
 	 * @see io.ripc.core.Flushable
+	 * @see io.ripc.core.WritePublisher
 	 */
 	Publisher<Void> writer(Publisher<O> publisher);
 

ripc-transport-netty4-examples/src/main/java/io/ripc/transport/netty4/tcp/TcpServerSample.java

@@ -9,7 +9,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
-import io.ripc.core.Flushable;
+import io.ripc.core.WritePublisher;
 import io.ripc.protocol.tcp.SimpleTcpInterceptor;
 import io.ripc.protocol.tcp.TcpConnection;
 import org.reactivestreams.Processor;
@@ -34,7 +34,7 @@ public static void main(String[] args) {
 					CompletionPublisher completionPublisher = new CompletionPublisher();
 					EchoProcessor echoProcessor = new EchoProcessor(completionPublisher);
 					connection.reader().subscribe(echoProcessor);
-					connection.writer(echoProcessor);
+					connection.writer(WritePublisher.adapt(echoProcessor).withFlushCount(1));
 					return completionPublisher;
 				});
 
@@ -83,8 +83,6 @@ private static class EchoProcessor implements Processor<ByteBuf, ByteBuf> {
 
 		private Subscriber<? super ByteBuf> writeSubscriber;
 
-		private Flushable flushable;
-
 
 		private EchoProcessor(CompletionPublisher publisher) {
 			this.completionPublisher = publisher;
@@ -106,7 +104,6 @@ public void onNext(ByteBuf byteBuf) {
 			else {
 				logger.debug("Echoing message: " + message);
 				this.writeSubscriber.onNext(Unpooled.buffer().writeBytes(("Hello " + message).getBytes(charset)));
-				this.flushable.flush();
 			}
 		}
 
@@ -121,7 +118,6 @@ public void onComplete() {
 		@Override
 		public void subscribe(Subscriber<? super ByteBuf> subscriber) {
 			this.writeSubscriber = subscriber;
-			this.flushable = (subscriber instanceof Flushable ? (Flushable) subscriber : new NoOpFlushable());
 			this.writeSubscriber.onSubscribe(new Subscription() {
 				@Override
 				public void request(long n) {
@@ -137,14 +133,6 @@ public void cancel() {
 		}
 	}
 
-	private static class NoOpFlushable implements Flushable {
-
-		@Override
-		public void flush() {
-			// no-op
-		}
-	}
-
 	/**
 	 * Naive implementation
 	 */