Bugfix 191 (#192)

* Fixed issue with LoopResources no being injected

Author: artem-v

services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/http/HttpGatewayClient.java

@@ -29,18 +29,44 @@ public final class HttpGatewayClient implements GatewayClient {
   private final GatewayClientCodec<ByteBuf> codec;
   private final HttpClient httpClient;
   private final LoopResources loopResources;
+  private final boolean ownsLoopResources;
 
   private final Sinks.One<Void> close = Sinks.one();
   private final Sinks.One<Void> onClose = Sinks.one();
 
   /**
-   * Creates instance of http client transport.
+   * Constructor.
    *
-   * @param settings client settings
+   * @param settings settings
+   * @param codec codec
    */
   public HttpGatewayClient(GatewayClientSettings settings, GatewayClientCodec<ByteBuf> codec) {
+    this(settings, codec, LoopResources.create("http-gateway-client"), true);
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param settings settings
+   * @param codec codec
+   * @param loopResources loopResources
+   */
+  public HttpGatewayClient(
+      GatewayClientSettings settings,
+      GatewayClientCodec<ByteBuf> codec,
+      LoopResources loopResources) {
+    this(settings, codec, loopResources, false);
+  }
+
+  private HttpGatewayClient(
+      GatewayClientSettings settings,
+      GatewayClientCodec<ByteBuf> codec,
+      LoopResources loopResources,
+      boolean ownsLoopResources) {
+
     this.codec = codec;
-    this.loopResources = LoopResources.create("http-gateway-client");
+    this.loopResources = loopResources;
+    this.ownsLoopResources = ownsLoopResources;
 
     HttpClient httpClient =
         HttpClient.create(ConnectionProvider.create("http-gateway-client"))
@@ -111,7 +137,7 @@ public Mono<Void> onClose() {
   }
 
   private Mono<Void> doClose() {
-    return Mono.defer(loopResources::disposeLater);
+    return ownsLoopResources ? Mono.defer(loopResources::disposeLater) : Mono.empty();
   }
 
   private ServiceMessage toMessage(HttpClientResponse httpResponse, ByteBuf content) {

services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/rsocket/RSocketGatewayClient.java

@@ -33,6 +33,7 @@ public final class RSocketGatewayClient implements GatewayClient {
   private final GatewayClientSettings settings;
   private final GatewayClientCodec<Payload> codec;
   private final LoopResources loopResources;
+  private final boolean ownsLoopResources;
 
   private final Sinks.One<Void> close = Sinks.one();
   private final Sinks.One<Void> onClose = Sinks.one();
@@ -41,15 +42,39 @@ public final class RSocketGatewayClient implements GatewayClient {
   private volatile Mono<?> rsocketMono;
 
   /**
-   * Constructor for gateway over rsocket client transport.
+   * Constructor.
    *
-   * @param settings client settings.
-   * @param codec client codec.
+   * @param settings settings
+   * @param codec codec
    */
   public RSocketGatewayClient(GatewayClientSettings settings, GatewayClientCodec<Payload> codec) {
+    this(settings, codec, LoopResources.create("rsocket-gateway-client"), true);
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param settings settings
+   * @param codec codec
+   * @param loopResources loopResources
+   */
+  public RSocketGatewayClient(
+      GatewayClientSettings settings,
+      GatewayClientCodec<Payload> codec,
+      LoopResources loopResources) {
+    this(settings, codec, loopResources, false);
+  }
+
+  private RSocketGatewayClient(
+      GatewayClientSettings settings,
+      GatewayClientCodec<Payload> codec,
+      LoopResources loopResources,
+      boolean ownsLoopResources) {
+
     this.settings = settings;
     this.codec = codec;
-    this.loopResources = LoopResources.create("rsocket-gateway-client");
+    this.loopResources = loopResources;
+    this.ownsLoopResources = ownsLoopResources;
 
     // Setup cleanup
     close
@@ -106,7 +131,7 @@ public Mono<Void> onClose() {
   }
 
   private Mono<Void> doClose() {
-    return Mono.defer(loopResources::disposeLater);
+    return ownsLoopResources ? Mono.defer(loopResources::disposeLater) : Mono.empty();
   }
 
   private Mono<RSocket> getOrConnect() {

services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/websocket/WebsocketGatewayClient.java

@@ -39,6 +39,7 @@ public final class WebsocketGatewayClient implements GatewayClient {
   private final GatewayClientSettings settings;
   private final HttpClient httpClient;
   private final LoopResources loopResources;
+  private final boolean ownsLoopResources;
 
   private final Sinks.One<Void> close = Sinks.one();
   private final Sinks.One<Void> onClose = Sinks.one();
@@ -53,9 +54,33 @@ public final class WebsocketGatewayClient implements GatewayClient {
    * @param codec client codec.
    */
   public WebsocketGatewayClient(GatewayClientSettings settings, GatewayClientCodec<ByteBuf> codec) {
+    this(settings, codec, LoopResources.create("websocket-gateway-client"), true);
+  }
+
+  /**
+   * Creates instance of websocket client transport.
+   *
+   * @param settings client settings
+   * @param codec client codec.
+   * @param loopResources loopResources.
+   */
+  public WebsocketGatewayClient(
+      GatewayClientSettings settings,
+      GatewayClientCodec<ByteBuf> codec,
+      LoopResources loopResources) {
+    this(settings, codec, loopResources, false);
+  }
+
+  private WebsocketGatewayClient(
+      GatewayClientSettings settings,
+      GatewayClientCodec<ByteBuf> codec,
+      LoopResources loopResources,
+      boolean ownsLoopResources) {
+
     this.settings = settings;
     this.codec = codec;
-    this.loopResources = LoopResources.create("websocket-gateway-client");
+    this.loopResources = loopResources;
+    this.ownsLoopResources = ownsLoopResources;
 
     HttpClient httpClient =
         HttpClient.create(ConnectionProvider.newConnection())
@@ -131,7 +156,7 @@ public Mono<Void> onClose() {
   }
 
   private Mono<Void> doClose() {
-    return Mono.defer(loopResources::disposeLater);
+    return ownsLoopResources ? Mono.defer(loopResources::disposeLater) : Mono.empty();
   }
 
   private Mono<WebsocketGatewayClientSession> getOrConnect() {