From ebc9c575e3187bbe158ce0e6bc3d36217ae32c56 Mon Sep 17 00:00:00 2001 From: Stephane Nicoll Date: Mon, 5 Oct 2020 13:58:57 +0200 Subject: [PATCH] Allow to customize RSocketServer's fragment size Closes gh-23247 --- .../rsocket/RSocketProperties.java | 15 ++++++++++++ .../RSocketServerAutoConfiguration.java | 1 + .../RSocketServerAutoConfigurationTests.java | 23 +++++++++++++++++++ .../netty/NettyRSocketServerFactory.java | 17 +++++++++++++- .../ConfigurableRSocketServerFactory.java | 8 +++++++ 5 files changed, 63 insertions(+), 1 deletion(-) 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 71e5ce4785..a2e6045349 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 @@ -22,6 +22,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.NestedConfigurationProperty; import org.springframework.boot.rsocket.server.RSocketServer; import org.springframework.boot.web.server.Ssl; +import org.springframework.util.unit.DataSize; /** * {@link ConfigurationProperties properties} for RSocket support. @@ -62,6 +63,12 @@ public class RSocketProperties { */ private String mappingPath; + /** + * Maximum transmission unit. Frames larger than the specified value are + * fragmented. + */ + private DataSize fragmentSize; + @NestedConfigurationProperty private Ssl ssl; @@ -97,6 +104,14 @@ public class RSocketProperties { this.mappingPath = mappingPath; } + public DataSize getFragmentSize() { + return this.fragmentSize; + } + + public void setFragmentSize(DataSize fragmentSize) { + this.fragmentSize = fragmentSize; + } + public Ssl getSsl() { return this.ssl; } 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 80902e1de3..b484e904ad 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 @@ -97,6 +97,7 @@ public class RSocketServerAutoConfiguration { PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); map.from(properties.getServer().getAddress()).to(factory::setAddress); map.from(properties.getServer().getPort()).to(factory::setPort); + map.from(properties.getServer().getFragmentSize()).to(factory::setFragmentSize); map.from(properties.getServer().getSsl()).to(factory::setSsl); factory.setRSocketServerCustomizers(customizers.orderedStream().collect(Collectors.toList())); return factory; 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 ae35408230..c9e736b92c 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 @@ -32,6 +32,7 @@ import org.springframework.core.codec.CharSequenceEncoder; import org.springframework.core.codec.StringDecoder; import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler; +import org.springframework.util.unit.DataSize; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; @@ -91,6 +92,28 @@ class RSocketServerAutoConfigurationTests { }); } + @Test + void shouldSetFragmentWhenRSocketServerFragmentSizeIsSet() { + reactiveWebContextRunner() + .withPropertyValues("spring.rsocket.server.port=0", "spring.rsocket.server.fragment-size=12KB") + .run((context) -> { + assertThat(context).hasSingleBean(RSocketServerFactory.class); + RSocketServerFactory factory = context.getBean(RSocketServerFactory.class); + assertThat(factory).hasFieldOrPropertyWithValue("fragmentSize", DataSize.ofKilobytes(12)); + }); + } + + @Test + void shouldFailToSetFragmentWhenRSocketServerFragmentSizeIsBelow64() { + reactiveWebContextRunner() + .withPropertyValues("spring.rsocket.server.port=0", "spring.rsocket.server.fragment-size=60B") + .run((context) -> { + assertThat(context).hasFailed(); + assertThat(context.getStartupFailure()) + .hasMessageContaining("The smallest allowed mtu size is 64 bytes, provided: 60"); + }); + } + @Test void shouldUseSslWhenRocketServerSslIsConfigured() { reactiveWebContextRunner() 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 278d066e41..2a51683aa6 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 @@ -33,6 +33,7 @@ import reactor.core.publisher.Mono; import reactor.netty.http.server.HttpServer; import reactor.netty.tcp.TcpServer; +import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.boot.rsocket.server.ConfigurableRSocketServerFactory; import org.springframework.boot.rsocket.server.RSocketServer; import org.springframework.boot.rsocket.server.RSocketServerCustomizer; @@ -42,6 +43,7 @@ import org.springframework.boot.web.server.Ssl; import org.springframework.boot.web.server.SslStoreProvider; import org.springframework.http.client.reactive.ReactorResourceFactory; import org.springframework.util.Assert; +import org.springframework.util.unit.DataSize; /** * {@link RSocketServerFactory} that can be used to create {@link RSocketServer}s backed @@ -55,6 +57,8 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur private int port = 9898; + private DataSize fragmentSize; + private InetAddress address; private RSocketServer.Transport transport = RSocketServer.Transport.TCP; @@ -74,6 +78,11 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur this.port = port; } + @Override + public void setFragmentSize(DataSize fragmentSize) { + this.fragmentSize = fragmentSize; + } + @Override public void setAddress(InetAddress address) { this.address = address; @@ -138,11 +147,17 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur public NettyRSocketServer create(SocketAcceptor socketAcceptor) { ServerTransport transport = createTransport(); io.rsocket.core.RSocketServer server = io.rsocket.core.RSocketServer.create(socketAcceptor); - this.rSocketServerCustomizers.forEach((customizer) -> customizer.customize(server)); + configureServer(server); Mono starter = server.bind(transport); return new NettyRSocketServer(starter, this.lifecycleTimeout); } + private void configureServer(io.rsocket.core.RSocketServer server) { + PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + map.from(this.fragmentSize).asInt(DataSize::toBytes).to(server::fragment); + this.rSocketServerCustomizers.forEach((customizer) -> customizer.customize(server)); + } + private ServerTransport createTransport() { if (this.transport == RSocketServer.Transport.WEBSOCKET) { return createWebSocketTransport(); 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 40de9b3f7f..c825fafa1a 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 @@ -20,6 +20,7 @@ import java.net.InetAddress; import org.springframework.boot.web.server.Ssl; import org.springframework.boot.web.server.SslStoreProvider; +import org.springframework.util.unit.DataSize; /** * A configurable {@link RSocketServerFactory}. @@ -36,6 +37,13 @@ public interface ConfigurableRSocketServerFactory { */ void setPort(int port); + /** + * Specify the maximum transmission unit. Frames larger than the specified + * {@code fragmentSize} are fragmented. + * @param fragmentSize the fragment size + */ + void setFragmentSize(DataSize fragmentSize); + /** * Set the specific network address that the server should bind to. * @param address the address to set (defaults to {@code null})