Propagate cancellation when a TransportException occurs

Problem
When a TransportException happens on the transport, it is propagated up
in the chain but the Susbscription is not cancelled. Then, any source
associated with the chain continue to emit events.

One example of that is the Keep-Alive Observable, which regularly emit
keep-alive messages at fixed interval, when the connection is closed, the
observable has to be cancelled.

Solution
In all DuplexConnection implementation, cancel the Subscription when we
saw a TransportException.

Author: stevegury

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