diff --git a/spring-boot-project/spring-boot-autoconfigure/pom.xml b/spring-boot-project/spring-boot-autoconfigure/pom.xml index 74846e2705..83e120d46b 100755 --- a/spring-boot-project/spring-boot-autoconfigure/pom.xml +++ b/spring-boot-project/spring-boot-autoconfigure/pom.xml @@ -112,6 +112,21 @@ reactor-netty true + + io.rsocket + rsocket-core + true + + + io.rsocket + rsocket-transport-netty + true + + + io.searchbox + jest + true + jakarta.json.bind jakarta.json.bind-api @@ -127,11 +142,6 @@ money-api true - - io.searchbox - jest - true - org.apache.kafka kafka-streams diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketMessagingAutoConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketMessagingAutoConfiguration.java new file mode 100644 index 0000000000..4d3b341fae --- /dev/null +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketMessagingAutoConfiguration.java @@ -0,0 +1,53 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.autoconfigure.rsocket; + +import io.rsocket.RSocketFactory; +import io.rsocket.transport.netty.server.TcpServerTransport; + +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.rsocket.MessageHandlerAcceptor; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; + +/** + * {@link EnableAutoConfiguration} for Spring RSocket support in Spring Messaging. + * + * @author Brian Clozel + * @since 2.2.0 + */ +@Configuration(proxyBeanMethods = false) +@ConditionalOnClass({ RSocketRequester.class, RSocketFactory.class, + TcpServerTransport.class }) +@AutoConfigureAfter(RSocketStrategiesAutoConfiguration.class) +public class RSocketMessagingAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public MessageHandlerAcceptor messageHandlerAcceptor( + RSocketStrategies rSocketStrategies) { + MessageHandlerAcceptor acceptor = new MessageHandlerAcceptor(); + acceptor.setRSocketStrategies(rSocketStrategies); + return acceptor; + } + +} diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketNettyServerCustomizer.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketNettyServerCustomizer.java new file mode 100644 index 0000000000..7bfb68281b --- /dev/null +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketNettyServerCustomizer.java @@ -0,0 +1,52 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.autoconfigure.rsocket; + +import io.rsocket.RSocketFactory; +import io.rsocket.transport.ServerTransport; +import io.rsocket.transport.netty.server.WebsocketRouteTransport; +import reactor.netty.http.server.HttpServer; + +import org.springframework.boot.web.embedded.netty.NettyServerCustomizer; +import org.springframework.messaging.rsocket.MessageHandlerAcceptor; + +/** + * {@libnk NettyServerCustomizer} that configures an RSocket Websocket endpoint. + * + * @author Brian Clozel + */ +class RSocketNettyServerCustomizer implements NettyServerCustomizer { + + private final String mappingPath; + + private final MessageHandlerAcceptor messageHandlerAcceptor; + + RSocketNettyServerCustomizer(String mappingPath, + MessageHandlerAcceptor messageHandlerAcceptor) { + this.mappingPath = mappingPath; + this.messageHandlerAcceptor = messageHandlerAcceptor; + } + + @Override + public HttpServer apply(HttpServer httpServer) { + final ServerTransport.ConnectionAcceptor acceptor = RSocketFactory.receive() + .acceptor(this.messageHandlerAcceptor).toConnectionAcceptor(); + return httpServer.route((routes) -> routes.ws(this.mappingPath, + WebsocketRouteTransport.newHandler(acceptor))); + } + +} 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 new file mode 100644 index 0000000000..531004d01c --- /dev/null +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketProperties.java @@ -0,0 +1,101 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.autoconfigure.rsocket; + +import java.net.InetAddress; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * {@link ConfigurationProperties properties} for RSocket support. + * + * @author Brian Clozel + * @since 2.2.0 + */ +@ConfigurationProperties("spring.rsocket") +public class RSocketProperties { + + private Server server = new Server(); + + public Server getServer() { + return this.server; + } + + static class Server { + + /** + * Server port. + */ + private Integer port; + + /** + * Network address to which the server should bind. + */ + private InetAddress address; + + /** + * RSocket transport protocol. + */ + private Transport transport = Transport.TCP; + + /** + * Path under which RSocket handles requests (only works with websocket + * transport). + */ + private String mappingPath; + + public Integer getPort() { + return this.port; + } + + public void setPort(Integer port) { + this.port = port; + } + + public InetAddress getAddress() { + return this.address; + } + + public void setAddress(InetAddress address) { + this.address = address; + } + + public Transport getTransport() { + return this.transport; + } + + public void setTransport(Transport transport) { + this.transport = transport; + } + + public String getMappingPath() { + return this.mappingPath; + } + + public void setMappingPath(String mappingPath) { + this.mappingPath = mappingPath; + } + + public enum Transport { + + TCP, WEBSOCKET + + } + + } + +} 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 new file mode 100644 index 0000000000..d493e6e64c --- /dev/null +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfiguration.java @@ -0,0 +1,146 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.autoconfigure.rsocket; + +import java.util.stream.Collectors; + +import io.netty.buffer.PooledByteBufAllocator; +import io.rsocket.RSocketFactory; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.condition.AllNestedConditions; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +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.netty.NettyRSocketBootstrap; +import org.springframework.boot.rsocket.netty.NettyRSocketServerFactory; +import org.springframework.boot.rsocket.server.RSocketServerFactory; +import org.springframework.boot.rsocket.server.ServerRSocketFactoryCustomizer; +import org.springframework.boot.web.embedded.netty.NettyReactiveWebServerFactory; +import org.springframework.boot.web.server.WebServerFactoryCustomizer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Conditional; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.client.reactive.ReactorResourceFactory; +import org.springframework.messaging.rsocket.MessageHandlerAcceptor; +import org.springframework.messaging.rsocket.RSocketStrategies; + +/** + * {@link EnableAutoConfiguration Auto-configuration} for RSocket servers. In the case of + * {@link org.springframework.boot.WebApplicationType#REACTIVE}, the RSocket server is + * added as a WebSocket endpoint on the existing + * {@link org.springframework.boot.web.embedded.netty.NettyWebServer}. If a specific + * server port is configured, a new standalone RSocket server is created. + * + * @author Brian Clozel + * @since 2.2.0 + */ +@Configuration(proxyBeanMethods = false) +@ConditionalOnClass({ RSocketFactory.class, RSocketStrategies.class, + PooledByteBufAllocator.class }) +@ConditionalOnBean(MessageHandlerAcceptor.class) +@AutoConfigureAfter(RSocketStrategiesAutoConfiguration.class) +@EnableConfigurationProperties(RSocketProperties.class) +public class RSocketServerAutoConfiguration { + + @Conditional(OnRSocketWebServerCondition.class) + @Configuration(proxyBeanMethods = false) + static class WebFluxServerAutoConfiguration { + + @Bean + public WebServerFactoryCustomizer rSocketWebsocketCustomizer( + RSocketProperties properties, + MessageHandlerAcceptor messageHandlerAcceptor) { + RSocketNettyServerCustomizer customizer = new RSocketNettyServerCustomizer( + properties.getServer().getMappingPath(), messageHandlerAcceptor); + return (factory) -> factory.addServerCustomizers(customizer); + } + + } + + @ConditionalOnProperty(prefix = "spring.rsocket.server", name = "port") + @Configuration(proxyBeanMethods = false) + static class EmbeddedServerAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public ReactorResourceFactory reactorServerResourceFactory() { + return new ReactorResourceFactory(); + } + + @Bean + @ConditionalOnMissingBean + public RSocketServerFactory rSocketServerFactory(RSocketProperties properties, + ReactorResourceFactory resourceFactory, + ObjectProvider customizers) { + NettyRSocketServerFactory factory = new NettyRSocketServerFactory(); + factory.setResourceFactory(resourceFactory); + PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + map.from(properties.getServer().getAddress()).to(factory::setAddress); + map.from(properties.getServer().getPort()).to(factory::setPort); + factory.setServerCustomizers( + customizers.orderedStream().collect(Collectors.toList())); + return factory; + } + + @Bean + public NettyRSocketBootstrap nettyRSocketBootstrap( + RSocketServerFactory rSocketServerFactory, + MessageHandlerAcceptor messageHandlerAcceptor) { + return new NettyRSocketBootstrap(rSocketServerFactory, + messageHandlerAcceptor); + } + + } + + static class OnRSocketWebServerCondition extends AllNestedConditions { + + OnRSocketWebServerCondition() { + super(ConfigurationPhase.REGISTER_BEAN); + } + + @ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE) + static class IsReactiveWebApplication { + + } + + @ConditionalOnProperty(prefix = "spring.rsocket.server", name = "port", + matchIfMissing = true) + static class HasNoPortConfigured { + + } + + @ConditionalOnProperty(prefix = "spring.rsocket.server", name = "mapping-path") + static class HasMappingPathConfigured { + + } + + @ConditionalOnProperty(prefix = "spring.rsocket.server", name = "transport", + havingValue = "websocket") + static class HasWebsocketTransportConfigured { + + } + + } + +} diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketStrategiesAutoConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketStrategiesAutoConfiguration.java new file mode 100644 index 0000000000..eac7092cde --- /dev/null +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketStrategiesAutoConfiguration.java @@ -0,0 +1,84 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.autoconfigure.rsocket; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.netty.buffer.PooledByteBufAllocator; +import io.rsocket.RSocketFactory; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration; +import org.springframework.boot.rsocket.messaging.RSocketStrategiesCustomizer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.core.annotation.Order; +import org.springframework.core.io.buffer.NettyDataBufferFactory; +import org.springframework.http.MediaType; +import org.springframework.http.codec.json.Jackson2JsonDecoder; +import org.springframework.http.codec.json.Jackson2JsonEncoder; +import org.springframework.messaging.rsocket.RSocketStrategies; + +/** + * {@link EnableAutoConfiguration} for {@link RSocketStrategies}. + * + * @author Brian Clozel + * @since 2.2.0 + */ +@Configuration(proxyBeanMethods = false) +@ConditionalOnClass({ RSocketFactory.class, RSocketStrategies.class, + PooledByteBufAllocator.class }) +@AutoConfigureAfter(JacksonAutoConfiguration.class) +public class RSocketStrategiesAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public RSocketStrategies rSocketStrategies( + ObjectProvider customizers) { + RSocketStrategies.Builder builder = RSocketStrategies.builder(); + builder.reactiveAdapterStrategy(ReactiveAdapterRegistry.getSharedInstance()); + customizers.stream().forEach((customizer) -> customizer.customize(builder)); + builder.dataBufferFactory( + new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)); + return builder.build(); + } + + @Configuration(proxyBeanMethods = false) + @ConditionalOnClass(ObjectMapper.class) + protected static class JacksonStrategyConfiguration { + + @Bean + @Order(0) + @ConditionalOnBean(ObjectMapper.class) + public RSocketStrategiesCustomizer jacksonStrategyCustomizer( + ObjectMapper objectMapper) { + return (strategy) -> { + MediaType[] supportedTypes = new MediaType[] { MediaType.APPLICATION_JSON, + new MediaType("application", "*+json") }; + strategy.decoder(new Jackson2JsonDecoder(objectMapper, supportedTypes)); + strategy.encoder(new Jackson2JsonEncoder(objectMapper, supportedTypes)); + }; + } + + } + +} diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/spring.factories b/spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/spring.factories index e22fb697c9..2ccb0d34e9 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/spring.factories +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/spring.factories @@ -99,6 +99,9 @@ org.springframework.boot.autoconfigure.mustache.MustacheAutoConfiguration,\ org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration,\ org.springframework.boot.autoconfigure.quartz.QuartzAutoConfiguration,\ org.springframework.boot.autoconfigure.reactor.core.ReactorCoreAutoConfiguration,\ +org.springframework.boot.autoconfigure.rsocket.RSocketMessagingAutoConfiguration,\ +org.springframework.boot.autoconfigure.rsocket.RSocketServerAutoConfiguration,\ +org.springframework.boot.autoconfigure.rsocket.RSocketStrategiesAutoConfiguration,\ org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration,\ org.springframework.boot.autoconfigure.security.servlet.SecurityRequestMatcherProviderAutoConfiguration,\ org.springframework.boot.autoconfigure.security.servlet.UserDetailsServiceAutoConfiguration,\ diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketMessagingAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketMessagingAutoConfigurationTests.java new file mode 100644 index 0000000000..8fd67d0e5b --- /dev/null +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketMessagingAutoConfigurationTests.java @@ -0,0 +1,98 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.autoconfigure.rsocket; + +import org.junit.Test; + +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.codec.CharSequenceEncoder; +import org.springframework.core.codec.StringDecoder; +import org.springframework.messaging.rsocket.MessageHandlerAcceptor; +import org.springframework.messaging.rsocket.RSocketStrategies; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link RSocketMessagingAutoConfiguration}. + * + * @author Brian Clozel + */ +public class RSocketMessagingAutoConfigurationTests { + + private ApplicationContextRunner contextRunner = new ApplicationContextRunner() + .withConfiguration( + AutoConfigurations.of(RSocketMessagingAutoConfiguration.class)) + .withUserConfiguration(BaseConfiguration.class); + + @Test + public void shouldCreateDefaultBeans() { + this.contextRunner.run((context) -> assertThat(context) + .getBeans(MessageHandlerAcceptor.class).hasSize(1)); + } + + @Test + public void shouldFailOnMissingStrategies() { + new ApplicationContextRunner() + .withConfiguration( + AutoConfigurations.of(RSocketMessagingAutoConfiguration.class)) + .run((context) -> { + assertThat(context).hasFailed(); + assertThat(context.getStartupFailure().getMessage()) + .contains("No qualifying bean of type " + + "'org.springframework.messaging.rsocket.RSocketStrategies' available"); + }); + } + + @Test + public void shouldUseCustomMessageHandlerAcceptor() { + this.contextRunner.withUserConfiguration(CustomMessageHandlerAcceptor.class) + .run((context) -> assertThat(context) + .getBeanNames(MessageHandlerAcceptor.class) + .containsOnly("customMessageHandlerAcceptor")); + } + + @Configuration + static class BaseConfiguration { + + @Bean + public RSocketStrategies rSocketStrategies() { + return RSocketStrategies.builder() + .encoder(CharSequenceEncoder.textPlainOnly()) + .decoder(StringDecoder.textPlainOnly()).build(); + } + + } + + @Configuration + static class CustomMessageHandlerAcceptor { + + @Bean + public MessageHandlerAcceptor customMessageHandlerAcceptor() { + MessageHandlerAcceptor acceptor = new MessageHandlerAcceptor(); + RSocketStrategies strategies = RSocketStrategies.builder() + .encoder(CharSequenceEncoder.textPlainOnly()) + .decoder(StringDecoder.textPlainOnly()).build(); + acceptor.setRSocketStrategies(strategies); + return acceptor; + } + + } + +} 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 new file mode 100644 index 0000000000..d44fd549a9 --- /dev/null +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketServerAutoConfigurationTests.java @@ -0,0 +1,119 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.autoconfigure.rsocket; + +import org.junit.Test; + +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.rsocket.netty.NettyRSocketBootstrap; +import org.springframework.boot.rsocket.server.RSocketServerFactory; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner; +import org.springframework.boot.web.server.WebServerFactoryCustomizer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.codec.CharSequenceEncoder; +import org.springframework.core.codec.StringDecoder; +import org.springframework.messaging.rsocket.MessageHandlerAcceptor; +import org.springframework.messaging.rsocket.RSocketStrategies; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link RSocketServerAutoConfiguration}. + * + * @author Brian Clozel + */ +public class RSocketServerAutoConfigurationTests { + + @Test + public void shouldNotCreateBeansByDefault() { + ApplicationContextRunner contextRunner = createContextRunner(); + contextRunner.run((context) -> assertThat(context) + .doesNotHaveBean(WebServerFactoryCustomizer.class) + .doesNotHaveBean(RSocketServerFactory.class) + .doesNotHaveBean(NettyRSocketBootstrap.class)); + } + + @Test + public void shouldNotCreateDefaultBeansForReactiveWebAppWithoutMapping() { + ReactiveWebApplicationContextRunner contextRunner = createReactiveWebContextRunner(); + contextRunner.run((context) -> assertThat(context) + .doesNotHaveBean(WebServerFactoryCustomizer.class) + .doesNotHaveBean(RSocketServerFactory.class) + .doesNotHaveBean(NettyRSocketBootstrap.class)); + } + + @Test + public void shouldNotCreateDefaultBeansForReactiveWebAppWithWrongTransport() { + ReactiveWebApplicationContextRunner contextRunner = createReactiveWebContextRunner(); + contextRunner + .withPropertyValues("spring.rsocket.server.transport=tcp", + "spring.rsocket.server.mapping-path=/rsocket") + .run((context) -> assertThat(context) + .doesNotHaveBean(WebServerFactoryCustomizer.class) + .doesNotHaveBean(RSocketServerFactory.class) + .doesNotHaveBean(NettyRSocketBootstrap.class)); + } + + @Test + public void shouldCreateDefaultBeansForReactiveWebApp() { + ReactiveWebApplicationContextRunner contextRunner = createReactiveWebContextRunner(); + contextRunner + .withPropertyValues("spring.rsocket.server.transport=websocket", + "spring.rsocket.server.mapping-path=/rsocket") + .run((context) -> assertThat(context) + .getBeanNames(WebServerFactoryCustomizer.class).hasSize(1) + .containsOnly("rSocketWebsocketCustomizer")); + } + + @Test + public void shouldCreateDefaultBeansForRSocketServerWhenPortIsSet() { + ReactiveWebApplicationContextRunner contextRunner = createReactiveWebContextRunner(); + contextRunner.withPropertyValues("spring.rsocket.server.port=0") + .run((context) -> assertThat(context) + .hasSingleBean(RSocketServerFactory.class) + .hasSingleBean(NettyRSocketBootstrap.class)); + } + + private ApplicationContextRunner createContextRunner() { + return new ApplicationContextRunner() + .withUserConfiguration(BaseConfiguration.class).withConfiguration( + AutoConfigurations.of(RSocketServerAutoConfiguration.class)); + } + + private ReactiveWebApplicationContextRunner createReactiveWebContextRunner() { + return new ReactiveWebApplicationContextRunner() + .withUserConfiguration(BaseConfiguration.class).withConfiguration( + AutoConfigurations.of(RSocketServerAutoConfiguration.class)); + } + + @Configuration + static class BaseConfiguration { + + @Bean + public MessageHandlerAcceptor messageHandlerAcceptor() { + MessageHandlerAcceptor messageHandlerAcceptor = new MessageHandlerAcceptor(); + messageHandlerAcceptor.setRSocketStrategies(RSocketStrategies.builder() + .encoder(CharSequenceEncoder.textPlainOnly()) + .decoder(StringDecoder.textPlainOnly()).build()); + return messageHandlerAcceptor; + } + + } + +} diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketStrategiesAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketStrategiesAutoConfigurationTests.java new file mode 100644 index 0000000000..55545b6d12 --- /dev/null +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/rsocket/RSocketStrategiesAutoConfigurationTests.java @@ -0,0 +1,114 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.autoconfigure.rsocket; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.Test; + +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.rsocket.messaging.RSocketStrategiesCustomizer; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.codec.CharSequenceEncoder; +import org.springframework.core.codec.StringDecoder; +import org.springframework.http.codec.json.Jackson2JsonDecoder; +import org.springframework.http.codec.json.Jackson2JsonEncoder; +import org.springframework.messaging.rsocket.RSocketStrategies; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link RSocketStrategiesAutoConfiguration} + * + * @author Brian Clozel + */ +public class RSocketStrategiesAutoConfigurationTests { + + private ApplicationContextRunner contextRunner = new ApplicationContextRunner() + .withUserConfiguration(BaseConfiguration.class).withConfiguration( + AutoConfigurations.of(RSocketStrategiesAutoConfiguration.class)); + + @Test + public void shouldCreateDefaultBeans() { + this.contextRunner.run((context) -> { + assertThat(context).getBeans(RSocketStrategies.class).hasSize(1); + RSocketStrategies strategies = context.getBean(RSocketStrategies.class); + assertThat(strategies.decoders()).hasSize(1) + .hasOnlyElementsOfType(Jackson2JsonDecoder.class); + assertThat(strategies.encoders()).hasSize(1) + .hasOnlyElementsOfType(Jackson2JsonEncoder.class); + }); + } + + @Test + public void shouldUseCustomStrategies() { + this.contextRunner.withUserConfiguration(UserStrategies.class).run((context) -> { + assertThat(context).getBeans(RSocketStrategies.class).hasSize(1); + assertThat(context.getBeanNamesForType(RSocketStrategies.class)) + .contains("customRSocketStrategies"); + }); + } + + @Test + public void shouldUseStrategiesCustomizer() { + this.contextRunner.withUserConfiguration(StrategiesCustomizer.class) + .run((context) -> { + assertThat(context).getBeans(RSocketStrategies.class).hasSize(1); + RSocketStrategies strategies = context + .getBean(RSocketStrategies.class); + assertThat(strategies.decoders()).hasSize(2) + .hasAtLeastOneElementOfType(StringDecoder.class); + assertThat(strategies.encoders()).hasSize(2) + .hasAtLeastOneElementOfType(CharSequenceEncoder.class); + }); + } + + @Configuration + static class BaseConfiguration { + + @Bean + public ObjectMapper objectMapper() { + return new ObjectMapper(); + } + + } + + @Configuration + static class UserStrategies { + + @Bean + public RSocketStrategies customRSocketStrategies() { + return RSocketStrategies.builder() + .encoder(CharSequenceEncoder.textPlainOnly()) + .decoder(StringDecoder.textPlainOnly()).build(); + } + + } + + @Configuration + static class StrategiesCustomizer { + + @Bean + public RSocketStrategiesCustomizer myCustomizer() { + return (strategies) -> strategies.encoder(CharSequenceEncoder.textPlainOnly()) + .decoder(StringDecoder.textPlainOnly()); + } + + } + +} diff --git a/spring-boot-project/spring-boot-dependencies/pom.xml b/spring-boot-project/spring-boot-dependencies/pom.xml index 3e221b96df..22e847e3ae 100644 --- a/spring-boot-project/spring-boot-dependencies/pom.xml +++ b/spring-boot-project/spring-boot-dependencies/pom.xml @@ -166,6 +166,7 @@ Californium-SR6 3.3.0 1.0.2 + 0.12.1-RC3 1.3.8 1.2.1 2.2.8 @@ -509,6 +510,11 @@ spring-boot-starter-oauth2-resource-server ${revision} + + org.springframework.boot + spring-boot-starter-quartz + ${revision} + org.springframework.boot spring-boot-starter-reactor-netty @@ -516,7 +522,7 @@ org.springframework.boot - spring-boot-starter-quartz + spring-boot-starter-rsocket ${revision} @@ -1067,6 +1073,16 @@ xml-path ${rest-assured.version} + + io.rsocket + rsocket-core + ${rsocket.version} + + + io.rsocket + rsocket-transport-netty + ${rsocket.version} + io.searchbox jest diff --git a/spring-boot-project/spring-boot-docs/src/main/asciidoc/appendix/application-properties.adoc b/spring-boot-project/spring-boot-docs/src/main/asciidoc/appendix/application-properties.adoc index 40be326e29..56e6cb95c2 100644 --- a/spring-boot-project/spring-boot-docs/src/main/asciidoc/appendix/application-properties.adoc +++ b/spring-boot-project/spring-boot-docs/src/main/asciidoc/appendix/application-properties.adoc @@ -63,6 +63,10 @@ include::{generated-resources-root}/config-docs/server.adoc[] include::{generated-resources-root}/config-docs/security.adoc[] +=== RSocket properties + +include::../../../target/generated-resources/config-docs/rsocket.adoc[] + === Actuator properties include::{generated-resources-root}/config-docs/actuator.adoc[] diff --git a/spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc b/spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc index 16a8dee544..6bca9b3442 100644 --- a/spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc +++ b/spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc @@ -3437,6 +3437,83 @@ both clients and servers. You can learn more about the resource configuration on the client side in the <>. +[[boot-features-rsocket]] +== RSocket + +https://rsocket.io[RSocket] is a binary protocol for use on byte stream transports. +It enables symmetric interaction models via async message passing over a single connection. + +Spring Framework, with the Spring Messaging module supports RSocket both on the server and +the client side. On the server side, it lets you create special `@Controller` beans +to handle incoming RSocket messages. +Methods in your controller are mapped to RSocket routes by using `@MessageMapping` annotations. + +The following code shows a typical `@Controller`: + +[source,java,indent=0] +---- + @Controller + public class MyRSocketController { + + @MessageMapping("chat.room.{name}") + public Flux enterChatRoom(@DestinationVariable String chatRoom, + Flux messages) { + // ... + } + + @MessageMapping("users.{user}.info") + Mono getUserInfo(@DestinationVariable String user) { + // ... + } + + } +---- + +[[boot-features-rsocket-server-auto-configuration]] +=== RSocket server Auto-configuration +Spring Boot provides auto-configuration for RSocket servers. The required dependencies +are provided by the `spring-boot-starter-rsocket`. + +Spring Boot will start an RSocket server as a new embedded server in your application, +or will plug the RSocket infrastructure into an existing reactive Web server. This +depends on the type of application and its configuration. + +In case of a WebFlux application (i.e. of type `WebApplicationType.REACTIVE`), the +RSocket server will be plugged into the existing Web Server only if the following +properties match: + +[source,properties,indent=0,subs="verbatim,quotes,attributes"] +---- + spring.rsocket.server.mapping-path=/rsocket # a mapping path is defined + spring.rsocket.server.transport=websocket # websocket is chosen as a transport + #spring.rsocket.server.port= # no port is defined +---- + +The only other way to create an RSocket server is to start an independent, embedded +RSocket server. Besides the dependency requirements, the only required configuration +is to define a port for that server: + +[source,properties,indent=0,subs="verbatim,quotes,attributes"] +---- + spring.rsocket.server.port=9898 # the only required configuration + spring.rsocket.server.transport=tcp # you're free to configure other properties +---- + + + +[[boot-features-rsocket-messaging]] +=== Spring Messaging RSocket support +Spring Boot will auto-configure the Spring Messaging infrastructure for RSocket. + +An `RSocketStrategies` bean is created to provide encoding and decoding support +for RSocket messages. By default, Spring Boot will try to auto-configure JSON +support with Jackson for `application/json` and `"application/*+json"` media types. +Check out the <> to know more +about customization possibilities. + +Developers can create `RSocketStrategiesCustomizer` beans to add other strategies, +assuming there are `Encoder` and `Decoder` implementations available. + [[boot-features-security]] diff --git a/spring-boot-project/spring-boot-docs/src/main/groovy/generateConfigurationPropertyTables.groovy b/spring-boot-project/spring-boot-docs/src/main/groovy/generateConfigurationPropertyTables.groovy index 183f68114c..56313059b3 100644 --- a/spring-boot-project/spring-boot-docs/src/main/groovy/generateConfigurationPropertyTables.groovy +++ b/spring-boot-project/spring-boot-docs/src/main/groovy/generateConfigurationPropertyTables.groovy @@ -40,6 +40,8 @@ def generateConfigMetadataDocumentation() { "spring.mvc", "spring.resources", "spring.webflux") .addSection("json") .withKeyPrefixes("spring.jackson", "spring.gson") + .addSection("rsocket") + .withKeyPrefixes("spring.rsocket") .addSection("templating") .withKeyPrefixes("spring.freemarker", "spring.groovy", "spring.mustache", "spring.thymeleaf") .addOverride("spring.groovy.template.configuration", "See GroovyMarkupConfigurer") diff --git a/spring-boot-project/spring-boot-starters/pom.xml b/spring-boot-project/spring-boot-starters/pom.xml index 023315e517..6b96374e81 100644 --- a/spring-boot-project/spring-boot-starters/pom.xml +++ b/spring-boot-project/spring-boot-starters/pom.xml @@ -61,6 +61,7 @@ spring-boot-starter-parent spring-boot-starter-quartz spring-boot-starter-reactor-netty + spring-boot-starter-rsocket spring-boot-starter-security spring-boot-starter-test spring-boot-starter-thymeleaf diff --git a/spring-boot-project/spring-boot-starters/spring-boot-starter-rsocket/pom.xml b/spring-boot-project/spring-boot-starters/spring-boot-starter-rsocket/pom.xml new file mode 100644 index 0000000000..0d685878de --- /dev/null +++ b/spring-boot-project/spring-boot-starters/spring-boot-starter-rsocket/pom.xml @@ -0,0 +1,58 @@ + + + + + 4.0.0 + + org.springframework.boot + spring-boot-starters + ${revision} + + spring-boot-starter-rsocket + Spring Boot RSocket Starter + Starter for building RSocket clients and servers. + + ${basedir}/../../.. + + + + org.springframework.boot + spring-boot-starter + + + org.springframework + spring-messaging + + + io.rsocket + rsocket-core + + + io.rsocket + rsocket-transport-netty + + + org.springframework.boot + spring-boot-starter-reactor-netty + + + org.springframework.boot + spring-boot-starter-json + + + diff --git a/spring-boot-project/spring-boot/pom.xml b/spring-boot-project/spring-boot/pom.xml index d356e354ea..978a6a0d6c 100644 --- a/spring-boot-project/spring-boot/pom.xml +++ b/spring-boot-project/spring-boot/pom.xml @@ -81,6 +81,16 @@ netty-tcnative-boringssl-static true + + io.rsocket + rsocket-core + true + + + io.rsocket + rsocket-transport-netty + true + io.undertow undertow-servlet @@ -258,6 +268,11 @@ slf4j-api true + + org.springframework + spring-messaging + true + org.springframework spring-orm 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 new file mode 100644 index 0000000000..1472e00712 --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/context/RSocketServerInitializedEvent.java @@ -0,0 +1,54 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.context; + +import org.springframework.boot.rsocket.server.RSocketServer; +import org.springframework.context.ApplicationEvent; + +/** + * Event to be published after the application context is refreshed and the + * {@link RSocketServer} is ready. Useful for obtaining the local port of a running + * server. + * + * @author Brian Clozel + * @since 2.2.0 + */ +@SuppressWarnings("serial") +public class RSocketServerInitializedEvent extends ApplicationEvent { + + public RSocketServerInitializedEvent(RSocketServer rSocketServer) { + super(rSocketServer); + } + + /** + * Access the {@link RSocketServer}. + * @return the embedded RSocket server + */ + public RSocketServer getrSocketServer() { + return getSource(); + } + + /** + * Access the source of the event (an {@link RSocketServer}). + * @return the embedded web server + */ + @Override + public RSocketServer getSource() { + return (RSocketServer) super.getSource(); + } + +} diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/messaging/RSocketStrategiesCustomizer.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/messaging/RSocketStrategiesCustomizer.java new file mode 100644 index 0000000000..749c7d99a1 --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/messaging/RSocketStrategiesCustomizer.java @@ -0,0 +1,37 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.messaging; + +import org.springframework.messaging.rsocket.RSocketStrategies; + +/** + * Callback interface that can be used to customize codecs configuration for an RSocket + * client and/or server with {@link RSocketStrategies}. + * + * @author Brian Clozel + * @since 2.2.0 + */ +@FunctionalInterface +public interface RSocketStrategiesCustomizer { + + /** + * Callback to customize a {@link RSocketStrategies.Builder} instance. + * @param strategies rSocket codec strategies to customize + */ + void customize(RSocketStrategies.Builder strategies); + +} diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketBootstrap.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketBootstrap.java new file mode 100644 index 0000000000..0c97dcb9c4 --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketBootstrap.java @@ -0,0 +1,72 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.netty; + +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.messaging.rsocket.MessageHandlerAcceptor; + +/** + * Bootstrap an {@link RSocketServer} and start it with the application context. + * + * @author Brian Clozel + * @since 2.2.0 + */ +public class NettyRSocketBootstrap + implements ApplicationEventPublisherAware, SmartLifecycle { + + private final RSocketServer rSocketServer; + + private ApplicationEventPublisher applicationEventPublisher; + + public NettyRSocketBootstrap(RSocketServerFactory serverFactoryProvider, + MessageHandlerAcceptor messageHandlerAcceptorProvider) { + this.rSocketServer = serverFactoryProvider.create(messageHandlerAcceptorProvider); + } + + @Override + public void setApplicationEventPublisher( + ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + @Override + public void start() { + this.rSocketServer.start(); + this.applicationEventPublisher + .publishEvent(new RSocketServerInitializedEvent(this.rSocketServer)); + } + + @Override + public void stop() { + this.rSocketServer.stop(); + } + + @Override + public boolean isRunning() { + RSocketServer server = this.rSocketServer; + if (server != null) { + return server.address() != null; + } + return false; + } + +} 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 new file mode 100644 index 0000000000..e8ffb00987 --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServer.java @@ -0,0 +1,96 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.netty; + +import java.net.InetSocketAddress; +import java.time.Duration; + +import io.rsocket.transport.netty.server.CloseableChannel; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import reactor.core.publisher.Mono; + +import org.springframework.boot.rsocket.server.RSocketServer; +import org.springframework.boot.rsocket.server.RSocketServerException; +import org.springframework.util.Assert; + +/** + * {@link RSocketServer} that is based on a Reactor Netty server. Usually this class + * should be created using the {@link NettyRSocketServerFactory} and not directly. + * + * @author Brian Clozel + * @since 2.2.0 + */ +public class NettyRSocketServer implements RSocketServer { + + private static final Log logger = LogFactory.getLog(NettyRSocketServer.class); + + private final Mono starter; + + private final Duration lifecycleTimeout; + + private CloseableChannel channel; + + public NettyRSocketServer(Mono starter, Duration lifecycleTimeout) { + Assert.notNull(starter, "starter must not be null"); + this.starter = starter; + this.lifecycleTimeout = lifecycleTimeout; + } + + @Override + public InetSocketAddress address() { + if (this.channel != null) { + return this.channel.address(); + } + return null; + } + + @Override + public void start() throws RSocketServerException { + if (this.lifecycleTimeout != null) { + this.channel = this.starter.block(this.lifecycleTimeout); + } + else { + this.channel = this.starter.block(); + } + 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(); + } + + }; + awaitThread.setContextClassLoader(getClass().getClassLoader()); + awaitThread.setDaemon(false); + awaitThread.start(); + } + + @Override + public void stop() throws RSocketServerException { + if (this.channel != null) { + this.channel.dispose(); + this.channel = null; + } + } + +} 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 new file mode 100644 index 0000000000..f559e5aa1c --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactory.java @@ -0,0 +1,165 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.netty; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; + +import io.rsocket.RSocketFactory; +import io.rsocket.SocketAcceptor; +import io.rsocket.transport.ServerTransport; +import io.rsocket.transport.netty.server.CloseableChannel; +import io.rsocket.transport.netty.server.TcpServerTransport; +import io.rsocket.transport.netty.server.WebsocketServerTransport; +import reactor.core.publisher.Mono; +import reactor.netty.http.server.HttpServer; +import reactor.netty.tcp.TcpServer; + +import org.springframework.boot.rsocket.server.ConfigurableRSocketServerFactory; +import org.springframework.boot.rsocket.server.RSocketServer; +import org.springframework.boot.rsocket.server.RSocketServerFactory; +import org.springframework.boot.rsocket.server.ServerRSocketFactoryCustomizer; +import org.springframework.http.client.reactive.ReactorResourceFactory; +import org.springframework.util.Assert; + +/** + * {@link RSocketServerFactory} that can be used to create {@link RSocketServer}s backed + * by Netty. + * + * @author Brian Clozel + * @since 2.2.0 + */ +public class NettyRSocketServerFactory + implements RSocketServerFactory, ConfigurableRSocketServerFactory { + + private int port = 9898; + + private InetAddress address; + + private RSocketServer.TRANSPORT transport = RSocketServer.TRANSPORT.TCP; + + private ReactorResourceFactory resourceFactory; + + private Duration lifecycleTimeout; + + private List serverCustomizers = new ArrayList<>(); + + @Override + public void setPort(int port) { + this.port = port; + } + + @Override + public void setAddress(InetAddress address) { + this.address = address; + } + + @Override + public void setTransport(RSocketServer.TRANSPORT transport) { + this.transport = transport; + } + + /** + * Set the {@link ReactorResourceFactory} to get the shared resources from. + * @param resourceFactory the server resources + */ + public void setResourceFactory(ReactorResourceFactory resourceFactory) { + this.resourceFactory = resourceFactory; + } + + /** + * Set {@link ServerRSocketFactoryCustomizer}s that should be applied to the RSocket + * server builder. Calling this method will replace any existing customizers. + * @param serverCustomizers the customizers to set + */ + public void setServerCustomizers( + Collection serverCustomizers) { + Assert.notNull(serverCustomizers, "ServerCustomizers must not be null"); + this.serverCustomizers = new ArrayList<>(serverCustomizers); + } + + /** + * Add {@link ServerRSocketFactoryCustomizer}s that should applied while building the + * server. + * @param serverCustomizers the customizers to add + */ + public void addServerCustomizers( + ServerRSocketFactoryCustomizer... serverCustomizers) { + Assert.notNull(serverCustomizers, "ServerCustomizer must not be null"); + this.serverCustomizers.addAll(Arrays.asList(serverCustomizers)); + } + + /** + * Set the maximum amount of time that should be waited when starting or stopping the + * server. + * @param lifecycleTimeout the lifecycle timeout + */ + public void setLifecycleTimeout(Duration lifecycleTimeout) { + this.lifecycleTimeout = lifecycleTimeout; + } + + @Override + public NettyRSocketServer create(SocketAcceptor socketAcceptor) { + ServerTransport transport = createTransport(); + RSocketFactory.ServerRSocketFactory factory = RSocketFactory.receive(); + for (ServerRSocketFactoryCustomizer customizer : this.serverCustomizers) { + factory = customizer.apply(factory); + } + Mono starter = factory.acceptor(socketAcceptor) + .transport(transport).start(); + return new NettyRSocketServer(starter, this.lifecycleTimeout); + } + + private ServerTransport createTransport() { + if (this.transport == RSocketServer.TRANSPORT.WEBSOCKET) { + if (this.resourceFactory != null) { + HttpServer httpServer = HttpServer.create() + .tcpConfiguration((tcpServer) -> tcpServer + .runOn(this.resourceFactory.getLoopResources())); + return WebsocketServerTransport.create(httpServer); + } + else { + return WebsocketServerTransport.create(getListenAddress()); + } + } + 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()); + + } + } + } + + private InetSocketAddress getListenAddress() { + if (this.address != null) { + return new InetSocketAddress(this.address.getHostAddress(), this.port); + } + return new InetSocketAddress(this.port); + } + +} 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 new file mode 100644 index 0000000000..a6bfe201f3 --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ConfigurableRSocketServerFactory.java @@ -0,0 +1,48 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.server; + +import java.net.InetAddress; + +/** + * A configurable {@link RSocketServerFactory}. + * + * @author Brian Clozel + * @since 2.2.0 + */ +public interface ConfigurableRSocketServerFactory { + + /** + * Set the port that the server should listen on. If not specified port '9898' will be + * used. + * @param port the port to set + */ + void setPort(int port); + + /** + * Set the specific network address that the server should bind to. + * @param address the address to set (defaults to {@code null}) + */ + void setAddress(InetAddress address); + + /** + * Set the transport that the RSocket server should use. + * @param transport the transport protocol to use + */ + 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 new file mode 100644 index 0000000000..768576353b --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServer.java @@ -0,0 +1,59 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.server; + +import java.net.InetSocketAddress; + +/** + * Simple interface that represents a fully configured RSocket server. Allows the server + * to be {@link #start() started} and {@link #stop() stopped}. + * + * @author Brian Clozel + * @since 2.2.0 + */ +public interface RSocketServer { + + /** + * Starts the RSocket server. Calling this method on an already started server has no + * effect. + * @throws RSocketServerException if the server cannot be started + */ + void start() throws RSocketServerException; + + /** + * Stops the RSocket server. Calling this method on an already stopped server has no + * effect. + * @throws RSocketServerException if the server cannot be stopped + */ + void stop() throws RSocketServerException; + + /** + * Return the address this server is listening on. + * @return the address + */ + InetSocketAddress address(); + + /** + * Choice of transport protocol for the RSocket server. + */ + enum TRANSPORT { + + TCP, 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 new file mode 100644 index 0000000000..b0c9dc84ea --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerException.java @@ -0,0 +1,32 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.server; + +/** + * Exceptions thrown by an RSocket server. + * + * @author Brian Clozel + * @since 2.2.0 + */ +@SuppressWarnings("serial") +public class RSocketServerException extends RuntimeException { + + public RSocketServerException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerFactory.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerFactory.java new file mode 100644 index 0000000000..ff5a2f13f6 --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/RSocketServerFactory.java @@ -0,0 +1,41 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.server; + +import io.rsocket.SocketAcceptor; + +/** + * Factory interface that can be used to create a reactive {@link RSocketServer}. + * + * @author Brian Clozel + * @since 2.2.0 + */ +@FunctionalInterface +public interface RSocketServerFactory { + + /** + * Gets a new fully configured but paused {@link RSocketServer} instance. Clients + * should not be able to connect to the returned server until + * {@link RSocketServer#start()} is called (which happens when the + * {@code ApplicationContext} has been fully refreshed). + * @param socketAcceptor the socket acceptor + * @return a fully configured and started {@link RSocketServer} + * @see RSocketServer#stop() + */ + RSocketServer create(SocketAcceptor socketAcceptor); + +} diff --git a/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ServerRSocketFactoryCustomizer.java b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ServerRSocketFactoryCustomizer.java new file mode 100644 index 0000000000..bf40e2a839 --- /dev/null +++ b/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ServerRSocketFactoryCustomizer.java @@ -0,0 +1,34 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.server; + +import java.util.function.Function; + +import io.rsocket.RSocketFactory; + +/** + * Mapping function that can be used to customize an RSocket server factory. + * + * @author Brian Clozel + * @see RSocketServerFactory + * @since 2.2.0 + */ +@FunctionalInterface +public interface ServerRSocketFactoryCustomizer extends + Function { + +} 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 new file mode 100644 index 0000000000..fda11d6e7e --- /dev/null +++ b/spring-boot-project/spring-boot/src/test/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactoryTests.java @@ -0,0 +1,179 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.boot.rsocket.netty; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.Arrays; + +import io.netty.buffer.PooledByteBufAllocator; +import io.rsocket.AbstractRSocket; +import io.rsocket.ConnectionSetupPayload; +import io.rsocket.Payload; +import io.rsocket.RSocket; +import io.rsocket.RSocketFactory; +import io.rsocket.SocketAcceptor; +import io.rsocket.transport.netty.client.TcpClientTransport; +import io.rsocket.transport.netty.client.WebsocketClientTransport; +import io.rsocket.util.DefaultPayload; +import org.assertj.core.api.Assertions; +import org.junit.After; +import org.junit.Rule; +import org.junit.Test; +import org.mockito.InOrder; +import reactor.core.publisher.Mono; + +import org.springframework.boot.rsocket.server.RSocketServer; +import org.springframework.boot.rsocket.server.ServerRSocketFactoryCustomizer; +import org.springframework.boot.testsupport.rule.OutputCapture; +import org.springframework.core.codec.CharSequenceEncoder; +import org.springframework.core.codec.StringDecoder; +import org.springframework.core.io.buffer.NettyDataBufferFactory; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.util.MimeTypeUtils; +import org.springframework.util.SocketUtils; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; + +/** + * Tests for {@link NettyRSocketServerFactory} + * + * @author Brian Clozel + */ +public class NettyRSocketServerFactoryTests { + + @Rule + public OutputCapture output = new OutputCapture(); + + private NettyRSocketServer rSocketServer; + + private RSocketRequester requester; + + private static final Duration TIMEOUT = Duration.ofSeconds(3); + + @After + public void tearDown() { + if (this.rSocketServer != null) { + try { + this.rSocketServer.stop(); + } + catch (Exception ex) { + // Ignore + } + } + if (this.requester != null) { + this.requester.rsocket().dispose(); + } + } + + private NettyRSocketServerFactory getFactory() { + return new NettyRSocketServerFactory(); + } + + @Test + public void specificPort() { + NettyRSocketServerFactory factory = getFactory(); + int specificPort = SocketUtils.findAvailableTcpPort(41000); + factory.setPort(specificPort); + this.rSocketServer = factory.create(new EchoRequestResponseAcceptor()); + this.rSocketServer.start(); + this.requester = getRSocketRequester(createRSocketTcpClient()); + 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); + } + + @Test + public void websocketTransport() { + NettyRSocketServerFactory factory = getFactory(); + factory.setTransport(RSocketServer.TRANSPORT.WEBSOCKET); + this.rSocketServer = factory.create(new EchoRequestResponseAcceptor()); + this.rSocketServer.start(); + this.requester = getRSocketRequester(createRSocketWebSocketClient()); + String payload = "test payload"; + String response = this.requester.route("test").data(payload) + .retrieveMono(String.class).block(TIMEOUT); + assertThat(response).isEqualTo(payload); + } + + @Test + public void serverCustomizers() { + NettyRSocketServerFactory factory = getFactory(); + ServerRSocketFactoryCustomizer[] customizers = new ServerRSocketFactoryCustomizer[2]; + for (int i = 0; i < customizers.length; i++) { + customizers[i] = mock(ServerRSocketFactoryCustomizer.class); + given(customizers[i].apply(any(RSocketFactory.ServerRSocketFactory.class))) + .will((invocation) -> invocation.getArgument(0)); + } + factory.setServerCustomizers(Arrays.asList(customizers[0], customizers[1])); + this.rSocketServer = factory.create(new EchoRequestResponseAcceptor()); + InOrder ordered = inOrder((Object[]) customizers); + for (ServerRSocketFactoryCustomizer customizer : customizers) { + ordered.verify(customizer) + .apply(any(RSocketFactory.ServerRSocketFactory.class)); + } + } + + private RSocket createRSocketTcpClient() { + Assertions.assertThat(this.rSocketServer).isNotNull(); + InetSocketAddress address = this.rSocketServer.address(); + return RSocketFactory.connect().dataMimeType(MimeTypeUtils.TEXT_PLAIN_VALUE) + .transport(TcpClientTransport.create(address)).start().block(); + } + + private RSocket createRSocketWebSocketClient() { + Assertions.assertThat(this.rSocketServer).isNotNull(); + InetSocketAddress address = this.rSocketServer.address(); + return RSocketFactory.connect().dataMimeType(MimeTypeUtils.TEXT_PLAIN_VALUE) + .transport(WebsocketClientTransport.create(address)).start().block(); + } + + private RSocketRequester getRSocketRequester(RSocket rSocketClient) { + RSocketStrategies strategies = RSocketStrategies.builder() + .decoder(StringDecoder.allMimeTypes()) + .encoder(CharSequenceEncoder.allMimeTypes()) + .dataBufferFactory( + new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT)) + .build(); + return RSocketRequester.create(rSocketClient, MimeTypeUtils.TEXT_PLAIN, + strategies); + } + + static class EchoRequestResponseAcceptor implements SocketAcceptor { + + @Override + public Mono accept(ConnectionSetupPayload setupPayload, + RSocket rSocket) { + return Mono.just(new AbstractRSocket() { + @Override + public Mono requestResponse(Payload payload) { + return Mono.just(DefaultPayload.create(payload)); + } + }); + } + + } + +}