Added .handleConnectionReset at RSocketClientChannel

Author: artem-v

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

@@ -8,11 +8,16 @@
 import io.scalecube.services.transport.api.ServiceMessageCodec;
 import java.lang.reflect.Type;
 import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.netty.channel.AbortedException;
 
 public class RSocketClientChannel implements ClientChannel {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(RSocketClientChannel.class);
+
   private final Mono<RSocket> rsocket;
   private final ServiceMessageCodec messageCodec;
 
@@ -26,15 +31,17 @@ public Mono<ServiceMessage> requestResponse(ServiceMessage message, Type respons
     return rsocket
         .flatMap(rsocket -> rsocket.requestResponse(toPayload(message)))
         .map(this::toMessage)
-        .map(msg -> ServiceMessageCodec.decodeData(msg, responseType));
+        .map(msg -> ServiceMessageCodec.decodeData(msg, responseType))
+        .doOnError(RSocketClientChannel::handleConnectionReset);
   }
 
   @Override
   public Flux<ServiceMessage> requestStream(ServiceMessage message, Type responseType) {
     return rsocket
         .flatMapMany(rsocket -> rsocket.requestStream(toPayload(message)))
         .map(this::toMessage)
-        .map(msg -> ServiceMessageCodec.decodeData(msg, responseType));
+        .map(msg -> ServiceMessageCodec.decodeData(msg, responseType))
+        .doOnError(RSocketClientChannel::handleConnectionReset);
   }
 
   @Override
@@ -43,7 +50,8 @@ public Flux<ServiceMessage> requestChannel(
     return rsocket
         .flatMapMany(rsocket -> rsocket.requestChannel(Flux.from(publisher).map(this::toPayload)))
         .map(this::toMessage)
-        .map(msg -> ServiceMessageCodec.decodeData(msg, responseType));
+        .map(msg -> ServiceMessageCodec.decodeData(msg, responseType))
+        .doOnError(RSocketClientChannel::handleConnectionReset);
   }
 
   private Payload toPayload(ServiceMessage request) {
@@ -57,4 +65,12 @@ private ServiceMessage toMessage(Payload payload) {
       payload.release();
     }
   }
+
+  private static void handleConnectionReset(Throwable throwable) {
+    if (AbortedException.isConnectionReset(throwable)) {
+      if (LOGGER.isDebugEnabled()) {
+        LOGGER.debug("[requestResponse] Connection has been reset");
+      }
+    }
+  }
 }