Added RSocketClientTransportFactory and RSocketClientTransportFactory

Author: artem-v

services-examples-parent/services-examples-runner/src/main/java/io/scalecube/services/examples/ExamplesRunner.java

@@ -1,5 +1,7 @@
 package io.scalecube.services.examples;
 
+import io.rsocket.transport.netty.client.TcpClientTransport;
+import io.rsocket.transport.netty.server.TcpServerTransport;
 import io.scalecube.config.ConfigRegistry;
 import io.scalecube.config.ConfigRegistrySettings;
 import io.scalecube.config.audit.Slf4JConfigEventListener;
@@ -58,20 +60,26 @@ public static void main(String[] args) {
             .transport(
                 () ->
                     new RSocketServiceTransport()
-                        .tcpClient(
+                        .clientTransportFactory(
                             loopResources ->
-                                TcpClient.newConnection()
-                                    .runOn(loopResources)
-                                    .wiretap(false)
-                                    .noProxy()
-                                    .noSSL())
-                        .tcpServer(
+                                address ->
+                                    TcpClientTransport.create(
+                                        TcpClient.newConnection()
+                                            .host(address.host())
+                                            .port(address.port())
+                                            .runOn(loopResources)
+                                            .wiretap(false)
+                                            .noProxy()
+                                            .noSSL()))
+                        .serverTransportFactory(
                             loopResources ->
-                                TcpServer.create()
-                                    .wiretap(false)
-                                    .port(config.servicePort())
-                                    .runOn(loopResources)
-                                    .noSSL()))
+                                () ->
+                                    TcpServerTransport.create(
+                                        TcpServer.create()
+                                            .wiretap(false)
+                                            .port(config.servicePort())
+                                            .runOn(loopResources)
+                                            .noSSL())))
             .services(new BenchmarkServiceImpl(), new GreetingServiceImpl())
             .startAwait();
 

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

@@ -3,7 +3,6 @@
 import io.rsocket.RSocket;
 import io.rsocket.core.RSocketConnector;
 import io.rsocket.frame.decoder.PayloadDecoder;
-import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.scalecube.net.Address;
 import io.scalecube.services.transport.api.ClientChannel;
 import io.scalecube.services.transport.api.ClientTransport;
