Merge pull request #283 from art-community/feature/update-rsocket-version

mizantrop2397 · web-flow · commit d70f234fa80e · 2020-05-15T01:11:47.000+03:00
update rsocket version
diff --git a/README.md b/README.md
@@ -47,6 +47,7 @@ art {
 For bugs, questions and discussions please use the [Github Issues](https://github.com/art-community/art/issues).
 
 Join us on Telegram: https://t.me/art_github
+
 Join us on Discord: https://discord.gg/jaqxB38
 
 ## Documentation
diff --git a/application-rsocket/build.gradle.kts b/application-rsocket/build.gradle.kts
@@ -32,11 +32,11 @@ art {
 
 dependencies {
     with(art.externalDependencyVersionsConfiguration) {
-        embedded("io.rsocket", "rsocket-core", rsocketVersion)
+        embedded("io.rsocket", "rsocket-core", "1.+")
                 .exclude("io.netty")
                 .exclude("io.projectreactor", "reactor-core")
                 .exclude("org.slf4j")
-        embedded("io.rsocket", "rsocket-transport-netty", rsocketVersion)
+        embedded("io.rsocket", "rsocket-transport-netty", "1.+")
                 .exclude("io.netty")
                 .exclude("io.projectreactor", "reactor-core")
                 .exclude("org.slf4j")
diff --git a/application-rsocket/src/main/java/ru/art/rsocket/communicator/RsocketCommunicator.java b/application-rsocket/src/main/java/ru/art/rsocket/communicator/RsocketCommunicator.java
@@ -19,6 +19,7 @@
 package ru.art.rsocket.communicator;
 
 import io.rsocket.*;
+import io.rsocket.core.*;
 import io.rsocket.transport.netty.client.*;
 import lombok.*;
 import org.apache.logging.log4j.*;
@@ -33,7 +34,7 @@
 import ru.art.rsocket.exception.*;
 import ru.art.rsocket.model.*;
 import ru.art.service.model.*;
-import static io.rsocket.RSocketFactory.*;
+import static io.rsocket.core.RSocketConnector.create;
 import static java.text.MessageFormat.*;
 import static java.time.Duration.*;
 import static java.util.Optional.*;
@@ -78,30 +79,29 @@ public class RsocketCommunicator {
 
     private RsocketCommunicator(RsocketCommunicationTargetConfiguration configuration) {
         dataFormat = configuration.dataFormat();
-        ClientRSocketFactory factory = connect();
+        RSocketConnector connector = create();
         if (configuration.resumable()) {
-            factory = factory.resume()
-                    .resumeSessionDuration(ofMillis(configuration.resumeSessionDuration()))
-                    .resumeStreamTimeout(ofMillis(configuration.resumeStreamTimeout()));
+            connector = connector.resume(new Resume()
+                    .sessionDuration(ofMillis(configuration.resumeSessionDuration()))
+                    .streamTimeout(ofMillis(configuration.resumeStreamTimeout()))
+            );
         }
-        rsocketModule().getClientInterceptors().forEach(factory::addRequesterPlugin);
-        configuration.interceptors().forEach(factory::addRequesterPlugin);
+        connector.interceptors(interceptorRegistry -> rsocketModule().getClientInterceptors().forEach(interceptorRegistry::forRequester));
+        connector.interceptors(interceptorRegistry -> configuration.interceptors().forEach(interceptorRegistry::forRequester));
         switch (configuration.transport()) {
             case TCP:
-                rsocketMono = factory.dataMimeType(toMimeType(configuration.dataFormat()))
+                rsocketMono = connector.dataMimeType(toMimeType(configuration.dataFormat()))
                         .metadataMimeType(toMimeType(configuration.dataFormat()))
-                        .transport(TcpClientTransport.create(configuration.host(), configuration.tcpPort()))
-                        .start()
+                        .connect(TcpClientTransport.create(configuration.host(), configuration.tcpPort()))
                         .doOnNext(rsocketModuleState()::registerRsocket)
                         .doOnSubscribe(subscription -> getLogger().info(format(RSOCKET_TCP_COMMUNICATOR_STARTED_MESSAGE, configuration.host(), configuration.tcpPort())));
                 getLogger().info(format(RSOCKET_TCP_COMMUNICATOR_CREATED_MESSAGE, configuration.host(), configuration.tcpPort()));
                 return;
             case WEB_SOCKET:
-                rsocketMono = factory
+                rsocketMono = connector
                         .dataMimeType(toMimeType(configuration.dataFormat()))
                         .metadataMimeType(toMimeType(configuration.dataFormat()))
-                        .transport(WebsocketClientTransport.create(configuration.host(), configuration.tcpPort()))
-                        .start()
+                        .connect(WebsocketClientTransport.create(configuration.host(), configuration.tcpPort()))
                         .doOnNext(rsocketModuleState()::registerRsocket)
                         .doOnSubscribe(subscription -> getLogger().info(format(RSOCKET_WS_COMMUNICATOR_STARTED_MESSAGE, configuration.host(), configuration.webSocketPort())));
                 getLogger().info(format(RSOCKET_WS_COMMUNICATOR_CREATED_MESSAGE, configuration.host(), configuration.tcpPort()));
diff --git a/application-rsocket/src/main/java/ru/art/rsocket/configuration/RsocketModuleConfiguration.java b/application-rsocket/src/main/java/ru/art/rsocket/configuration/RsocketModuleConfiguration.java
@@ -19,6 +19,7 @@
 package ru.art.rsocket.configuration;
 
 import io.rsocket.RSocketFactory.*;
+import io.rsocket.core.*;
 import io.rsocket.plugins.*;
 import io.rsocket.transport.netty.server.*;
 import lombok.*;
@@ -101,7 +102,7 @@ default RsocketCommunicationTargetConfiguration getCommunicationTargetConfigurat
 
     Function<? extends TcpServerTransport, ? extends TcpServerTransport> getTcpServerTransportConfigurator();
 
-    Function<? extends ServerRSocketFactory, ? extends ServerRSocketFactory> getServerFactoryConfigurator();
+    Function<? extends RSocketServer, ? extends RSocketServer> getServerConfigurator();
 
     int getFragmentationMtu();
 
@@ -127,7 +128,7 @@ class RsocketModuleDefaultConfiguration implements RsocketModuleConfiguration {
         private final Function<? extends TcpServer, ? extends TcpServer> tcpServerConfigurator = identity();
         private final Function<? extends WebsocketServerTransport, ? extends WebsocketServerTransport> webSocketServerTransportConfigurator = identity();
         private final Function<? extends TcpServerTransport, ? extends TcpServerTransport> tcpServerTransportConfigurator = identity();
-        private final Function<? extends ServerRSocketFactory, ? extends ServerRSocketFactory> serverFactoryConfigurator = identity();
+        private final Function<? extends RSocketServer, ? extends RSocketServer> serverConfigurator = identity();
         @Getter(lazy = true, onMethod = @__({@SuppressWarnings("unchecked")}))
         private final List<RSocketInterceptor> serverInterceptors = initializeInterceptors();
         @Getter(lazy = true, onMethod = @__({@SuppressWarnings("unchecked")}))
diff --git a/application-rsocket/src/main/java/ru/art/rsocket/server/RsocketServer.java b/application-rsocket/src/main/java/ru/art/rsocket/server/RsocketServer.java
@@ -18,6 +18,7 @@
 
 package ru.art.rsocket.server;
 
+import io.rsocket.core.*;
 import io.rsocket.transport.netty.server.*;
 import lombok.*;
 import org.apache.logging.log4j.Logger;
@@ -28,7 +29,6 @@
 import ru.art.rsocket.exception.*;
 import ru.art.rsocket.socket.*;
 import ru.art.rsocket.specification.*;
-import static io.rsocket.RSocketFactory.*;
 import static java.lang.System.*;
 import static java.lang.Thread.*;
 import static java.text.MessageFormat.*;
@@ -62,39 +62,37 @@ private RsocketServer(RsocketTransport transport) {
     }
 
     private Mono<CloseableChannel> createServer() {
-        ServerRSocketFactory socketFactory = receive().fragment(rsocketModule().getFragmentationMtu());
+        RSocketServer server = RSocketServer.create().fragment(rsocketModule().getFragmentationMtu());
         if (rsocketModule().isResumableServer()) {
-            socketFactory = socketFactory.resume()
-                    .resumeSessionDuration(ofMillis(rsocketModule().getServerResumeSessionDuration()))
-                    .resumeStreamTimeout(ofMillis(rsocketModule().getServerResumeStreamTimeout()));
+            server.resume(new Resume()
+                    .sessionDuration(ofMillis(rsocketModule().getServerResumeSessionDuration()))
+                    .streamTimeout(ofMillis(rsocketModule().getServerResumeStreamTimeout())));
         }
-        rsocketModule().getServerInterceptors().forEach(socketFactory::addResponderPlugin);
-        ServerTransportAcceptor acceptor = rsocketModule()
-                .getServerFactoryConfigurator()
-                .apply(cast(socketFactory))
+        server.interceptors(interceptorRegistry -> rsocketModule().getServerInterceptors().forEach(interceptorRegistry::forResponder));
+        RSocketServer acceptor = rsocketModule()
+                .getServerConfigurator()
+                .apply(cast(server))
                 .acceptor((setup, sendingSocket) -> just(new RsocketAcceptor(sendingSocket, setup)));
         Mono<CloseableChannel> channel;
         switch (transport) {
             case TCP:
-                channel = acceptor.transport(rsocketModule().getTcpServerTransportConfigurator()
+                channel = acceptor.bind(rsocketModule().getTcpServerTransportConfigurator()
                         .apply(cast(TcpServerTransport.create(rsocketModule()
                                 .getTcpServerConfigurator()
                                 .apply(cast(TcpServer.create()
                                         .host(rsocketModule().getServerHost())
                                         .port(rsocketModule().getServerTcpPort())))))))
-                        .start()
                         .onTerminateDetach();
                 rsocketModuleState().setTcpServer(this);
                 break;
             case WEB_SOCKET:
-                channel = acceptor.transport(rsocketModule()
+                channel = acceptor.bind(rsocketModule()
                         .getWebSocketServerTransportConfigurator()
                         .apply(cast(WebsocketServerTransport.create(rsocketModule()
                                 .getWebSocketServerConfigurator()
                                 .apply(cast(HttpServer.create()
                                         .host(rsocketModule().getServerHost())
                                         .port(rsocketModule().getServerWebSocketPort())))))))
-                        .start()
                         .onTerminateDetach();
                 rsocketModuleState().setWebSocketServer(this);
                 break;
diff --git a/application-rsocket/src/main/java/ru/art/rsocket/socket/RsocketAcceptor.java b/application-rsocket/src/main/java/ru/art/rsocket/socket/RsocketAcceptor.java
@@ -21,6 +21,7 @@
 import io.rsocket.*;
 import org.reactivestreams.*;
 import reactor.core.publisher.*;
+import reactor.util.annotation.*;
 import ru.art.rsocket.flux.*;
 import ru.art.rsocket.model.*;
 import ru.art.rsocket.service.*;
@@ -39,7 +40,7 @@
 import static ru.art.rsocket.writer.ServiceResponsePayloadWriter.*;
 import static ru.art.service.ServiceController.*;
 
-public class RsocketAcceptor extends AbstractRSocket {
+public class RsocketAcceptor implements RSocket {
     private final CurrentRsocketState state;
 
     public RsocketAcceptor(RSocket socket, ConnectionSetupPayload setupPayload) {
