Merge pull request #5 from robertroeser/master

throwing a TransportException when a connection is closed

Author: rdegnan

reactivesocket-aeron/src/examples/java/io/reactivesocket/aeron/example/fireandforget/Forget.java

@@ -18,6 +18,7 @@
 import io.reactivesocket.ConnectionSetupHandler;
 import io.reactivesocket.ConnectionSetupPayload;
 import io.reactivesocket.Payload;
+import io.reactivesocket.ReactiveSocket;
 import io.reactivesocket.RequestHandler;
 import io.reactivesocket.aeron.server.ReactiveSocketAeronServer;
 import io.reactivesocket.exceptions.SetupException;
@@ -33,7 +34,7 @@ public static void main(String... args) {
 
         ReactiveSocketAeronServer server = ReactiveSocketAeronServer.create(host, 39790, new ConnectionSetupHandler() {
             @Override
-            public RequestHandler apply(ConnectionSetupPayload setupPayload) throws SetupException {
+            public RequestHandler apply(ConnectionSetupPayload setupPayload, ReactiveSocket rs) throws SetupException {
                 return new RequestHandler() {
                     @Override
                     public Publisher<Payload> handleRequestResponse(Payload payload) {

reactivesocket-aeron/src/examples/java/io/reactivesocket/aeron/example/requestreply/Pong.java

@@ -18,6 +18,7 @@
 import io.reactivesocket.ConnectionSetupHandler;
 import io.reactivesocket.ConnectionSetupPayload;
 import io.reactivesocket.Payload;
+import io.reactivesocket.ReactiveSocket;
 import io.reactivesocket.RequestHandler;
 import io.reactivesocket.aeron.server.ReactiveSocketAeronServer;
 import io.reactivesocket.exceptions.SetupException;
@@ -44,7 +45,7 @@ public static void main(String... args) {
 
         ReactiveSocketAeronServer server = ReactiveSocketAeronServer.create(host, 39790, new ConnectionSetupHandler() {
             @Override
-            public RequestHandler apply(ConnectionSetupPayload setupPayload) throws SetupException {
+            public RequestHandler apply(ConnectionSetupPayload setupPayload, ReactiveSocket rs) throws SetupException {
                 return new RequestHandler() {
                     @Override
                     public Publisher<Payload> handleRequestResponse(Payload payload) {

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-aeron/src/test/java/io/reactivesocket/aeron/client/ReactiveSocketAeronTest.java

@@ -112,7 +112,7 @@ public void requestResponseN(int count) throws Exception {
         AtomicLong counter = new AtomicLong();
         ReactiveSocketAeronServer.create(new ConnectionSetupHandler() {
             @Override
-            public RequestHandler apply(ConnectionSetupPayload setupPayload) throws SetupException {
+            public RequestHandler apply(ConnectionSetupPayload setupPayload, ReactiveSocket rs) throws SetupException {
                 return new RequestHandler.Builder()
                   .withRequestResponse(new Function<Payload, Publisher<Payload>>() {
                       Frame frame = Frame.from(ByteBuffer.allocate(1));
@@ -194,7 +194,7 @@ public void onNext(Payload payload) {
     }
 
     public void requestStreamN(int count) throws Exception {
-        ReactiveSocketAeronServer.create(setupPayload ->
+        ReactiveSocketAeronServer.create((setupPayload, rs) ->
           new RequestHandler.Builder()
             .withRequestStream(payload -> {
                 ByteBuffer data = payload.getData();
@@ -264,7 +264,7 @@ public void testReconnection() throws Exception {
 
         ReactiveSocketAeronServer server = ReactiveSocketAeronServer.create(new ConnectionSetupHandler() {
             @Override
-            public RequestHandler apply(ConnectionSetupPayload setupPayload) throws SetupException {
+            public RequestHandler apply(ConnectionSetupPayload setupPayload, ReactiveSocket rs) throws SetupException {
                 return new RequestHandler() {
                     Frame frame = Frame.from(ByteBuffer.allocate(1));
 

reactivesocket-jsr-356/src/test/java/io/reactivesocket/javax/websocket/ClientServerEndpoint.java

@@ -25,7 +25,7 @@
 
 public class ClientServerEndpoint extends ReactiveSocketWebSocketServer {
     public ClientServerEndpoint() {
-        super(setupPayload -> new RequestHandler() {
+        super((setupPayload, rs) -> new RequestHandler() {
             @Override
             public Publisher<Payload> handleRequestResponse(Payload payload) {
                 return s -> {

reactivesocket-jsr-356/src/test/java/io/reactivesocket/javax/websocket/PongEndpoint.java

@@ -34,7 +34,7 @@ public class PongEndpoint extends ReactiveSocketWebSocketServer {
     }
 
     public PongEndpoint() {
-        super(setupPayload -> new RequestHandler() {
+        super((setupPayload, rs) -> new RequestHandler() {
             @Override
             public Publisher<Payload> handleRequestResponse(Payload payload) {
                 return new Publisher<Payload>() {

reactivesocket-local/src/test/java/io/reactivesocket/local/ClientServerTest.java

@@ -41,7 +41,7 @@ public class ClientServerTest {
     public static void setup() throws Exception {
         server = LocalServerReactiveSocketFactory.INSTANCE.callAndWait(new LocalServerReactiveSocketFactory.Config("test", new ConnectionSetupHandler() {
             @Override
-            public RequestHandler apply(ConnectionSetupPayload setupPayload) throws SetupException {
+            public RequestHandler apply(ConnectionSetupPayload setupPayload, ReactiveSocket rs) throws SetupException {
                 return new RequestHandler() {
                     @Override
                     public Publisher<Payload> handleRequestResponse(Payload payload) {

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) {