Merge pull request #17 from ReactiveSocket/stevegury/propagate-cancellation

Propagate cancellation when a TransportException occurs

Author: robertroeser

reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/client/AeronClientDuplexConnection.java

@@ -79,22 +79,22 @@ public void dispose() {
     public void addOutput(Publisher<Frame> o, Completable callback) {
         o
             .subscribe(new Subscriber<Frame>() {
-                private Subscription s;
+                private Subscription subscription;
 
                 @Override
                 public void onSubscribe(Subscription s) {
-                    this.s = s;
+                    this.subscription = s;
                     s.request(128);
 
                 }
 
                 @Override
                 public void onNext(Frame frame) {
                     if (isTraceEnabled()) {
-                        trace("onNext subscription => {} and frame => {}", s.toString(), frame.toString());
+                        trace("onNext subscription => {} and frame => {}", subscription.toString(), frame.toString());
                     }
 
-                    final FrameHolder fh = FrameHolder.get(frame, publication, s);
+                    final FrameHolder fh = FrameHolder.get(frame, publication, subscription);
                     boolean offer;
                     do {
                         offer = frameSendQueue.offer(fh);
@@ -105,6 +105,7 @@ public void onNext(Frame frame) {
                 public void onError(Throwable t) {
                     if (t instanceof NotConnectedException) {
                         callback.error(new TransportException(t));
+                        subscription.cancel();
                     } else {
                         callback.error(t);
                     }

reactivesocket-netty/src/main/java/io/reactivesocket/netty/tcp/client/ClientTcpDuplexConnection.java

@@ -98,8 +98,11 @@ public final Observable<Frame> getInput() {
     @Override
     public void addOutput(Publisher<Frame> o, Completable callback) {
         o.subscribe(new Subscriber<Frame>() {
+            private Subscription subscription;
+
             @Override
             public void onSubscribe(Subscription s) {
+                subscription = s;
                 s.request(Long.MAX_VALUE);
             }
 
@@ -126,6 +129,9 @@ public void onNext(Frame frame) {
             @Override
             public void onError(Throwable t) {
                 callback.error(t);
+                if (t instanceof TransportException) {
+                    subscription.cancel();
+                }
             }
 
             @Override

reactivesocket-netty/src/main/java/io/reactivesocket/netty/websocket/client/ClientWebSocketDuplexConnection.java

@@ -119,8 +119,11 @@ public final Observable<Frame> getInput() {
     @Override
     public void addOutput(Publisher<Frame> o, Completable callback) {
         o.subscribe(new Subscriber<Frame>() {
+            private Subscription subscription;
+
             @Override
             public void onSubscribe(Subscription s) {
+                subscription = s;
                 s.request(Long.MAX_VALUE);
             }
 
@@ -148,6 +151,9 @@ public void onNext(Frame frame) {
             @Override
             public void onError(Throwable t) {
                 callback.error(t);
+                if (t instanceof TransportException) {
+                    subscription.cancel();
+                }
             }
 
             @Override