Added RSocketClientTransportFactory and RSocketClientTransportFactory

Author: artem-v

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

@@ -14,6 +14,7 @@
 import io.scalecube.services.ServiceEndpoint;
 import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
 import io.scalecube.services.discovery.api.ServiceDiscovery;
+import io.scalecube.services.transport.rsocket.RSocketServerTransportFactory;
 import io.scalecube.services.transport.rsocket.RSocketServiceTransport;
 import java.net.InetSocketAddress;
 import java.nio.file.Path;

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

@@ -1,9 +1,30 @@
 package io.scalecube.services.transport.rsocket;
 
 import io.rsocket.transport.ClientTransport;
+import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.scalecube.net.Address;
+import java.util.function.Function;
+import reactor.netty.resources.LoopResources;
+import reactor.netty.tcp.TcpClient;
 
 public interface RSocketClientTransportFactory {
 
+  /**
+   * Returns default rsocket tcp client transport factory.
+   *
+   * @see TcpClientTransport
+   * @return factory function for {@link RSocketClientTransportFactory}
+   */
+  static Function<LoopResources, RSocketClientTransportFactory> tcp() {
+    return (LoopResources loopResources) ->
+        (RSocketClientTransportFactory)
+            address ->
+                TcpClientTransport.create(
+                    TcpClient.newConnection()
+                        .host(address.host())
+                        .port(address.port())
+                        .runOn(loopResources));
+  }
+
   ClientTransport clientTransport(Address address);
 }

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

@@ -2,8 +2,39 @@
 
 import io.rsocket.transport.ServerTransport;
 import io.rsocket.transport.netty.server.CloseableChannel;
+import io.rsocket.transport.netty.server.TcpServerTransport;
+import java.net.InetSocketAddress;
+import java.util.function.Function;
+import reactor.netty.resources.LoopResources;
+import reactor.netty.tcp.TcpServer;
 
 public interface RSocketServerTransportFactory {
 
+  /**
+   * Returns default rsocket tcp server transport factory (shall listen on port {@code 0}).
+   *
+   * @see TcpServerTransport
+   * @return factory function for {@link RSocketServerTransportFactory}
+   */
+  static Function<LoopResources, RSocketServerTransportFactory> tcp() {
+    return tcp(0);
+  }
+
+  /**
+   * Returns default rsocket tcp server transport factory.
+   *
+   * @param port port
+   * @see TcpServerTransport
+   * @return factory function for {@link RSocketServerTransportFactory}
+   */
+  static Function<LoopResources, RSocketServerTransportFactory> tcp(int port) {
+    return (LoopResources loopResources) ->
+        () ->
+            TcpServerTransport.create(
+                TcpServer.create()
+                    .runOn(loopResources)
+                    .bindAddress(() -> new InetSocketAddress(port)));
+  }
+
   ServerTransport<CloseableChannel> serverTransport();
 }

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

@@ -6,38 +6,29 @@
 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;
 import io.scalecube.services.transport.api.ServerTransport;
 import io.scalecube.services.transport.api.ServiceMessageCodec;
 import io.scalecube.services.transport.api.ServiceTransport;
-import java.net.InetSocketAddress;
 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;
 
 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, RSocketServerTransportFactory> serverTransportFactory =
-      defaultServerTransportFactory();
+      RSocketServerTransportFactory.tcp();
   private Function<LoopResources, RSocketClientTransportFactory> clientTransportFactory =
-      defaultClientTransportFactory();
+      RSocketClientTransportFactory.tcp();
 
   // resources
   private EventLoopGroup eventLoopGroup;
@@ -182,36 +173,6 @@ private Mono<Void> shutdownEventLoopGroup() {
     return Mono.defer(() -> FutureMono.from((Future) eventLoopGroup.shutdownGracefully()));
   }
 
-  private Function<LoopResources, RSocketServerTransportFactory> defaultServerTransportFactory() {
-    return (LoopResources serverLoopResources) ->
-        () ->
-            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
   public String toString() {
     return "RSocketServiceTransport{"