Apply new rsocket API

Author: segabriel

pom.xml

@@ -23,7 +23,7 @@
 
   <properties>
     <jackson.version>2.11.0</jackson.version>
-    <scalecube-cluster.version>2.4.11-SNAPSHOT</scalecube-cluster.version>
+    <scalecube-cluster.version>2.4.11</scalecube-cluster.version>
     <scalecube-commons.version>1.0.2</scalecube-commons.version>
     <scalecube-benchmarks.version>1.2.2</scalecube-benchmarks.version>
     <scalecube-config.version>0.4.3</scalecube-config.version>
@@ -96,6 +96,11 @@
       </dependency>
 
       <!-- Logging -->
+      <dependency>
+        <groupId>org.slf4j</groupId>
+        <artifactId>slf4j-api</artifactId>
+        <version>${slf4j.version}</version>
+      </dependency>
       <dependency>
         <groupId>org.apache.logging.log4j</groupId>
         <artifactId>log4j-bom</artifactId>

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

@@ -1,7 +1,7 @@
 package io.scalecube.services.transport.rsocket;
 
 import io.rsocket.RSocket;
-import io.rsocket.RSocketFactory;
+import io.rsocket.core.RSocketConnector;
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.scalecube.net.Address;
@@ -46,16 +46,10 @@ 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());
-
-    Mono<RSocket> rsocketMono =
-        RSocketFactory.connect()
-            .frameDecoder(PayloadDecoder.DEFAULT)
-            .errorConsumer(
-                th -> LOGGER.warn("Exception occurred at rsocket client transport: " + th))
-            .transport(() -> TcpClientTransport.create(tcpClient))
-            .start();
-
-    return rsocketMono
+    return RSocketConnector.create()
+        .payloadDecoder(PayloadDecoder.DEFAULT)
+        .errorConsumer(th1 -> LOGGER.warn("Exception occurred at rsocket client transport: " + th1))
+        .connect(() -> TcpClientTransport.create(tcpClient))
         .doOnSuccess(
             rsocket -> {
               LOGGER.info("Connected successfully on {}", address);

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

@@ -1,6 +1,7 @@
 package io.scalecube.services.transport.rsocket;
 
 import io.rsocket.RSocketFactory;
+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;
@@ -52,14 +53,12 @@ public Mono<ServerTransport> bind(ServiceMethodRegistry methodRegistry) {
                     connection.onDispose(
                         () -> LOGGER.info("Connection closed on {}", connection.channel()));
                   });
-
-          return RSocketFactory.receive()
-              .frameDecoder(PayloadDecoder.DEFAULT)
+          return RSocketServer.create()
+              .acceptor(new RSocketServiceAcceptor(codec, methodRegistry))
+              .payloadDecoder(PayloadDecoder.DEFAULT)
               .errorConsumer(
                   th -> LOGGER.warn("Exception occurred at rsocket server transport: " + th))
-              .acceptor(new RSocketServiceAcceptor(codec, methodRegistry))
-              .transport(() -> TcpServerTransport.create(tcpServer))
-              .start()
+              .bind(TcpServerTransport.create(tcpServer))
               .doOnSuccess(channel -> serverChannel = channel)
               .thenReturn(this);
         });