@@ -14,7 +13,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
-import reactor.netty.tcp.TcpClient;
 
 public class RSocketClientTransport implements ClientTransport {
 
@@ -24,17 +22,18 @@ public class RSocketClientTransport implements ClientTransport {
       ThreadLocal.withInitial(ConcurrentHashMap::new);
 
   private final ServiceMessageCodec messageCodec;
-  private final TcpClient tcpClient;
+  private final RSocketClientTransportFactory clientTransportFactory;
 
   /**
    * Constructor for this transport.
    *
-   * @param messageCodec message codec
-   * @param tcpClient tcp client
+   * @param messageCodec messageCodec
+   * @param clientTransportFactory clientTransportFactory
    */
-  public RSocketClientTransport(ServiceMessageCodec messageCodec, TcpClient tcpClient) {
+  public RSocketClientTransport(
+      ServiceMessageCodec messageCodec, RSocketClientTransportFactory clientTransportFactory) {
     this.messageCodec = messageCodec;
-    this.tcpClient = tcpClient;
+    this.clientTransportFactory = clientTransportFactory;
   }
 
   @Override
@@ -46,10 +45,9 @@ public ClientChannel create(Address address) {
   }
 
   private Mono<RSocket> connect(Address address, Map<Address, Mono<RSocket>> monoMap) {
-    TcpClient tcpClient = this.tcpClient.host(address.host()).port(address.port());
     return RSocketConnector.create()
         .payloadDecoder(PayloadDecoder.DEFAULT)
-        .connect(() -> TcpClientTransport.create(tcpClient))
+        .connect(() -> clientTransportFactory.clientTransport(address))
         .doOnSuccess(
             rsocket -> {
               LOGGER.debug("[rsocket][client] Connected successfully on {}", address);
@@ -80,7 +78,6 @@ private Mono<RSocket> connect(Address address, Map<Address, Mono<RSocket>> monoM
   public String toString() {
     return new StringJoiner(", ", RSocketClientTransport.class.getSimpleName() + "[", "]")
         .add("messageCodec=" + messageCodec)
-        .add("tcpClient=" + tcpClient)
         .toString();
   }
 }

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

@@ -0,0 +1,9 @@
+package io.scalecube.services.transport.rsocket;
+
+import io.rsocket.transport.ClientTransport;
+import io.scalecube.net.Address;
+
+public interface RSocketClientTransportFactory {
+
+  ClientTransport clientTransport(Address address);
+}

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

@@ -3,7 +3,6 @@
 import io.rsocket.core.RSocketServer;
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.transport.netty.server.CloseableChannel;
-import io.rsocket.transport.netty.server.TcpServerTransport;
 import io.scalecube.net.Address;
 import io.scalecube.services.methods.ServiceMethodRegistry;
 import io.scalecube.services.transport.api.ServerTransport;
@@ -13,27 +12,26 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
-import reactor.netty.tcp.TcpServer;
 
-/** RSocket server transport implementation. */
 public class RSocketServerTransport implements ServerTransport {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(RSocketServerTransport.class);
 
   private final ServiceMessageCodec messageCodec;
-  private final TcpServer tcpServer;
+  private final RSocketServerTransportFactory serverTransportFactory;
 
   private CloseableChannel serverChannel; // calculated
 
   /**
    * Constructor for this server transport.
    *
-   * @param messageCodec message codec
-   * @param tcpServer tcp server
+   * @param messageCodec messageCodec
+   * @param serverTransportFactory serverTransportFactory
    */
-  public RSocketServerTransport(ServiceMessageCodec messageCodec, TcpServer tcpServer) {
+  public RSocketServerTransport(
+      ServiceMessageCodec messageCodec, RSocketServerTransportFactory serverTransportFactory) {
     this.messageCodec = messageCodec;
-    this.tcpServer = tcpServer;
+    this.serverTransportFactory = serverTransportFactory;
   }
 
   @Override
@@ -45,24 +43,13 @@ public Address address() {
   @Override
   public Mono<ServerTransport> bind(ServiceMethodRegistry methodRegistry) {
     return Mono.defer(
-        () -> {
-          TcpServer tcpServer =
-              this.tcpServer.doOnConnection(
-                  connection -> {
-                    LOGGER.debug(
-                        "[rsocket][server] Accepted connection on {}", connection.channel());
-                    connection.onDispose(
-                        () ->
-                            LOGGER.debug(
-                                "[rsocket][server] Connection closed on {}", connection.channel()));
-                  });
-          return RSocketServer.create()
-              .acceptor(new RSocketServiceAcceptor(messageCodec, methodRegistry))
-              .payloadDecoder(PayloadDecoder.DEFAULT)
-              .bind(TcpServerTransport.create(tcpServer))
-              .doOnSuccess(channel -> serverChannel = channel)
-              .thenReturn(this);
-        });
+        () ->
+            RSocketServer.create()
+                .acceptor(new RSocketServiceAcceptor(messageCodec, methodRegistry))
+                .payloadDecoder(PayloadDecoder.DEFAULT)
+                .bind(serverTransportFactory.serverTransport())
+                .doOnSuccess(channel -> serverChannel = channel)
+                .thenReturn(this));
   }
 
   @Override
@@ -88,7 +75,6 @@ public Mono<Void> stop() {
   public String toString() {
     return new StringJoiner(", ", RSocketServerTransport.class.getSimpleName() + "[", "]")
         .add("messageCodec=" + messageCodec)
-        .add("tcpServer=" + tcpServer)
         .add("serverChannel=" + serverChannel)
         .toString();
   }

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

@@ -0,0 +1,9 @@
+package io.scalecube.services.transport.rsocket;
+
+import io.rsocket.transport.ServerTransport;
+import io.rsocket.transport.netty.server.CloseableChannel;
+
+public interface RSocketServerTransportFactory {
+
+  ServerTransport<CloseableChannel> serverTransport();
+}

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

@@ -6,6 +6,8 @@
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import io.netty.util.concurrent.Future;
+import io.rsocket.transport.netty.client.TcpClientTransport;
+import io.rsocket.transport.netty.server.TcpServerTransport;
 import io.scalecube.services.transport.api.ClientTransport;
 import io.scalecube.services.transport.api.DataCodec;
 import io.scalecube.services.transport.api.HeadersCodec;
@@ -16,21 +18,26 @@
 import java.util.Collection;
 import java.util.concurrent.ThreadFactory;
 import java.util.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.FutureMono;
 import reactor.netty.resources.LoopResources;
 import reactor.netty.tcp.TcpClient;
 import reactor.netty.tcp.TcpServer;
 
-/** RSocket service transport. */
 public class RSocketServiceTransport implements ServiceTransport {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(RSocketServiceTransport.class);
+
   private int numOfWorkers = Runtime.getRuntime().availableProcessors();
   private HeadersCodec headersCodec;
   private Collection<DataCodec> dataCodecs;
-  private Function<LoopResources, TcpServer> tcpServerProvider = defaultTcpServerProvider();
-  private Function<LoopResources, TcpClient> tcpClientProvider = defaultTcpClientProvider();
+  private Function<LoopResources, RSocketServerTransportFactory> serverTransportFactory =
+      defaultServerTransportFactory();
+  private Function<LoopResources, RSocketClientTransportFactory> clientTransportFactory =
+      defaultClientTransportFactory();
 
   // resources
   private EventLoopGroup eventLoopGroup;
@@ -52,8 +59,8 @@ private RSocketServiceTransport(RSocketServiceTransport other) {
     this.eventLoopGroup = other.eventLoopGroup;
     this.clientLoopResources = other.clientLoopResources;
     this.serverLoopResources = other.serverLoopResources;
-    this.tcpServerProvider = other.tcpServerProvider;
-    this.tcpClientProvider = other.tcpClientProvider;
+    this.serverTransportFactory = other.serverTransportFactory;
+    this.clientTransportFactory = other.clientTransportFactory;
   }
 
   /**
@@ -93,26 +100,28 @@ public RSocketServiceTransport dataCodecs(Collection<DataCodec> dataCodecs) {
   }
 
   /**
-   * Setter for {@code tcpServerProvider}.
+   * Setter for {@code serverTransportFactory}.
    *
-   * @param factory {@code TcpServer} provider function
+   * @param serverTransportFactory serverTransportFactory
    * @return new {@code RSocketServiceTransport} instance
    */
-  public RSocketServiceTransport tcpServer(Function<LoopResources, TcpServer> factory) {
+  public RSocketServiceTransport serverTransportFactory(
+      Function<LoopResources, RSocketServerTransportFactory> serverTransportFactory) {
     RSocketServiceTransport rst = new RSocketServiceTransport(this);
-    rst.tcpServerProvider = factory;
+    rst.serverTransportFactory = serverTransportFactory;
     return rst;
   }
 
   /**
-   * Setter for {@code tcpClientProvider}.
+   * Setter for {@code clientTransportFactory}.
    *
-   * @param factory {@code TcpClient} provider function
+   * @param clientTransportFactory clientTransportFactory
    * @return new {@code RSocketServiceTransport} instance
    */
-  public RSocketServiceTransport tcpClient(Function<LoopResources, TcpClient> factory) {
+  public RSocketServiceTransport clientTransportFactory(
+      Function<LoopResources, RSocketClientTransportFactory> clientTransportFactory) {
     RSocketServiceTransport rst = new RSocketServiceTransport(this);
-    rst.tcpClientProvider = factory;
+    rst.clientTransportFactory = clientTransportFactory;
     return rst;
   }
 
@@ -125,7 +134,7 @@ public RSocketServiceTransport tcpClient(Function<LoopResources, TcpClient> fact
   public ClientTransport clientTransport() {
     return new RSocketClientTransport(
         new ServiceMessageCodec(headersCodec, dataCodecs),
-        tcpClientProvider.apply(clientLoopResources));
+        clientTransportFactory.apply(clientLoopResources));
   }
 
   /**
@@ -137,7 +146,7 @@ public ClientTransport clientTransport() {
   public ServerTransport serverTransport() {
     return new RSocketServerTransport(
         new ServiceMessageCodec(headersCodec, dataCodecs),
-        tcpServerProvider.apply(serverLoopResources));
+        serverTransportFactory.apply(serverLoopResources));
   }
 
   @Override
@@ -173,14 +182,34 @@ private Mono<Void> shutdownEventLoopGroup() {
     return Mono.defer(() -> FutureMono.from((Future) eventLoopGroup.shutdownGracefully()));
   }
 
-  private Function<LoopResources, TcpServer> defaultTcpServerProvider() {
+  private Function<LoopResources, RSocketServerTransportFactory> defaultServerTransportFactory() {
     return (LoopResources serverLoopResources) ->
-        TcpServer.create().runOn(serverLoopResources).bindAddress(() -> new InetSocketAddress(0));
-  }
-
-  private Function<LoopResources, TcpClient> defaultTcpClientProvider() {
-    return (LoopResources clientLoopResources) ->
-        TcpClient.newConnection().runOn(clientLoopResources);
+        () ->
+            TcpServerTransport.create(
+                TcpServer.create()
+                    .runOn(serverLoopResources)
+                    .bindAddress(() -> new InetSocketAddress(0))
+                    .doOnConnection(
+                        connection -> {
+                          LOGGER.debug(
+                              "[rsocket][server] Accepted connection on {}", connection.channel());
+                          connection.onDispose(
+                              () ->
+                                  LOGGER.debug(
+                                      "[rsocket][server] Connection closed on {}",
+                                      connection.channel()));
+                        }));
+  }
+
+  private Function<LoopResources, RSocketClientTransportFactory> defaultClientTransportFactory() {
+    return (LoopResources loopResources) ->
+        (RSocketClientTransportFactory)
+            address ->
+                TcpClientTransport.create(
+                    TcpClient.newConnection()
+                        .host(address.host())
+                        .port(address.port())
+                        .runOn(loopResources));
   }
 
   @Override