Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polishing RSocket module according latest SF

Browse Source
This commit is contained in:
Artem Bilan
2019-06-20 09:56:07 -04:00
parent d79c06a987
commit 3ec6b56931
3 changed files with 17 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java
View File

@@ -20,7 +20,6 @@ import java.lang.reflect.Method;
import java.util.Collections;
import java.util.List;
import java.util.function.BiFunction;
import java.util.function.Predicate;
import org.springframework.context.ApplicationContext;
import org.springframework.core.MethodParameter;
@@ -69,6 +68,10 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler
private MimeType defaultMetadataMimeType = IntegrationRSocket.COMPOSITE_METADATA;
IntegrationRSocketAcceptor() {
setHandlerPredicate((clazz) -> false);
}
/**
* Configure the default content type to use for data payloads.
* <p>By default this is not set. However a server acceptor will use the
@@ -117,11 +120,6 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler
return Collections.singletonList(new MessageHandlerMethodArgumentResolver());
}
@Override
protected Predicate<Class<?>> initHandlerPredicate() {
return (clazz) -> false;
}
@Override
public RSocket apply(ConnectionSetupPayload setupPayload, RSocket sendingRSocket) {
return createRSocket(setupPayload, sendingRSocket);

2
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java
View File

@@ -134,7 +134,7 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler
/**
* Configure a type for a request {@link Publisher} elements.
* @param publisherElementType the type of the request {@link Publisher} elements.
* @see RSocketRequester.RequestSpec#data(Publisher, Class)
* @see RSocketRequester.RequestSpec#data(Object, Class)
*/
public void setPublisherElementType(Class<?> publisherElementType) {
setPublisherElementTypeExpression(new ValueExpression<>(publisherElementType));

24
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java
View File

@@ -50,7 +50,7 @@ import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.rsocket.MessageHandlerAcceptor;
import org.springframework.messaging.rsocket.RSocketMessageHandler;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketRequesterMethodArgumentResolver;
import org.springframework.messaging.rsocket.RSocketStrategies;
@@ -124,7 +124,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.doOnBound(server -> port = server.port());
server = RSocketFactory.receive()
.frameDecoder(PayloadDecoder.ZERO_COPY)
.acceptor(serverContext.getBean(MessageHandlerAcceptor.class))
.acceptor(serverContext.getBean(RSocketMessageHandler.class).serverAcceptor())
.transport(TcpServerTransport.create(tcpServer))
.start()
.block();
@@ -521,11 +521,11 @@ public class RSocketOutboundGatewayIntegrationTests {
public static class ClientConfig extends CommonConfig {
@Bean
public MessageHandlerAcceptor clientAcceptor() {
MessageHandlerAcceptor acceptor = new MessageHandlerAcceptor();
acceptor.setHandlers(Collections.singletonList(controller()));
acceptor.setRSocketStrategies(rsocketStrategies());
return acceptor;
public RSocketMessageHandler messageHandler() {
RSocketMessageHandler handler = new RSocketMessageHandler();
handler.setRSocketStrategies(rsocketStrategies());
handler.setHandlers(Collections.singletonList(controller()));
return handler;
}
@Bean(destroyMethod = "dispose")
@@ -536,7 +536,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.dataMimeType("text/plain")
.metadataMimeType("message/x.rsocket.routing.v0")
.frameDecoder(PayloadDecoder.ZERO_COPY)
.acceptor(clientAcceptor())
.acceptor(messageHandler().clientAcceptor())
.transport(TcpClientTransport.create("localhost", port))
.start()
.block();
@@ -565,10 +565,10 @@ public class RSocketOutboundGatewayIntegrationTests {
static class ServerConfig extends CommonConfig {
@Bean
public MessageHandlerAcceptor serverAcceptor() {
MessageHandlerAcceptor acceptor = new MessageHandlerAcceptor();
acceptor.setRSocketStrategies(rsocketStrategies());
return acceptor;
public RSocketMessageHandler messageHandler() {
RSocketMessageHandler handler = new RSocketMessageHandler();
handler.setRSocketStrategies(rsocketStrategies());
return handler;
}
}