Fix for requestChannel and requestStream (#186)

In the current implementation stream and channel methods are broken.

Fixed stream and channel methods to work end to end. Also added examples for the same.
They are now implemented using the existing `RemoteReceiver` and `RemoteSender` classes.
One major problem with using `ConnectableUnicastProcessor` was that it does not discriminate between subscriber of request and sender of `requestN` and `cancel` frames. So, if the request source is completed, sending `requestN` and `cancel` frames result in them being rejected by the transport. This is one of the reasons `RemoteReceiver` and `RemoteSender` classes were created.

 __ request-response implementation is not changed in this change as that needs to be optimized for a single item request-response__

Channel and stream work now.

Author: NiteshKant

reactivesocket-core/src/main/java/io/reactivesocket/ClientReactiveSocket.java

@@ -20,10 +20,13 @@
 import io.reactivesocket.exceptions.CancelException;
 import io.reactivesocket.exceptions.Exceptions;
 import io.reactivesocket.internal.KnownErrorFilter;
+import io.reactivesocket.internal.RemoteReceiver;
+import io.reactivesocket.internal.RemoteSender;
 import io.reactivesocket.lease.Lease;
 import io.reactivesocket.lease.LeaseImpl;
 import io.reactivesocket.reactivestreams.extensions.DefaultSubscriber;
 import io.reactivesocket.reactivestreams.extensions.Px;
+import io.reactivesocket.reactivestreams.extensions.internal.ValidatingSubscription;
 import io.reactivesocket.reactivestreams.extensions.internal.processors.ConnectableUnicastProcessor;
 import io.reactivesocket.reactivestreams.extensions.internal.subscribers.CancellableSubscriber;
 import io.reactivesocket.reactivestreams.extensions.internal.subscribers.Subscribers;
@@ -50,8 +53,8 @@ public class ClientReactiveSocket implements ReactiveSocket {
     private final StreamIdSupplier streamIdSupplier;
     private final KeepAliveProvider keepAliveProvider;
 
-    private final Int2ObjectHashMap<Processor<Frame, Frame>> senders;
-    private final Int2ObjectHashMap<Subscriber<? super Payload>> receivers;
+    private final Int2ObjectHashMap<Subscription> senders;
+    private final Int2ObjectHashMap<Subscriber<Frame>> receivers;
 
     private volatile Subscription transportReceiveSubscription;
     private CancellableSubscriber<Void> keepAliveSendSub;
@@ -81,86 +84,37 @@ public Publisher<Void> fireAndForget(Payload payload) {
         }
     }
 
+    @Override
     public Publisher<Payload> requestResponse(Payload payload) {
         final int streamId = nextStreamId();
         final Frame requestFrame = Frame.Request.from(streamId, FrameType.REQUEST_RESPONSE, payload, 1);
 
-        return doSendReceive(Px.just(requestFrame), streamId, 1, false);
+        return handleRequestResponse(Px.just(requestFrame), streamId, 1, false);
     }
 
     @Override
     public Publisher<Payload> requestStream(Payload payload) {
         final int streamId = nextStreamId();
         final Frame requestFrame = Frame.Request.from(streamId, FrameType.REQUEST_STREAM, payload, 1);
 
-        return doSendReceive(Px.just(requestFrame), streamId, 1, true);
+        return handleStreamResponse(Px.just(requestFrame), streamId);
     }
 
     @Override
     public Publisher<Payload> requestSubscription(Payload payload) {
         final int streamId = nextStreamId();
         final Frame requestFrame = Frame.Request.from(streamId, FrameType.REQUEST_SUBSCRIPTION, payload, 1);
 
-        return doSendReceive(Px.just(requestFrame), streamId, 1, true);
+        return handleStreamResponse(Px.just(requestFrame), streamId);
     }
 
     @Override
     public Publisher<Payload> requestChannel(Publisher<Payload> payloads) {
         final int streamId = nextStreamId();
-        Px<Frame> frames = Px
-            .from(payloads)
-            .map(payload -> Frame.Request.from(streamId, FrameType.REQUEST_CHANNEL, payload, 1));
-        return doSendReceive(frames, streamId, 1, true);
-    }
-
-    private Publisher<Payload> doSendReceive(final Publisher<Frame> payload, final int streamId, final int initialRequestN, final boolean sendRequestN) {
-        ConnectableUnicastProcessor<Frame> sender = new ConnectableUnicastProcessor<>();
-
-        synchronized (this) {
-            senders.put(streamId, sender);
-        }
-
-        final Runnable cleanup = () -> {
-            synchronized (this) {
-                receivers.remove(streamId);
-                senders.remove(streamId);
-            }
-        };
-
-        return Px
-            .<Payload>create(subscriber -> {
-                synchronized (this) {
-                    receivers.put(streamId, subscriber);
-                }
-
-                payload.subscribe(sender);
-
-                subscriber.onSubscribe(new Subscription() {
-
-                    @Override
-                    public void request(long n) {
-                        if (sendRequestN) {
-                            sender.onNext(Frame.RequestN.from(streamId, n));
-                        }
-                    }
-
-                    @Override
-                    public void cancel() {
-                        sender.onNext(Frame.Cancel.from(streamId));
-                        sender.cancel();
-                    }
-                });
-
-                try {
-                    Px.from(connection.send(sender))
-                      .doOnError(th -> subscriber.onError(th))
-                      .subscribe(DefaultSubscriber.defaultInstance());
-                } catch (Throwable t) {
-                    subscriber.onError(t);
-                }
-            })
-            .doOnRequest(subscription -> sender.start(initialRequestN))
-            .doOnTerminate(cleanup);
+        return handleStreamResponse(Px.from(payloads)
+                                      .map(payload -> {
+                                   return Frame.Request.from(streamId, FrameType.REQUEST_CHANNEL, payload, 1);
+                               }), streamId);
     }
 
     @Override
@@ -194,6 +148,79 @@ public ClientReactiveSocket start(Consumer<Lease> leaseConsumer) {
         return this;
     }
 
+    private Publisher<Payload> handleRequestResponse(final Publisher<Frame> payload, final int streamId,
+                                                     final int initialRequestN, final boolean sendRequestN) {
+        ConnectableUnicastProcessor<Frame> sender = new ConnectableUnicastProcessor<>();
+
+        synchronized (this) {
+            senders.put(streamId, sender);
+        }
+
+        final Runnable cleanup = () -> {
+            synchronized (this) {
+                receivers.remove(streamId);
+                senders.remove(streamId);
+            }
+        };
+
+        return Px
+                .<Payload>create(subscriber -> {
+                    @SuppressWarnings("rawtypes")
+                    Subscriber raw = subscriber;
+                    @SuppressWarnings("unchecked")
+                    Subscriber<Frame> fs = raw;
+                    synchronized (this) {
+                        receivers.put(streamId, fs);
+                    }
+
+                    payload.subscribe(sender);
+
+                    subscriber.onSubscribe(new Subscription() {
+
+                        @Override
+                        public void request(long n) {
+                            if (sendRequestN) {
+                                sender.onNext(Frame.RequestN.from(streamId, n));
+                            }
+                        }
+
+                        @Override
+                        public void cancel() {
+                            sender.onNext(Frame.Cancel.from(streamId));
+                            sender.cancel();
+                        }
+                    });
+
+                    Px.from(connection.send(sender))
+                      .doOnError(th -> subscriber.onError(th))
+                      .subscribe(DefaultSubscriber.defaultInstance());
+
+                })
+                .doOnRequest(subscription -> sender.start(initialRequestN))
+                .doOnTerminate(cleanup);
+    }
+
+    private Publisher<Payload> handleStreamResponse(Publisher<Frame> request, final int streamId) {
+        RemoteSender sender = new RemoteSender(request, () -> senders.remove(streamId), streamId, 1);
+        Publisher<Frame> src = s -> {
+            CancellableSubscriber<Void> sendSub = doOnError(throwable -> {
+                s.onError(throwable);
+            });
+            ValidatingSubscription<? super Frame> sub = ValidatingSubscription.create(s, () -> {
+                sendSub.cancel();
+            }, requestN -> {
+                transportReceiveSubscription.request(requestN);
+            });
+            connection.send(sender).subscribe(sendSub);
+            s.onSubscribe(sub);
+        };
+
+        RemoteReceiver receiver = new RemoteReceiver(src, connection, streamId, () -> receivers.remove(streamId), true);
+        senders.put(streamId, sender);
+        receivers.put(streamId, receiver);
+        return receiver;
+    }
+
     private void startKeepAlive() {
         keepAliveSendSub = doOnError(errorConsumer);
         connection.send(Px.from(keepAliveProvider.ticks())
@@ -254,7 +281,7 @@ private void handleStreamZero(FrameType type, Frame frame) {
 
     @SuppressWarnings("unchecked")
     private void handleFrame(int streamId, FrameType type, Frame frame) {
-        Subscriber<? super Payload> receiver;
+        Subscriber<Frame> receiver;
         synchronized (this) {
             receiver = receivers.get(streamId);
         }
@@ -270,13 +297,13 @@ private void handleFrame(int streamId, FrameType type, Frame frame) {
                     receiver.onComplete();
                     break;
                 case CANCEL: {
-                    Processor sender;
-                    synchronized (ClientReactiveSocket.this) {
+                    Subscription sender;
+                    synchronized (this) {
                         sender = senders.remove(streamId);
                         receivers.remove(streamId);
                     }
                     if (sender != null) {
-                        ((ConnectableUnicastProcessor) sender).cancel();
+                        sender.cancel();
                     }
                     receiver.onError(new CancelException("cancelling stream id " + streamId));
                     break;
@@ -285,13 +312,13 @@ private void handleFrame(int streamId, FrameType type, Frame frame) {
                     receiver.onNext(frame);
                     break;
                 case REQUEST_N: {
-                    Processor sender;
-                    synchronized (ClientReactiveSocket.this) {
+                    Subscription sender;
+                    synchronized (this) {
                         sender = senders.get(streamId);
                     }
                     if (sender != null) {
                         int n = Frame.RequestN.requestN(frame);
-                        ((ConnectableUnicastProcessor) sender).requestMore(n);
+                        sender.request(n);
                     }
                     break;
                 }

reactivesocket-core/src/main/java/io/reactivesocket/Frame.java

@@ -39,6 +39,9 @@
  * This provides encoding, decoding and field accessors.
  */
 public class Frame implements Payload {
+
+    private static final Logger logger = LoggerFactory.getLogger(Frame.class);
+
     public static final ByteBuffer NULL_BYTEBUFFER = FrameHeaderFlyweight.NULL_BYTEBUFFER;
     public static final int DATA_MTU = 32 * 1024;
     public static final int METADATA_MTU = 32 * 1024;
@@ -55,11 +58,11 @@ public class Frame implements Payload {
         FramePool tmpPool;
 
         try {
-            System.out.println("Creating thread pooled named " + FRAME_POOLER_CLASS_NAME);
+            logger.info("Creating thread pooled named " + FRAME_POOLER_CLASS_NAME);
             tmpPool = (FramePool)Class.forName(FRAME_POOLER_CLASS_NAME).newInstance();
         }
         catch (final Exception ex) {
-            ex.printStackTrace();
+            logger.error("Error initializing frame pool.", ex);
             tmpPool = new UnpooledFrame();
         }
 
@@ -299,7 +302,7 @@ public static String dataMimeType(final Frame frame) {
     }
 
     public static class Error {
-        private static final Logger logger = LoggerFactory.getLogger(Error.class);
+        private static final Logger errorLogger = LoggerFactory.getLogger(Error.class);
 
         private Error() {}
 
@@ -313,8 +316,8 @@ public static Frame from(
             final Frame frame = POOL.acquireFrame(
                 ErrorFrameFlyweight.computeFrameLength(metadata.remaining(), data.remaining()));
 
-            if (logger.isDebugEnabled()) {
-                logger.debug("an error occurred, creating error frame", throwable);
+            if (errorLogger.isDebugEnabled()) {
+                errorLogger.debug("an error occurred, creating error frame", throwable);
             }
 
             frame.length = ErrorFrameFlyweight.encode(

reactivesocket-core/src/main/java/io/reactivesocket/ServerReactiveSocket.java

@@ -175,11 +175,11 @@ private Publisher<Void> handleFrame(Frame frame) {
                 case REQUEST_N:
                     return handleRequestN(streamId, frame);
                 case REQUEST_STREAM:
-                    return handleReceive(streamId, requestStream(frame));
+                    return doReceive(streamId, requestStream(frame));
                 case FIRE_AND_FORGET:
                     return handleFireAndForget(streamId, fireAndForget(frame));
                 case REQUEST_SUBSCRIPTION:
-                    return handleReceive(streamId, requestSubscription(frame));
+                    return doReceive(streamId, requestSubscription(frame));
                 case REQUEST_CHANNEL:
                     return handleChannel(streamId, frame);
                 case RESPONSE:
@@ -288,6 +288,14 @@ private Publisher<Void> handleReceive(int streamId, Publisher<Payload> response)
 
     }
 
+    private Publisher<Void> doReceive(int streamId, Publisher<Payload> response) {
+        Px<Frame> resp = Px.from(response)
+                           .map(payload -> Response.from(streamId, FrameType.RESPONSE, payload));
+        RemoteSender sender = new RemoteSender(resp, () -> subscriptions.remove(streamId), streamId, 2);
+        subscriptions.put(streamId, sender);
+        return connection.send(sender);
+    }
+
     private Publisher<Void> handleChannel(int streamId, Frame firstFrame) {
         int initialRequestN = Request.initialRequestN(firstFrame);
         Frame firstAsNext = Request.from(streamId, FrameType.NEXT, firstFrame, initialRequestN);

reactivesocket-core/src/main/java/io/reactivesocket/internal/RemoteReceiver.java

@@ -53,6 +53,7 @@
  */
 public final class RemoteReceiver implements Processor<Frame, Payload> {
 
+    private final Publisher<Frame> transportSource;
     private final DuplexConnection connection;
     private final int streamId;
     private final Runnable cleanup;
@@ -64,10 +65,22 @@ public final class RemoteReceiver implements Processor<Frame, Payload> {
     private volatile boolean missedComplete;
     private volatile Throwable missedError;
 
+    public RemoteReceiver(Publisher<Frame> transportSource, DuplexConnection connection, int streamId,
+                          Runnable cleanup, boolean sendRequestN) {
+        this.transportSource = transportSource;
+        this.connection = connection;
+        this.streamId = streamId;
+        this.cleanup = cleanup;
+        this.sendRequestN = sendRequestN;
+        requestFrame = null;
+        transportSubscription = null;
+    }
+
     public RemoteReceiver(DuplexConnection connection, int streamId, Runnable cleanup, Frame requestFrame,
                           Subscription transportSubscription, boolean sendRequestN) {
         this.requestFrame = requestFrame;
         this.transportSubscription = transportSubscription;
+        transportSource = null;
         this.connection = connection;
         this.streamId = streamId;
         this.cleanup = cleanup;
@@ -109,7 +122,9 @@ public void subscribe(Subscriber<? super Payload> s) {
             return;
         }
 
-        if (transportSubscription != null) {
+        if (transportSource != null) {
+            transportSource.subscribe(this);
+        } else if (transportSubscription != null) {
             onSubscribe(transportSubscription);
             onNext(requestFrame);
         }

reactivesocket-examples/build.gradle

@@ -45,4 +45,6 @@ dependencies {
     compile 'org.slf4j:slf4j-log4j12:1.7.21'
 
     jmh group: 'org.openjdk.jmh', name: 'jmh-generator-annprocess', version: '1.15'
+
+    compile 'org.slf4j:slf4j-log4j12:1.7.21'
 }