From f04962301db544e0e86d5bf9ee78de8bad334b19 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 21 Jun 2019 16:19:57 -0400 Subject: [PATCH] Fix Rsocket module according latest SF changes --- .../integration/rsocket/IntegrationRSocket.java | 17 +++++++++-------- .../rsocket/IntegrationRSocketAcceptor.java | 6 ++++-- .../rsocket/inbound/RSocketInboundGateway.java | 2 +- .../outbound/RSocketOutboundGateway.java | 2 +- .../RSocketOutboundGatewayIntegrationTests.java | 4 ++-- 5 files changed, 17 insertions(+), 14 deletions(-) diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java index 550f086b01..d3f9a65a9d 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java @@ -32,11 +32,12 @@ import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.ReactiveMessageHandler; import org.springframework.messaging.handler.DestinationPatternsMessageCondition; import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler; -import org.springframework.messaging.rsocket.RSocketPayloadReturnValueHandler; import org.springframework.messaging.rsocket.RSocketRequester; -import org.springframework.messaging.rsocket.RSocketRequesterMethodArgumentResolver; +import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler; +import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver; import org.springframework.messaging.support.MessageBuilder; import org.springframework.messaging.support.MessageHeaderAccessor; import org.springframework.util.Assert; @@ -57,14 +58,14 @@ import reactor.core.publisher.MonoProcessor; * obtains the response from a "reply" header. *

* Essentially, this is an adapted for Spring Integration copy - * of the {@link org.springframework.messaging.rsocket.MessagingRSocket} because + * of the {@link org.springframework.messaging.rsocket.annotation.support.MessagingRSocket} because * that one is not public. * * @author Artem Bilan * * @since 5.2 * - * @see org.springframework.messaging.rsocket.MessagingRSocket + * @see org.springframework.messaging.rsocket.annotation.support.MessagingRSocket */ class IntegrationRSocket extends AbstractRSocket { @@ -75,7 +76,7 @@ class IntegrationRSocket extends AbstractRSocket { static final List METADATA_MIME_TYPES = Arrays.asList(COMPOSITE_METADATA, ROUTING); - private final Function, Mono> handler; + private final ReactiveMessageHandler handler; private final RouteMatcher routeMatcher; @@ -87,7 +88,7 @@ class IntegrationRSocket extends AbstractRSocket { private final MimeType metadataMimeType; - IntegrationRSocket(Function, Mono> handler, RouteMatcher routeMatcher, + IntegrationRSocket(ReactiveMessageHandler handler, RouteMatcher routeMatcher, RSocketRequester requester, MimeType dataMimeType, MimeType metadataMimeType, DataBufferFactory bufferFactory) { @@ -149,7 +150,7 @@ class IntegrationRSocket extends AbstractRSocket { DataBuffer dataBuffer = retainDataAndReleasePayload(payload); int refCount = refCount(dataBuffer); Message message = MessageBuilder.createMessage(dataBuffer, headers); - return Mono.defer(() -> this.handler.apply(message)) + return Mono.defer(() -> this.handler.handleMessage(message)) .doFinally((signal) -> { if (refCount(dataBuffer) == refCount) { DataBufferUtils.release(dataBuffer); @@ -173,7 +174,7 @@ class IntegrationRSocket extends AbstractRSocket { .doOnSubscribe((subscription) -> read.set(true)); Message> message = MessageBuilder.createMessage(buffers, headers); - return Mono.defer(() -> this.handler.apply(message)) + return Mono.defer(() -> this.handler.handleMessage(message)) .doFinally((signal) -> { // Subscription should have happened by now due to ChannelSendOperator if (!read.get()) { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java index d2e1e7b84e..58d4c4eeda 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketAcceptor.java @@ -30,9 +30,9 @@ import org.springframework.messaging.handler.CompositeMessageCondition; import org.springframework.messaging.handler.DestinationPatternsMessageCondition; import org.springframework.messaging.handler.invocation.reactive.HandlerMethodArgumentResolver; import org.springframework.messaging.handler.invocation.reactive.SyncHandlerMethodArgumentResolver; -import org.springframework.messaging.rsocket.RSocketMessageHandler; import org.springframework.messaging.rsocket.RSocketRequester; import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler; import org.springframework.util.Assert; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -79,6 +79,7 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler * required for clients but can also be used on servers as a fallback. * @param defaultDataMimeType the MimeType to use */ + @Override public void setDefaultDataMimeType(@Nullable MimeType defaultDataMimeType) { this.defaultDataMimeType = defaultDataMimeType; } @@ -89,6 +90,7 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler *

By default this is set to {@code "message/x.rsocket.composite-metadata.v0"} * @param mimeType the MimeType to use */ + @Override public void setDefaultMetadataMimeType(MimeType mimeType) { Assert.notNull(mimeType, "'metadataMimeType' is required"); this.defaultMetadataMimeType = mimeType; @@ -138,7 +140,7 @@ class IntegrationRSocketAcceptor extends RSocketMessageHandler ? MimeTypeUtils.parseMimeType(setupPayload.metadataMimeType()) : this.defaultMetadataMimeType; Assert.notNull(dataMimeType, "No `metadataMimeType` in the ConnectionSetupPayload and no default value"); - return new IntegrationRSocket(this::handleMessage, getRouteMatcher(), + return new IntegrationRSocket(this, getRouteMatcher(), RSocketRequester.wrap(rsocket, dataMimeType, metadataMimeType, rsocketStrategies), dataMimeType, metadataMimeType, rsocketStrategies.dataBufferFactory()); } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java index 4cc669c55e..0801b00621 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java @@ -41,8 +41,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler; -import org.springframework.messaging.rsocket.RSocketPayloadReturnValueHandler; import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler; import org.springframework.util.Assert; import org.springframework.util.MimeType; diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java index 534c6b08c9..35d4bbe026 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java @@ -28,7 +28,7 @@ import org.springframework.integration.rsocket.ClientRSocketConnector; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.rsocket.RSocketRequester; -import org.springframework.messaging.rsocket.RSocketRequesterMethodArgumentResolver; +import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java index 0a8fa6bca6..a177b07fe5 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java @@ -50,10 +50,10 @@ 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.RSocketMessageHandler; import org.springframework.messaging.rsocket.RSocketRequester; -import org.springframework.messaging.rsocket.RSocketRequesterMethodArgumentResolver; import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler; +import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver; import org.springframework.messaging.support.ErrorMessage; import org.springframework.stereotype.Controller; import org.springframework.test.annotation.DirtiesContext;