Removed ConnectionClosedException

Author: artem-v

services-api/src/main/java/io/scalecube/services/exceptions/ConnectionClosedException.java

@@ -4,19 +4,17 @@
 import io.rsocket.RSocket;
 import io.rsocket.util.ByteBufPayload;
 import io.scalecube.services.api.ServiceMessage;
-import io.scalecube.services.exceptions.ConnectionClosedException;
 import io.scalecube.services.transport.api.ClientChannel;
 import io.scalecube.services.transport.api.ServiceMessageCodec;
 import java.lang.reflect.Type;
-import java.nio.channels.ClosedChannelException;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 public class RSocketClientChannel implements ClientChannel {
 
-  private Mono<RSocket> rsocket;
-  private ServiceMessageCodec messageCodec;
+  private final Mono<RSocket> rsocket;
+  private final ServiceMessageCodec messageCodec;
 
   public RSocketClientChannel(Mono<RSocket> rsocket, ServiceMessageCodec codec) {
     this.rsocket = rsocket;
@@ -26,27 +24,15 @@ public RSocketClientChannel(Mono<RSocket> rsocket, ServiceMessageCodec codec) {
   @Override
   public Mono<ServiceMessage> requestResponse(ServiceMessage message, Type responseType) {
     return rsocket
-        .flatMap(
-            rsocket ->
-                rsocket
-                    .requestResponse(toPayload(message))
-                    .onErrorMap(
-                        ClosedChannelException.class,
-                        e -> new ConnectionClosedException("Connection closed")))
+        .flatMap(rsocket -> rsocket.requestResponse(toPayload(message)))
         .map(this::toMessage)
         .map(msg -> ServiceMessageCodec.decodeData(msg, responseType));
   }
 
   @Override
   public Flux<ServiceMessage> requestStream(ServiceMessage message, Type responseType) {
     return rsocket
-        .flatMapMany(
-            rsocket ->
-                rsocket
-                    .requestStream(toPayload(message))
-                    .onErrorMap(
-                        ClosedChannelException.class,
-                        e -> new ConnectionClosedException("Connection closed")))
+        .flatMapMany(rsocket -> rsocket.requestStream(toPayload(message)))
         .map(this::toMessage)
         .map(msg -> ServiceMessageCodec.decodeData(msg, responseType));
   }
@@ -55,13 +41,7 @@ public Flux<ServiceMessage> requestStream(ServiceMessage message, Type responseT
   public Flux<ServiceMessage> requestChannel(
       Publisher<ServiceMessage> publisher, Type responseType) {
     return rsocket
-        .flatMapMany(
-            rsocket ->
-                rsocket
-                    .requestChannel(Flux.from(publisher).map(this::toPayload))
-                    .onErrorMap(
-                        ClosedChannelException.class,
-                        e -> new ConnectionClosedException("Connection closed")))
+        .flatMapMany(rsocket -> rsocket.requestChannel(Flux.from(publisher).map(this::toPayload)))
         .map(this::toMessage)
         .map(msg -> ServiceMessageCodec.decodeData(msg, responseType));
   }

services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketClientChannel.java

@@ -10,9 +10,9 @@
 import io.scalecube.services.api.ServiceMessage;
 import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
 import io.scalecube.services.discovery.api.ServiceDiscoveryEvent;
-import io.scalecube.services.exceptions.ConnectionClosedException;
 import io.scalecube.services.sut.QuoteService;
 import io.scalecube.services.sut.SimpleQuoteService;
+import java.nio.channels.ClosedChannelException;
 import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
@@ -94,7 +94,7 @@ public void test_remote_node_died_mono_never() throws Exception {
     TimeUnit.MILLISECONDS.sleep(100);
 
     assertEquals(0, latch1.getCount());
-    assertEquals(ConnectionClosedException.class, exceptionHolder.get().getClass());
+    assertEquals(ClosedChannelException.class, exceptionHolder.get().getClass());
     assertTrue(sub1.get().isDisposed());
   }
 
@@ -122,7 +122,7 @@ public void test_remote_node_died_many_never() throws Exception {
     TimeUnit.MILLISECONDS.sleep(100);
 
     assertEquals(0, latch1.getCount());
-    assertEquals(ConnectionClosedException.class, exceptionHolder.get().getClass());
+    assertEquals(ClosedChannelException.class, exceptionHolder.get().getClass());
     assertTrue(sub1.get().isDisposed());
   }
 
@@ -154,7 +154,7 @@ public void test_remote_node_died_many_then_never() throws Exception {
     TimeUnit.MILLISECONDS.sleep(100);
 
     assertEquals(0, latch1.getCount());
-    assertEquals(ConnectionClosedException.class, exceptionHolder.get().getClass());
+    assertEquals(ClosedChannelException.class, exceptionHolder.get().getClass());
     assertTrue(sub1.get().isDisposed());
   }
 }