throwing a TransportException when a connection is closed

Author: robertroeser

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

@@ -19,6 +19,8 @@
 import io.reactivesocket.DuplexConnection;
 import io.reactivesocket.Frame;
 import io.reactivesocket.aeron.internal.Loggable;
+import io.reactivesocket.aeron.internal.NotConnectedException;
+import io.reactivesocket.exceptions.TransportException;
 import io.reactivesocket.rx.Completable;
 import io.reactivesocket.rx.Disposable;
 import io.reactivesocket.rx.Observable;
@@ -101,7 +103,11 @@ public void onNext(Frame frame) {
 
                 @Override
                 public void onError(Throwable t) {
-                    callback.error(t);
+                    if (t instanceof NotConnectedException) {
+                        callback.error(new TransportException(t));
+                    } else {
+                        callback.error(t);
+                    }
                 }
 
                 @Override

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

@@ -28,6 +28,7 @@
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import io.reactivesocket.DuplexConnection;
 import io.reactivesocket.Frame;
+import io.reactivesocket.exceptions.TransportException;
 import io.reactivesocket.rx.Completable;
 import io.reactivesocket.rx.Observable;
 import io.reactivesocket.rx.Observer;
@@ -39,6 +40,7 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.nio.channels.ClosedChannelException;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 public class ClientTcpDuplexConnection implements DuplexConnection {
@@ -121,7 +123,11 @@ public void onNext(Frame frame) {
                     channelFuture.addListener(future -> {
                         Throwable cause = future.cause();
                         if (cause != null) {
-                            callback.error(cause);
+                            if (cause instanceof ClosedChannelException) {
+                                onError(new TransportException(cause));
+                            } else {
+                                onError(cause);
+                            }
                         }
                     });
                 } catch (Throwable t) {

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

@@ -27,6 +27,7 @@
 import io.netty.handler.codec.http.websocketx.*;
 import io.reactivesocket.DuplexConnection;
 import io.reactivesocket.Frame;
+import io.reactivesocket.exceptions.TransportException;
 import io.reactivesocket.rx.Completable;
 import io.reactivesocket.rx.Observable;
 import io.reactivesocket.rx.Observer;
@@ -39,6 +40,7 @@
 import java.net.SocketAddress;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.nio.channels.ClosedChannelException;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 public class ClientWebSocketDuplexConnection implements DuplexConnection {
@@ -136,7 +138,11 @@ public void onNext(Frame frame) {
                     channelFuture.addListener(future -> {
                         Throwable cause = future.cause();
                         if (cause != null) {
-                            callback.error(cause);
+                            if (cause instanceof ClosedChannelException) {
+                                onError(new TransportException(cause));
+                            } else {
+                                onError(cause);
+                            }
                         }
                     });
                 } catch (Throwable t) {

reactivesocket-netty/src/test/java/io/reactivesocket/netty/tcp/ClientServerTest.java

@@ -52,7 +52,7 @@ public class ClientServerTest {
     static EventLoopGroup bossGroup = new NioEventLoopGroup(1);
     static EventLoopGroup workerGroup = new NioEventLoopGroup(4);
 
-    static ReactiveSocketServerHandler serverHandler = ReactiveSocketServerHandler.create(setupPayload ->
+    static ReactiveSocketServerHandler serverHandler = ReactiveSocketServerHandler.create((setupPayload, rs) ->
         new RequestHandler() {
             @Override
             public Publisher<Payload> handleRequestResponse(Payload payload) {

reactivesocket-netty/src/test/java/io/reactivesocket/netty/tcp/Ping.java

@@ -80,11 +80,13 @@ public ByteBuffer getMetadata() {
                     .toObservable(
                         reactiveSocket
                             .requestResponse(keyPayload))
+                    .doOnError(t -> t.printStackTrace())
                     .doOnNext(s -> {
                         long diff = System.nanoTime() - start;
                         histogram.recordValue(diff);
                     });
             }, 16)
+            .doOnError(t -> t.printStackTrace())
             .subscribe(new Subscriber<Payload>() {
                 @Override
                 public void onCompleted() {

reactivesocket-netty/src/test/java/io/reactivesocket/netty/tcp/Pong.java

@@ -43,7 +43,7 @@ public static void main(String... args) throws Exception {
         r.nextBytes(response);
 
         ReactiveSocketServerHandler serverHandler =
-            ReactiveSocketServerHandler.create(setupPayload -> new RequestHandler() {
+            ReactiveSocketServerHandler.create((setupPayload, rs) -> new RequestHandler() {
                 @Override
                 public Publisher<Payload> handleRequestResponse(Payload payload) {
                     return new Publisher<Payload>() {

reactivesocket-netty/src/test/java/io/reactivesocket/netty/websocket/ClientServerTest.java

@@ -51,7 +51,7 @@ public class ClientServerTest {
     static EventLoopGroup bossGroup = new NioEventLoopGroup(1);
     static EventLoopGroup workerGroup = new NioEventLoopGroup(4);
 
-    static ReactiveSocketServerHandler serverHandler = ReactiveSocketServerHandler.create(setupPayload ->
+    static ReactiveSocketServerHandler serverHandler = ReactiveSocketServerHandler.create((setupPayload, rs) ->
         new RequestHandler() {
             @Override
             public Publisher<Payload> handleRequestResponse(Payload payload) {

reactivesocket-netty/src/test/java/io/reactivesocket/netty/websocket/Pong.java

@@ -46,7 +46,7 @@ public static void main(String... args) throws Exception {
         r.nextBytes(response);
 
         ReactiveSocketServerHandler serverHandler =
-            ReactiveSocketServerHandler.create(setupPayload -> new RequestHandler() {
+            ReactiveSocketServerHandler.create((setupPayload, rs) -> new RequestHandler() {
                 @Override
                 public Publisher<Payload> handleRequestResponse(Payload payload) {
                     return new Publisher<Payload>() {