Enhanced .close() in WebsocketGatewaClientTransport

artem-v · web-flow · commit 5ddb28ef1cf1 · 2025-03-24T10:54:52.000+02:00
diff --git a/services-gateway/src/main/java/io/scalecube/services/gateway/client/websocket/WebsocketGatewayClientTransport.java b/services-gateway/src/main/java/io/scalecube/services/gateway/client/websocket/WebsocketGatewayClientTransport.java
@@ -4,6 +4,7 @@
 import io.netty.channel.ChannelOption;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
+import io.netty.resolver.DefaultAddressResolverGroup;
 import io.scalecube.services.Address;
 import io.scalecube.services.ServiceReference;
 import io.scalecube.services.api.ServiceMessage;
@@ -26,7 +27,6 @@
 import reactor.netty.Connection;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.resources.ConnectionProvider;
-import reactor.netty.resources.LoopResources;
 
 public final class WebsocketGatewayClientTransport implements ClientChannel, ClientTransport {
 
@@ -39,10 +39,8 @@ public final class WebsocketGatewayClientTransport implements ClientChannel, Cli
   private static final int CONNECT_TIMEOUT_MILLIS = (int) Duration.ofSeconds(5).toMillis();
 
   private final GatewayClientCodec clientCodec;
-  private final LoopResources loopResources;
   private final Duration keepAliveInterval;
   private final Function<HttpClient, HttpClient> operator;
-  private final boolean ownsLoopResources;
 
   private final AtomicLong sidCounter = new AtomicLong();
   private final AtomicReference<WebsocketGatewayClientSession> clientSessionReference =
@@ -52,11 +50,6 @@ private WebsocketGatewayClientTransport(Builder builder) {
     this.clientCodec = builder.clientCodec;
     this.keepAliveInterval = builder.keepAliveInterval;
     this.operator = builder.operator;
-    this.loopResources =
-        builder.loopResources == null
-            ? LoopResources.create("websocket-gateway-client", 1, true)
-            : builder.loopResources;
-    this.ownsLoopResources = builder.loopResources == null;
   }
 
   @Override
@@ -70,7 +63,7 @@ public ClientChannel create(ServiceReference serviceReference) {
           final HttpClient httpClient =
               operator.apply(
                   HttpClient.create(ConnectionProvider.newConnection())
-                      .runOn(loopResources)
+                      .resolver(DefaultAddressResolverGroup.INSTANCE)
                       .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, CONNECT_TIMEOUT_MILLIS)
                       .option(ChannelOption.TCP_NODELAY, true)
                       .headers(headers -> headers.set(HttpHeaderNames.CONTENT_TYPE, CONTENT_TYPE)));
@@ -192,15 +185,15 @@ private static Throwable getRootCause(Throwable throwable) {
 
   @Override
   public void close() {
-    if (ownsLoopResources) {
-      loopResources.dispose();
+    final var session = clientSessionReference.get();
+    if (session != null) {
+      session.close().doOnError(ex -> {}).subscribe();
     }
   }
 
   public static class Builder {
 
     private GatewayClientCodec clientCodec = CLIENT_CODEC;
-    private LoopResources loopResources;
     private Duration keepAliveInterval = Duration.ZERO;
     private Function<HttpClient, HttpClient> operator = client -> client;
 
@@ -211,11 +204,6 @@ public Builder clientCodec(GatewayClientCodec clientCodec) {
       return this;
     }
 
-    public Builder loopResources(LoopResources loopResources) {
-      this.loopResources = loopResources;
-      return this;
-    }
-
     public Builder httpClient(UnaryOperator<HttpClient> operator) {
       this.operator = this.operator.andThen(operator);
       return this;
