Renamings, fixed WebsocketGatewayAcceptor (added filtering of reactor…

….netty's AbortedException)
scalecube · Apr 6, 2021 · e16bd15 · e16bd15
1 parent 1e1e463
commit e16bd15
Show file tree

Hide file tree

Showing 8 changed files with 29 additions and 18 deletions.
diff --git a/...ateway-netty/src/main/java/io/scalecube/services/gateway/ws/WebsocketGatewayAcceptor.java b/...ateway-netty/src/main/java/io/scalecube/services/gateway/ws/WebsocketGatewayAcceptor.java
@@ -10,6 +10,7 @@
 import static io.scalecube.services.gateway.ws.GatewayMessages.validateSidOnSession;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.scalecube.services.ServiceCall;
 import io.scalecube.services.api.ServiceMessage;
@@ -35,6 +36,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.DisposableChannel;
+import reactor.netty.channel.AbortedException;
 import reactor.netty.http.server.HttpServerRequest;
 import reactor.netty.http.server.HttpServerResponse;
 import reactor.netty.http.websocket.WebsocketInbound;
@@ -127,16 +129,25 @@ private Mono<Void> onConnect(WebsocketGatewaySession session) {
 
     session
         .receive()
-        .doOnError(th -> gatewayHandler.onSessionError(session, th))
         .subscribe(
             byteBuf -> {
+              if (byteBuf == Unpooled.EMPTY_BUFFER) {
+                return;
+              }
+
               if (!byteBuf.isReadable()) {
                 ReferenceCountUtil.safestRelease(byteBuf);
                 return;
               }
+
               Mono.deferContextual(context -> onRequest(session, byteBuf, (Context) context))
                   .contextWrite(context -> gatewayHandler.onRequest(session, byteBuf, context))
                   .subscribe();
+            },
+            th -> {
+              if (!(th instanceof AbortedException)) {
+                gatewayHandler.onSessionError(session, th);
+              }
             });
 
     return session.onClose(() -> gatewayHandler.onSessionClose(session));

diff --git a/.../rsocket/RsocketClientConnectionTest.java → .../rsocket/RSocketClientConnectionTest.java b/.../rsocket/RsocketClientConnectionTest.java → .../rsocket/RSocketClientConnectionTest.java
@@ -35,7 +35,7 @@
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
-class RsocketClientConnectionTest extends BaseTest {
+class RSocketClientConnectionTest extends BaseTest {
 
   public static final GatewayClientCodec<Payload> CLIENT_CODEC =
       GatewayClientTransports.RSOCKET_CLIENT_CODEC;

diff --git a/...rsocket/RsocketClientErrorMapperTest.java → ...rsocket/RSocketClientErrorMapperTest.java b/...rsocket/RsocketClientErrorMapperTest.java → ...rsocket/RSocketClientErrorMapperTest.java
@@ -13,11 +13,11 @@
 import org.junit.jupiter.api.extension.RegisterExtension;
 import reactor.test.StepVerifier;
 
-class RsocketClientErrorMapperTest extends BaseTest {
+class RSocketClientErrorMapperTest extends BaseTest {
 
   @RegisterExtension
-  static RsocketGatewayExtension extension =
-      new RsocketGatewayExtension(
+  static RSocketGatewayExtension extension =
+      new RSocketGatewayExtension(
           ServiceInfo.fromServiceInstance(new ErrorServiceImpl())
               .errorMapper(ERROR_MAPPER)
               .build());

diff --git a/...eway/rsocket/RsocketGatewayExtension.java → ...eway/rsocket/RSocketGatewayExtension.java b/...eway/rsocket/RsocketGatewayExtension.java → ...eway/rsocket/RSocketGatewayExtension.java
@@ -4,15 +4,15 @@
 import io.scalecube.services.gateway.AbstractGatewayExtension;
 import io.scalecube.services.gateway.transport.GatewayClientTransports;
 
-class RsocketGatewayExtension extends AbstractGatewayExtension {
+class RSocketGatewayExtension extends AbstractGatewayExtension {
 
   private static final String GATEWAY_ALIAS_NAME = "rsws";
 
-  RsocketGatewayExtension(Object serviceInstance) {
+  RSocketGatewayExtension(Object serviceInstance) {
     this(ServiceInfo.fromServiceInstance(serviceInstance).build());
   }
 
-  RsocketGatewayExtension(ServiceInfo serviceInfo) {
+  RSocketGatewayExtension(ServiceInfo serviceInfo) {
     super(
         serviceInfo,
         opts -> new RSocketGateway(opts.id(GATEWAY_ALIAS_NAME)),

diff --git a/...gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketGatewayTest.java b/...gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketGatewayTest.java
@@ -26,7 +26,7 @@ class RSocketGatewayTest extends BaseTest {
   private static final Duration TIMEOUT = Duration.ofSeconds(3);
 
   @RegisterExtension
-  static RsocketGatewayExtension extension = new RsocketGatewayExtension(new GreetingServiceImpl());
+  static RSocketGatewayExtension extension = new RSocketGatewayExtension(new GreetingServiceImpl());
 
   private GreetingService service;
 

diff --git a/...t/RsocketLocalGatewayErrorMapperTest.java → ...t/RSocketLocalGatewayErrorMapperTest.java b/...t/RsocketLocalGatewayErrorMapperTest.java → ...t/RSocketLocalGatewayErrorMapperTest.java
@@ -13,11 +13,11 @@
 import org.junit.jupiter.api.extension.RegisterExtension;
 import reactor.test.StepVerifier;
 
-class RsocketLocalGatewayErrorMapperTest extends BaseTest {
+class RSocketLocalGatewayErrorMapperTest extends BaseTest {
 
   @RegisterExtension
-  static RsocketLocalGatewayExtension extension =
-      new RsocketLocalGatewayExtension(
+  static RSocketLocalGatewayExtension extension =
+      new RSocketLocalGatewayExtension(
           ServiceInfo.fromServiceInstance(new ErrorServiceImpl()).errorMapper(ERROR_MAPPER).build(),
           opts ->
               new RSocketGateway(opts.call(opts.call().errorMapper(ERROR_MAPPER)), ERROR_MAPPER));

diff --git a/...rsocket/RsocketLocalGatewayExtension.java → ...rsocket/RSocketLocalGatewayExtension.java b/...rsocket/RsocketLocalGatewayExtension.java → ...rsocket/RSocketLocalGatewayExtension.java
@@ -6,19 +6,19 @@
 import io.scalecube.services.gateway.transport.GatewayClientTransports;
 import java.util.function.Function;
 
-class RsocketLocalGatewayExtension extends AbstractLocalGatewayExtension {
+class RSocketLocalGatewayExtension extends AbstractLocalGatewayExtension {
 
   private static final String GATEWAY_ALIAS_NAME = "rsws";
 
-  RsocketLocalGatewayExtension(Object serviceInstance) {
+  RSocketLocalGatewayExtension(Object serviceInstance) {
     this(ServiceInfo.fromServiceInstance(serviceInstance).build());
   }
 
-  RsocketLocalGatewayExtension(ServiceInfo serviceInfo) {
+  RSocketLocalGatewayExtension(ServiceInfo serviceInfo) {
     this(serviceInfo, RSocketGateway::new);
   }
 
-  RsocketLocalGatewayExtension(
+  RSocketLocalGatewayExtension(
       ServiceInfo serviceInfo, Function<GatewayOptions, RSocketGateway> gatewaySupplier) {
     super(
         serviceInfo,

diff --git a/...ay-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketLocalGatewayTest.java b/...ay-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketLocalGatewayTest.java
@@ -25,8 +25,8 @@ class RSocketLocalGatewayTest extends BaseTest {
   private static final Duration TIMEOUT = Duration.ofSeconds(3);
 
   @RegisterExtension
-  static RsocketLocalGatewayExtension extension =
-      new RsocketLocalGatewayExtension(new GreetingServiceImpl());
+  static RSocketLocalGatewayExtension extension =
+      new RSocketLocalGatewayExtension(new GreetingServiceImpl());
 
   private GreetingService service;