Commit dac62476 authored by Brian Clozel's avatar Brian Clozel

Merge branch '2.2.x'

Closes gh-21208
parents 71565a17 a63ab468
/* /*
* Copyright 2012-2019 the original author or authors. * Copyright 2012-2020 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
...@@ -18,7 +18,7 @@ package org.springframework.boot.autoconfigure.rsocket; ...@@ -18,7 +18,7 @@ package org.springframework.boot.autoconfigure.rsocket;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import io.rsocket.RSocketFactory; import io.rsocket.core.RSocketServer;
import io.rsocket.frame.decoder.PayloadDecoder; import io.rsocket.frame.decoder.PayloadDecoder;
import io.rsocket.transport.netty.server.TcpServerTransport; import io.rsocket.transport.netty.server.TcpServerTransport;
import reactor.netty.http.server.HttpServer; import reactor.netty.http.server.HttpServer;
...@@ -36,6 +36,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties ...@@ -36,6 +36,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.boot.rsocket.context.RSocketServerBootstrap; import org.springframework.boot.rsocket.context.RSocketServerBootstrap;
import org.springframework.boot.rsocket.netty.NettyRSocketServerFactory; import org.springframework.boot.rsocket.netty.NettyRSocketServerFactory;
import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.boot.rsocket.server.RSocketServerFactory; import org.springframework.boot.rsocket.server.RSocketServerFactory;
import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor; import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
...@@ -57,7 +58,7 @@ import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHa ...@@ -57,7 +58,7 @@ import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHa
* @since 2.2.0 * @since 2.2.0
*/ */
@Configuration(proxyBeanMethods = false) @Configuration(proxyBeanMethods = false)
@ConditionalOnClass({ RSocketFactory.class, RSocketStrategies.class, HttpServer.class, TcpServerTransport.class }) @ConditionalOnClass({ RSocketServer.class, RSocketStrategies.class, HttpServer.class, TcpServerTransport.class })
@ConditionalOnBean(RSocketMessageHandler.class) @ConditionalOnBean(RSocketMessageHandler.class)
@AutoConfigureAfter(RSocketStrategiesAutoConfiguration.class) @AutoConfigureAfter(RSocketStrategiesAutoConfiguration.class)
@EnableConfigurationProperties(RSocketProperties.class) @EnableConfigurationProperties(RSocketProperties.class)
...@@ -69,10 +70,12 @@ public class RSocketServerAutoConfiguration { ...@@ -69,10 +70,12 @@ public class RSocketServerAutoConfiguration {
@Bean @Bean
@ConditionalOnMissingBean @ConditionalOnMissingBean
@SuppressWarnings("deprecation")
RSocketWebSocketNettyRouteProvider rSocketWebsocketRouteProvider(RSocketProperties properties, RSocketWebSocketNettyRouteProvider rSocketWebsocketRouteProvider(RSocketProperties properties,
RSocketMessageHandler messageHandler, ObjectProvider<ServerRSocketFactoryProcessor> processors) { RSocketMessageHandler messageHandler, ObjectProvider<ServerRSocketFactoryProcessor> processors,
ObjectProvider<RSocketServerCustomizer> customizers) {
return new RSocketWebSocketNettyRouteProvider(properties.getServer().getMappingPath(), return new RSocketWebSocketNettyRouteProvider(properties.getServer().getMappingPath(),
messageHandler.responder(), processors.orderedStream()); messageHandler.responder(), processors.orderedStream(), customizers.orderedStream());
} }
} }
...@@ -89,14 +92,17 @@ public class RSocketServerAutoConfiguration { ...@@ -89,14 +92,17 @@ public class RSocketServerAutoConfiguration {
@Bean @Bean
@ConditionalOnMissingBean @ConditionalOnMissingBean
@SuppressWarnings("deprecation")
RSocketServerFactory rSocketServerFactory(RSocketProperties properties, ReactorResourceFactory resourceFactory, RSocketServerFactory rSocketServerFactory(RSocketProperties properties, ReactorResourceFactory resourceFactory,
ObjectProvider<ServerRSocketFactoryProcessor> processors) { ObjectProvider<ServerRSocketFactoryProcessor> processors,
ObjectProvider<RSocketServerCustomizer> customizers) {
NettyRSocketServerFactory factory = new NettyRSocketServerFactory(); NettyRSocketServerFactory factory = new NettyRSocketServerFactory();
factory.setResourceFactory(resourceFactory); factory.setResourceFactory(resourceFactory);
factory.setTransport(properties.getServer().getTransport()); factory.setTransport(properties.getServer().getTransport());
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties.getServer().getAddress()).to(factory::setAddress); map.from(properties.getServer().getAddress()).to(factory::setAddress);
map.from(properties.getServer().getPort()).to(factory::setPort); map.from(properties.getServer().getPort()).to(factory::setPort);
factory.setRSocketServerCustomizers(customizers.orderedStream().collect(Collectors.toList()));
factory.setSocketFactoryProcessors(processors.orderedStream().collect(Collectors.toList())); factory.setSocketFactoryProcessors(processors.orderedStream().collect(Collectors.toList()));
return factory; return factory;
} }
...@@ -109,13 +115,12 @@ public class RSocketServerAutoConfiguration { ...@@ -109,13 +115,12 @@ public class RSocketServerAutoConfiguration {
} }
@Bean @Bean
ServerRSocketFactoryProcessor frameDecoderServerFactoryCustomizer(RSocketMessageHandler rSocketMessageHandler) { RSocketServerCustomizer frameDecoderRSocketServerCustomizer(RSocketMessageHandler rSocketMessageHandler) {
return (serverRSocketFactory) -> { return (server) -> {
if (rSocketMessageHandler.getRSocketStrategies() if (rSocketMessageHandler.getRSocketStrategies()
.dataBufferFactory() instanceof NettyDataBufferFactory) { .dataBufferFactory() instanceof NettyDataBufferFactory) {
return serverRSocketFactory.frameDecoder(PayloadDecoder.ZERO_COPY); server.payloadDecoder(PayloadDecoder.ZERO_COPY);
} }
return serverRSocketFactory;
}; };
} }
......
/* /*
* Copyright 2012-2019 the original author or authors. * Copyright 2012-2020 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
...@@ -22,10 +22,12 @@ import java.util.stream.Stream; ...@@ -22,10 +22,12 @@ import java.util.stream.Stream;
import io.rsocket.RSocketFactory; import io.rsocket.RSocketFactory;
import io.rsocket.SocketAcceptor; import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.ServerTransport; import io.rsocket.transport.ServerTransport;
import io.rsocket.transport.netty.server.WebsocketRouteTransport; import io.rsocket.transport.netty.server.WebsocketRouteTransport;
import reactor.netty.http.server.HttpServerRoutes; import reactor.netty.http.server.HttpServerRoutes;
import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor; import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor;
import org.springframework.boot.web.embedded.netty.NettyRouteProvider; import org.springframework.boot.web.embedded.netty.NettyRouteProvider;
...@@ -34,6 +36,7 @@ import org.springframework.boot.web.embedded.netty.NettyRouteProvider; ...@@ -34,6 +36,7 @@ import org.springframework.boot.web.embedded.netty.NettyRouteProvider;
* *
* @author Brian Clozel * @author Brian Clozel
*/ */
@SuppressWarnings("deprecation")
class RSocketWebSocketNettyRouteProvider implements NettyRouteProvider { class RSocketWebSocketNettyRouteProvider implements NettyRouteProvider {
private final String mappingPath; private final String mappingPath;
...@@ -42,21 +45,24 @@ class RSocketWebSocketNettyRouteProvider implements NettyRouteProvider { ...@@ -42,21 +45,24 @@ class RSocketWebSocketNettyRouteProvider implements NettyRouteProvider {
private final List<ServerRSocketFactoryProcessor> processors; private final List<ServerRSocketFactoryProcessor> processors;
private final List<RSocketServerCustomizer> customizers;
RSocketWebSocketNettyRouteProvider(String mappingPath, SocketAcceptor socketAcceptor, RSocketWebSocketNettyRouteProvider(String mappingPath, SocketAcceptor socketAcceptor,
Stream<ServerRSocketFactoryProcessor> processors) { Stream<ServerRSocketFactoryProcessor> processors, Stream<RSocketServerCustomizer> customizers) {
this.mappingPath = mappingPath; this.mappingPath = mappingPath;
this.socketAcceptor = socketAcceptor; this.socketAcceptor = socketAcceptor;
this.processors = processors.collect(Collectors.toList()); this.processors = processors.collect(Collectors.toList());
this.customizers = customizers.collect(Collectors.toList());
} }
@Override @Override
public HttpServerRoutes apply(HttpServerRoutes httpServerRoutes) { public HttpServerRoutes apply(HttpServerRoutes httpServerRoutes) {
RSocketFactory.ServerRSocketFactory server = RSocketFactory.receive(); RSocketServer server = RSocketServer.create(this.socketAcceptor);
for (ServerRSocketFactoryProcessor processor : this.processors) { RSocketFactory.ServerRSocketFactory factory = new RSocketFactory.ServerRSocketFactory(server);
server = processor.process(server); this.processors.forEach((processor) -> processor.process(factory));
} this.customizers.forEach((customizer) -> customizer.customize(server));
ServerTransport.ConnectionAcceptor acceptor = server.acceptor(this.socketAcceptor).toConnectionAcceptor(); ServerTransport.ConnectionAcceptor connectionAcceptor = server.asConnectionAcceptor();
return httpServerRoutes.ws(this.mappingPath, WebsocketRouteTransport.newHandler(acceptor)); return httpServerRoutes.ws(this.mappingPath, WebsocketRouteTransport.newHandler(connectionAcceptor));
} }
} }
/* /*
* Copyright 2012-2019 the original author or authors. * Copyright 2012-2020 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
...@@ -18,7 +18,7 @@ package org.springframework.boot.autoconfigure.security.rsocket; ...@@ -18,7 +18,7 @@ package org.springframework.boot.autoconfigure.security.rsocket;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor; import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.rsocket.EnableRSocketSecurity; import org.springframework.security.config.annotation.rsocket.EnableRSocketSecurity;
...@@ -29,6 +29,7 @@ import org.springframework.security.rsocket.core.SecuritySocketAcceptorIntercept ...@@ -29,6 +29,7 @@ import org.springframework.security.rsocket.core.SecuritySocketAcceptorIntercept
* server. * server.
* *
* @author Madhura Bhave * @author Madhura Bhave
* @author Brian Clozel
* @since 2.2.0 * @since 2.2.0
*/ */
@Configuration(proxyBeanMethods = false) @Configuration(proxyBeanMethods = false)
...@@ -37,8 +38,8 @@ import org.springframework.security.rsocket.core.SecuritySocketAcceptorIntercept ...@@ -37,8 +38,8 @@ import org.springframework.security.rsocket.core.SecuritySocketAcceptorIntercept
public class RSocketSecurityAutoConfiguration { public class RSocketSecurityAutoConfiguration {
@Bean @Bean
ServerRSocketFactoryProcessor springSecurityRSocketSecurity(SecuritySocketAcceptorInterceptor interceptor) { RSocketServerCustomizer springSecurityRSocketSecurity(SecuritySocketAcceptorInterceptor interceptor) {
return (factory) -> factory.addSocketAcceptorPlugin(interceptor); return (server) -> server.interceptors((registry) -> registry.forSocketAcceptor(interceptor));
} }
} }
/* /*
* Copyright 2012-2019 the original author or authors. * Copyright 2012-2020 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
...@@ -21,8 +21,8 @@ import org.junit.jupiter.api.Test; ...@@ -21,8 +21,8 @@ import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.rsocket.context.RSocketPortInfoApplicationContextInitializer; import org.springframework.boot.rsocket.context.RSocketPortInfoApplicationContextInitializer;
import org.springframework.boot.rsocket.context.RSocketServerBootstrap; import org.springframework.boot.rsocket.context.RSocketServerBootstrap;
import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.boot.rsocket.server.RSocketServerFactory; import org.springframework.boot.rsocket.server.RSocketServerFactory;
import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor;
import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner; import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner;
import org.springframework.boot.web.server.WebServerFactoryCustomizer; import org.springframework.boot.web.server.WebServerFactoryCustomizer;
...@@ -78,8 +78,7 @@ class RSocketServerAutoConfigurationTests { ...@@ -78,8 +78,7 @@ class RSocketServerAutoConfigurationTests {
void shouldCreateDefaultBeansForRSocketServerWhenPortIsSet() { void shouldCreateDefaultBeansForRSocketServerWhenPortIsSet() {
reactiveWebContextRunner().withPropertyValues("spring.rsocket.server.port=0") reactiveWebContextRunner().withPropertyValues("spring.rsocket.server.port=0")
.run((context) -> assertThat(context).hasSingleBean(RSocketServerFactory.class) .run((context) -> assertThat(context).hasSingleBean(RSocketServerFactory.class)
.hasSingleBean(RSocketServerBootstrap.class) .hasSingleBean(RSocketServerBootstrap.class).hasSingleBean(RSocketServerCustomizer.class));
.hasSingleBean(ServerRSocketFactoryProcessor.class));
} }
@Test @Test
...@@ -87,8 +86,7 @@ class RSocketServerAutoConfigurationTests { ...@@ -87,8 +86,7 @@ class RSocketServerAutoConfigurationTests {
reactiveWebContextRunner().withPropertyValues("spring.rsocket.server.port=0") reactiveWebContextRunner().withPropertyValues("spring.rsocket.server.port=0")
.withInitializer(new RSocketPortInfoApplicationContextInitializer()).run((context) -> { .withInitializer(new RSocketPortInfoApplicationContextInitializer()).run((context) -> {
assertThat(context).hasSingleBean(RSocketServerFactory.class) assertThat(context).hasSingleBean(RSocketServerFactory.class)
.hasSingleBean(RSocketServerBootstrap.class) .hasSingleBean(RSocketServerBootstrap.class).hasSingleBean(RSocketServerCustomizer.class);
.hasSingleBean(ServerRSocketFactoryProcessor.class);
assertThat(context.getEnvironment().getProperty("local.rsocket.server.port")).isNotNull(); assertThat(context.getEnvironment().getProperty("local.rsocket.server.port")).isNotNull();
}); });
} }
......
/* /*
* Copyright 2012-2019 the original author or authors. * Copyright 2012-2020 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
......
/* /*
* Copyright 2012-2019 the original author or authors. * Copyright 2012-2020 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
...@@ -16,28 +16,26 @@ ...@@ -16,28 +16,26 @@
package org.springframework.boot.autoconfigure.security.rsocket; package org.springframework.boot.autoconfigure.security.rsocket;
import io.rsocket.RSocketFactory; import io.rsocket.core.RSocketServer;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.autoconfigure.rsocket.RSocketMessagingAutoConfiguration; import org.springframework.boot.autoconfigure.rsocket.RSocketMessagingAutoConfiguration;
import org.springframework.boot.autoconfigure.rsocket.RSocketStrategiesAutoConfiguration; import org.springframework.boot.autoconfigure.rsocket.RSocketStrategiesAutoConfiguration;
import org.springframework.boot.autoconfigure.security.reactive.ReactiveUserDetailsServiceAutoConfiguration; import org.springframework.boot.autoconfigure.security.reactive.ReactiveUserDetailsServiceAutoConfiguration;
import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor; import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.boot.test.context.FilteredClassLoader; import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.security.config.annotation.rsocket.RSocketSecurity; import org.springframework.security.config.annotation.rsocket.RSocketSecurity;
import org.springframework.security.rsocket.core.SecuritySocketAcceptorInterceptor; import org.springframework.security.rsocket.core.SecuritySocketAcceptorInterceptor;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
/** /**
* Tests for {@link RSocketSecurityAutoConfiguration}. * Tests for {@link RSocketSecurityAutoConfiguration}.
* *
* @author Madhura Bhave * @author Madhura Bhave
* @author Brian Clozel
*/ */
class RSocketSecurityAutoConfigurationTests { class RSocketSecurityAutoConfigurationTests {
...@@ -58,14 +56,15 @@ class RSocketSecurityAutoConfigurationTests { ...@@ -58,14 +56,15 @@ class RSocketSecurityAutoConfigurationTests {
@Test @Test
void autoConfigurationAddsCustomizerForServerRSocketFactory() { void autoConfigurationAddsCustomizerForServerRSocketFactory() {
RSocketFactory.ServerRSocketFactory factory = mock(RSocketFactory.ServerRSocketFactory.class); RSocketServer server = RSocketServer.create();
ArgumentCaptor<SecuritySocketAcceptorInterceptor> captor = ArgumentCaptor
.forClass(SecuritySocketAcceptorInterceptor.class);
this.contextRunner.run((context) -> { this.contextRunner.run((context) -> {
ServerRSocketFactoryProcessor customizer = context.getBean(ServerRSocketFactoryProcessor.class); RSocketServerCustomizer customizer = context.getBean(RSocketServerCustomizer.class);
customizer.process(factory); customizer.customize(server);
verify(factory).addSocketAcceptorPlugin(captor.capture()); server.interceptors((registry) -> registry.forSocketAcceptor((interceptors) -> {
assertThat(captor.getValue()).isInstanceOf(SecuritySocketAcceptorInterceptor.class); assertThat(interceptors).isNotEmpty();
assertThat(interceptors)
.anyMatch((interceptor) -> interceptor instanceof SecuritySocketAcceptorInterceptor);
}));
}); });
} }
......
...@@ -1403,7 +1403,7 @@ bom { ...@@ -1403,7 +1403,7 @@ bom {
] ]
} }
} }
library("RSocket", "1.0.0-RC6") { library("RSocket", "1.0.0-RC7") {
group("io.rsocket") { group("io.rsocket") {
imports = [ imports = [
"rsocket-bom" "rsocket-bom"
......
/* /*
* Copyright 2012-2019 the original author or authors. * Copyright 2012-2020 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
...@@ -37,6 +37,7 @@ import reactor.netty.tcp.TcpServer; ...@@ -37,6 +37,7 @@ import reactor.netty.tcp.TcpServer;
import org.springframework.boot.rsocket.server.ConfigurableRSocketServerFactory; import org.springframework.boot.rsocket.server.ConfigurableRSocketServerFactory;
import org.springframework.boot.rsocket.server.RSocketServer; import org.springframework.boot.rsocket.server.RSocketServer;
import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.boot.rsocket.server.RSocketServerFactory; import org.springframework.boot.rsocket.server.RSocketServerFactory;
import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor; import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor;
import org.springframework.http.client.reactive.ReactorResourceFactory; import org.springframework.http.client.reactive.ReactorResourceFactory;
...@@ -63,6 +64,8 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur ...@@ -63,6 +64,8 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur
private List<ServerRSocketFactoryProcessor> socketFactoryProcessors = new ArrayList<>(); private List<ServerRSocketFactoryProcessor> socketFactoryProcessors = new ArrayList<>();
private List<RSocketServerCustomizer> rSocketServerCustomizers = new ArrayList<>();
@Override @Override
public void setPort(int port) { public void setPort(int port) {
this.port = port; this.port = port;
...@@ -91,7 +94,10 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur ...@@ -91,7 +94,10 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur
* {@link ServerRSocketFactory} while building the server. Calling this method will * {@link ServerRSocketFactory} while building the server. Calling this method will
* replace any existing processors. * replace any existing processors.
* @param socketFactoryProcessors processors to apply before the server starts * @param socketFactoryProcessors processors to apply before the server starts
* @deprecated in favor of {@link #setRSocketServerCustomizers(Collection)} as of
* 2.2.7
*/ */
@Deprecated
public void setSocketFactoryProcessors( public void setSocketFactoryProcessors(
Collection<? extends ServerRSocketFactoryProcessor> socketFactoryProcessors) { Collection<? extends ServerRSocketFactoryProcessor> socketFactoryProcessors) {
Assert.notNull(socketFactoryProcessors, "SocketFactoryProcessors must not be null"); Assert.notNull(socketFactoryProcessors, "SocketFactoryProcessors must not be null");
...@@ -102,12 +108,38 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur ...@@ -102,12 +108,38 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur
* Add {@link ServerRSocketFactoryProcessor}s that should be called to process the * Add {@link ServerRSocketFactoryProcessor}s that should be called to process the
* {@link ServerRSocketFactory} while building the server. * {@link ServerRSocketFactory} while building the server.
* @param socketFactoryProcessors processors to apply before the server starts * @param socketFactoryProcessors processors to apply before the server starts
* @deprecated in favor of
* {@link #addRSocketServerCustomizers(RSocketServerCustomizer...)} as of 2.2.7
*/ */
@Deprecated
public void addSocketFactoryProcessors(ServerRSocketFactoryProcessor... socketFactoryProcessors) { public void addSocketFactoryProcessors(ServerRSocketFactoryProcessor... socketFactoryProcessors) {
Assert.notNull(socketFactoryProcessors, "SocketFactoryProcessors must not be null"); Assert.notNull(socketFactoryProcessors, "SocketFactoryProcessors must not be null");
this.socketFactoryProcessors.addAll(Arrays.asList(socketFactoryProcessors)); this.socketFactoryProcessors.addAll(Arrays.asList(socketFactoryProcessors));
} }
/**
* Set {@link RSocketServerCustomizer}s that should be called to configure the
* {@link io.rsocket.core.RSocketServer} while building the server. Calling this
* method will replace any existing customizers.
* @param rSocketServerCustomizers customizers to apply before the server starts
* @since 2.2.7
*/
public void setRSocketServerCustomizers(Collection<? extends RSocketServerCustomizer> rSocketServerCustomizers) {
Assert.notNull(rSocketServerCustomizers, "RSocketServerCustomizers must not be null");
this.rSocketServerCustomizers = new ArrayList<>(rSocketServerCustomizers);
}
/**
* Add {@link RSocketServerCustomizer}s that should be called to configure the
* {@link io.rsocket.core.RSocketServer}.
* @param rSocketServerCustomizers customizers to apply before the server starts
* @since 2.2.7
*/
public void addRSocketServerCustomizers(RSocketServerCustomizer... rSocketServerCustomizers) {
Assert.notNull(rSocketServerCustomizers, "RSocketServerCustomizers must not be null");
this.rSocketServerCustomizers.addAll(Arrays.asList(rSocketServerCustomizers));
}
/** /**
* Set the maximum amount of time that should be waited when starting or stopping the * Set the maximum amount of time that should be waited when starting or stopping the
* server. * server.
...@@ -118,13 +150,14 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur ...@@ -118,13 +150,14 @@ public class NettyRSocketServerFactory implements RSocketServerFactory, Configur
} }
@Override @Override
@SuppressWarnings("deprecation")
public NettyRSocketServer create(SocketAcceptor socketAcceptor) { public NettyRSocketServer create(SocketAcceptor socketAcceptor) {
ServerTransport<CloseableChannel> transport = createTransport(); ServerTransport<CloseableChannel> transport = createTransport();
RSocketFactory.ServerRSocketFactory factory = RSocketFactory.receive(); io.rsocket.core.RSocketServer server = io.rsocket.core.RSocketServer.create(socketAcceptor);
for (ServerRSocketFactoryProcessor processor : this.socketFactoryProcessors) { RSocketFactory.ServerRSocketFactory factory = new ServerRSocketFactory(server);
factory = processor.process(factory); this.rSocketServerCustomizers.forEach((customizer) -> customizer.customize(server));
} this.socketFactoryProcessors.forEach((processor) -> processor.process(factory));
Mono<CloseableChannel> starter = factory.acceptor(socketAcceptor).transport(transport).start(); Mono<CloseableChannel> starter = server.bind(transport);
return new NettyRSocketServer(starter, this.lifecycleTimeout); return new NettyRSocketServer(starter, this.lifecycleTimeout);
} }
......
/*
* Copyright 2012-2020 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.core.RSocketServer;
/**
* Callback interface that can be used to customize a {@link RSocketServer}.
*
* @author Brian Clozel
* @see RSocketServer
* @since 2.3.0
*/
@FunctionalInterface
public interface RSocketServerCustomizer {
/**
* Callback to customize a {@link RSocketServer} instance.
* @param rSocketServer the RSocket server to customize
*/
void customize(RSocketServer rSocketServer);
}
...@@ -25,8 +25,10 @@ import io.rsocket.RSocketFactory.ServerRSocketFactory; ...@@ -25,8 +25,10 @@ import io.rsocket.RSocketFactory.ServerRSocketFactory;
* @author Brian Clozel * @author Brian Clozel
* @see RSocketServerFactory * @see RSocketServerFactory
* @since 2.2.0 * @since 2.2.0
* @deprecated in favor of {@link RSocketServerCustomizer} as of 2.2.7
*/ */
@FunctionalInterface @FunctionalInterface
@Deprecated
public interface ServerRSocketFactoryProcessor { public interface ServerRSocketFactoryProcessor {
/** /**
......
...@@ -37,6 +37,7 @@ import org.mockito.InOrder; ...@@ -37,6 +37,7 @@ import org.mockito.InOrder;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import org.springframework.boot.rsocket.server.RSocketServer; import org.springframework.boot.rsocket.server.RSocketServer;
import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor; import org.springframework.boot.rsocket.server.ServerRSocketFactoryProcessor;
import org.springframework.core.codec.CharSequenceEncoder; import org.springframework.core.codec.CharSequenceEncoder;
import org.springframework.core.codec.StringDecoder; import org.springframework.core.codec.StringDecoder;
...@@ -49,6 +50,7 @@ import org.springframework.util.SocketUtils; ...@@ -49,6 +50,7 @@ import org.springframework.util.SocketUtils;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.will;
import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
...@@ -130,6 +132,7 @@ class NettyRSocketServerFactoryTests { ...@@ -130,6 +132,7 @@ class NettyRSocketServerFactoryTests {
} }
@Test @Test
@SuppressWarnings("deprecation")
void serverProcessors() { void serverProcessors() {
NettyRSocketServerFactory factory = getFactory(); NettyRSocketServerFactory factory = getFactory();
ServerRSocketFactoryProcessor[] processors = new ServerRSocketFactoryProcessor[2]; ServerRSocketFactoryProcessor[] processors = new ServerRSocketFactoryProcessor[2];
...@@ -146,6 +149,23 @@ class NettyRSocketServerFactoryTests { ...@@ -146,6 +149,23 @@ class NettyRSocketServerFactoryTests {
} }
} }
@Test
void serverCustomizers() {
NettyRSocketServerFactory factory = getFactory();
RSocketServerCustomizer[] customizers = new RSocketServerCustomizer[2];
for (int i = 0; i < customizers.length; i++) {
customizers[i] = mock(RSocketServerCustomizer.class);
will((invocation) -> invocation.getArgument(0)).given(customizers[i])
.customize(any(io.rsocket.core.RSocketServer.class));
}
factory.setRSocketServerCustomizers(Arrays.asList(customizers));
this.server = factory.create(new EchoRequestResponseAcceptor());
InOrder ordered = inOrder((Object[]) customizers);
for (RSocketServerCustomizer customizer : customizers) {
ordered.verify(customizer).customize(any(io.rsocket.core.RSocketServer.class));
}
}
private RSocketRequester createRSocketTcpClient() { private RSocketRequester createRSocketTcpClient() {
Assertions.assertThat(this.server).isNotNull(); Assertions.assertThat(this.server).isNotNull();
InetSocketAddress address = this.server.address(); InetSocketAddress address = this.server.address();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment