diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketProperties.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketProperties.java index 792d63cf8d..232eb1db42 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketProperties.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketProperties.java @@ -51,7 +51,7 @@ public class RSocketProperties { /** * RSocket transport protocol. */ - private RSocketServer.TRANSPORT transport = RSocketServer.TRANSPORT.TCP; + private RSocketServer.Transport transport = RSocketServer.Transport.TCP; /** * Path under which RSocket handles requests (only works with websocket @@ -75,11 +75,11 @@ public class RSocketProperties { this.address = address; } - public RSocketServer.TRANSPORT getTransport() { + public RSocketServer.Transport getTransport() { return this.transport; } - public void setTransport(RSocketServer.TRANSPORT transport) { + public void setTransport(RSocketServer.Transport transport) { this.transport = transport; } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfiguration.java index 1bfdcfde1e..b6e3baca86 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfiguration.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfiguration.java @@ -34,8 +34,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.boot.context.properties.PropertyMapper; +import org.springframework.boot.rsocket.context.RSocketServerBootstrap; import org.springframework.boot.rsocket.netty.NettyRSocketServerFactory; -import org.springframework.boot.rsocket.server.RSocketServerBootstrap; import org.springframework.boot.rsocket.server.RSocketServerFactory; import org.springframework.boot.rsocket.server.ServerRSocketFactoryCustomizer; import org.springframework.context.annotation.Bean; diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfigurationTests.java index cf4fedaeea..79b6619c23 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfigurationTests.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfigurationTests.java @@ -20,7 +20,7 @@ import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.rsocket.context.RSocketPortInfoApplicationContextInitializer; -import org.springframework.boot.rsocket.server.RSocketServerBootstrap; +import org.springframework.boot.rsocket.context.RSocketServerBootstrap; import org.springframework.boot.rsocket.server.RSocketServerFactory; import org.springframework.boot.rsocket.server.ServerRSocketFactoryCustomizer; import org.springframework.boot.test.context.runner.ApplicationContextRunner; diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketPortInfoApplicationContextInitializer.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketPortInfoApplicationContextInitializer.java index 09fb5d05f5..4b85aff802 100644 --- a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketPortInfoApplicationContextInitializer.java +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketPortInfoApplicationContextInitializer.java @@ -63,7 +63,7 @@ public class RSocketPortInfoApplicationContextInitializer @Override public void onApplicationEvent(RSocketServerInitializedEvent event) { - setPortProperty(this.applicationContext, event.getrSocketServer().address().getPort()); + setPortProperty(this.applicationContext, event.getServer().address().getPort()); } private void setPortProperty(ApplicationContext context, int port) { diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerBootstrap.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketServerBootstrap.java similarity index 67% rename from spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerBootstrap.java rename to spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketServerBootstrap.java index 484c32b17d..118522a129 100644 --- a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerBootstrap.java +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketServerBootstrap.java @@ -14,14 +14,16 @@ * limitations under the License. */ -package org.springframework.boot.rsocket.server; +package org.springframework.boot.rsocket.context; import io.rsocket.SocketAcceptor; -import org.springframework.boot.rsocket.context.RSocketServerInitializedEvent; +import org.springframework.boot.rsocket.server.RSocketServer; +import org.springframework.boot.rsocket.server.RSocketServerFactory; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.SmartLifecycle; +import org.springframework.util.Assert; /** * Bootstrap an {@link RSocketServer} and start it with the application context. @@ -31,33 +33,34 @@ import org.springframework.context.SmartLifecycle; */ public class RSocketServerBootstrap implements ApplicationEventPublisherAware, SmartLifecycle { - private final RSocketServer rSocketServer; + private final RSocketServer server; - private ApplicationEventPublisher applicationEventPublisher; + private ApplicationEventPublisher eventPublisher; - public RSocketServerBootstrap(RSocketServerFactory serverFactoryProvider, SocketAcceptor socketAcceptor) { - this.rSocketServer = serverFactoryProvider.create(socketAcceptor); + public RSocketServerBootstrap(RSocketServerFactory serverFactory, SocketAcceptor socketAcceptor) { + Assert.notNull(serverFactory, "ServerFactory must not be null"); + this.server = serverFactory.create(socketAcceptor); } @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { - this.applicationEventPublisher = applicationEventPublisher; + this.eventPublisher = applicationEventPublisher; } @Override public void start() { - this.rSocketServer.start(); - this.applicationEventPublisher.publishEvent(new RSocketServerInitializedEvent(this.rSocketServer)); + this.server.start(); + this.eventPublisher.publishEvent(new RSocketServerInitializedEvent(this.server)); } @Override public void stop() { - this.rSocketServer.stop(); + this.server.stop(); } @Override public boolean isRunning() { - RSocketServer server = this.rSocketServer; + RSocketServer server = this.server; if (server != null) { return server.address() != null; } diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketServerInitializedEvent.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketServerInitializedEvent.java index 1472e00712..a4f075364b 100644 --- a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketServerInitializedEvent.java +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketServerInitializedEvent.java @@ -27,18 +27,17 @@ import org.springframework.context.ApplicationEvent; * @author Brian Clozel * @since 2.2.0 */ -@SuppressWarnings("serial") public class RSocketServerInitializedEvent extends ApplicationEvent { - public RSocketServerInitializedEvent(RSocketServer rSocketServer) { - super(rSocketServer); + public RSocketServerInitializedEvent(RSocketServer server) { + super(server); } /** * Access the {@link RSocketServer}. * @return the embedded RSocket server */ - public RSocketServer getrSocketServer() { + public RSocketServer getServer() { return getSource(); } diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServer.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServer.java index e8ffb00987..cb2df6779f 100644 --- a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServer.java +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServer.java @@ -61,25 +61,13 @@ public class NettyRSocketServer implements RSocketServer { @Override public void start() throws RSocketServerException { - if (this.lifecycleTimeout != null) { - this.channel = this.starter.block(this.lifecycleTimeout); - } - else { - this.channel = this.starter.block(); - } + this.channel = block(this.starter, this.lifecycleTimeout); logger.info("Netty RSocket started on port(s): " + address().getPort()); startDaemonAwaitThread(this.channel); } private void startDaemonAwaitThread(CloseableChannel channel) { - Thread awaitThread = new Thread("rsocket") { - - @Override - public void run() { - channel.onClose().block(); - } - - }; + Thread awaitThread = new Thread(() -> channel.onClose().block(), "rsocket"); awaitThread.setContextClassLoader(getClass().getClassLoader()); awaitThread.setDaemon(false); awaitThread.start(); @@ -93,4 +81,8 @@ public class NettyRSocketServer implements RSocketServer { } } + private T block(Mono mono, Duration timeout) { + return (timeout != null) ? mono.block(timeout) : mono.block(); + } + } diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactory.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactory.java index 7cc67e8114..a0027895ad 100644 --- a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactory.java +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactory.java @@ -54,7 +54,7 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur private InetAddress address; - private RSocketServer.TRANSPORT transport = RSocketServer.TRANSPORT.TCP; + private RSocketServer.Transport transport = RSocketServer.Transport.TCP; private ReactorResourceFactory resourceFactory; @@ -73,7 +73,7 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur } @Override - public void setTransport(RSocketServer.TRANSPORT transport) { + public void setTransport(RSocketServer.Transport transport) { this.transport = transport; } @@ -126,26 +126,28 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur } private ServerTransport createTransport() { - if (this.transport == RSocketServer.TRANSPORT.WEBSOCKET) { - if (this.resourceFactory != null) { - HttpServer httpServer = HttpServer.create().tcpConfiguration((tcpServer) -> tcpServer - .runOn(this.resourceFactory.getLoopResources()).addressSupplier(this::getListenAddress)); - return WebsocketServerTransport.create(httpServer); - } - else { - return WebsocketServerTransport.create(getListenAddress()); - } + if (this.transport == RSocketServer.Transport.WEBSOCKET) { + return createWebSocketTransport(); } - else { - if (this.resourceFactory != null) { - TcpServer tcpServer = TcpServer.create().runOn(this.resourceFactory.getLoopResources()) - .addressSupplier(this::getListenAddress); - return TcpServerTransport.create(tcpServer); - } - else { - return TcpServerTransport.create(getListenAddress()); - } + return createTcpTransport(); + } + + private ServerTransport createWebSocketTransport() { + if (this.resourceFactory != null) { + HttpServer httpServer = HttpServer.create().tcpConfiguration((tcpServer) -> tcpServer + .runOn(this.resourceFactory.getLoopResources()).addressSupplier(this::getListenAddress)); + return WebsocketServerTransport.create(httpServer); } + return WebsocketServerTransport.create(getListenAddress()); + } + + private ServerTransport createTcpTransport() { + if (this.resourceFactory != null) { + TcpServer tcpServer = TcpServer.create().runOn(this.resourceFactory.getLoopResources()) + .addressSupplier(this::getListenAddress); + return TcpServerTransport.create(tcpServer); + } + return TcpServerTransport.create(getListenAddress()); } private InetSocketAddress getListenAddress() { diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ConfigurableRSocketServerFactory.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ConfigurableRSocketServerFactory.java index a6bfe201f3..cb97410542 100644 --- a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ConfigurableRSocketServerFactory.java +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ConfigurableRSocketServerFactory.java @@ -43,6 +43,6 @@ public interface ConfigurableRSocketServerFactory { * Set the transport that the RSocket server should use. * @param transport the transport protocol to use */ - void setTransport(RSocketServer.TRANSPORT transport); + void setTransport(RSocketServer.Transport transport); } diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServer.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServer.java index 768576353b..831d6854e6 100644 --- a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServer.java +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServer.java @@ -50,9 +50,17 @@ public interface RSocketServer { /** * Choice of transport protocol for the RSocket server. */ - enum TRANSPORT { + enum Transport { - TCP, WEBSOCKET + /** + * TCP transport protocol. + */ + TCP, + + /** + * WebSocket transport protocol. + */ + WEBSOCKET } diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerException.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerException.java index b0c9dc84ea..658f641162 100644 --- a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerException.java +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerException.java @@ -22,7 +22,6 @@ package org.springframework.boot.rsocket.server; * @author Brian Clozel * @since 2.2.0 */ -@SuppressWarnings("serial") public class RSocketServerException extends RuntimeException { public RSocketServerException(String message, Throwable cause) { diff --git a/spring-boot-project/spring-boot/src/test/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactoryTests.java b/spring-boot-project/spring-boot/src/test/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactoryTests.java index b811018c50..0178c64772 100644 --- a/spring-boot-project/spring-boot/src/test/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactoryTests.java +++ b/spring-boot-project/spring-boot/src/test/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactoryTests.java @@ -59,7 +59,7 @@ import static org.mockito.Mockito.mock; */ class NettyRSocketServerFactoryTests { - private NettyRSocketServer rSocketServer; + private NettyRSocketServer server; private RSocketRequester requester; @@ -67,9 +67,9 @@ class NettyRSocketServerFactoryTests { @AfterEach void tearDown() { - if (this.rSocketServer != null) { + if (this.server != null) { try { - this.rSocketServer.stop(); + this.server.stop(); } catch (Exception ex) { // Ignore @@ -89,47 +89,44 @@ class NettyRSocketServerFactoryTests { NettyRSocketServerFactory factory = getFactory(); int specificPort = SocketUtils.findAvailableTcpPort(41000); factory.setPort(specificPort); - this.rSocketServer = factory.create(new EchoRequestResponseAcceptor()); - this.rSocketServer.start(); + this.server = factory.create(new EchoRequestResponseAcceptor()); + this.server.start(); this.requester = createRSocketTcpClient(); String payload = "test payload"; String response = this.requester.route("test").data(payload).retrieveMono(String.class).block(TIMEOUT); - - assertThat(this.rSocketServer.address().getPort()).isEqualTo(specificPort); + assertThat(this.server.address().getPort()).isEqualTo(specificPort); assertThat(response).isEqualTo(payload); - assertThat(this.rSocketServer.address().getPort()).isEqualTo(specificPort); + assertThat(this.server.address().getPort()).isEqualTo(specificPort); } @Test void websocketTransport() { NettyRSocketServerFactory factory = getFactory(); - factory.setTransport(RSocketServer.TRANSPORT.WEBSOCKET); - this.rSocketServer = factory.create(new EchoRequestResponseAcceptor()); - this.rSocketServer.start(); + factory.setTransport(RSocketServer.Transport.WEBSOCKET); + this.server = factory.create(new EchoRequestResponseAcceptor()); + this.server.start(); this.requester = createRSocketWebSocketClient(); String payload = "test payload"; String response = this.requester.route("test").data(payload).retrieveMono(String.class).block(TIMEOUT); - assertThat(response).isEqualTo(payload); } @Test void websocketTransportWithReactorResource() { NettyRSocketServerFactory factory = getFactory(); - factory.setTransport(RSocketServer.TRANSPORT.WEBSOCKET); + factory.setTransport(RSocketServer.Transport.WEBSOCKET); ReactorResourceFactory resourceFactory = new ReactorResourceFactory(); resourceFactory.afterPropertiesSet(); factory.setResourceFactory(resourceFactory); int specificPort = SocketUtils.findAvailableTcpPort(41000); factory.setPort(specificPort); - this.rSocketServer = factory.create(new EchoRequestResponseAcceptor()); - this.rSocketServer.start(); + this.server = factory.create(new EchoRequestResponseAcceptor()); + this.server.start(); this.requester = createRSocketWebSocketClient(); String payload = "test payload"; String response = this.requester.route("test").data(payload).retrieveMono(String.class).block(TIMEOUT); - assertThat(response).isEqualTo(payload); - assertThat(this.rSocketServer.address().getPort()).isEqualTo(specificPort); + assertThat(this.server.address().getPort()).isEqualTo(specificPort); } @Test @@ -142,7 +139,7 @@ class NettyRSocketServerFactoryTests { .will((invocation) -> invocation.getArgument(0)); } factory.setServerCustomizers(Arrays.asList(customizers[0], customizers[1])); - this.rSocketServer = factory.create(new EchoRequestResponseAcceptor()); + this.server = factory.create(new EchoRequestResponseAcceptor()); InOrder ordered = inOrder((Object[]) customizers); for (ServerRSocketFactoryCustomizer customizer : customizers) { ordered.verify(customizer).apply(any(RSocketFactory.ServerRSocketFactory.class)); @@ -150,14 +147,14 @@ class NettyRSocketServerFactoryTests { } private RSocketRequester createRSocketTcpClient() { - Assertions.assertThat(this.rSocketServer).isNotNull(); - InetSocketAddress address = this.rSocketServer.address(); + Assertions.assertThat(this.server).isNotNull(); + InetSocketAddress address = this.server.address(); return createRSocketRequesterBuilder().connectTcp(address.getHostString(), address.getPort()).block(); } private RSocketRequester createRSocketWebSocketClient() { - Assertions.assertThat(this.rSocketServer).isNotNull(); - InetSocketAddress address = this.rSocketServer.address(); + Assertions.assertThat(this.server).isNotNull(); + InetSocketAddress address = this.server.address(); return createRSocketRequesterBuilder().connect(WebsocketClientTransport.create(address)).block(); }