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

Fix Rsocket module according latest SF changes

Browse Source
This commit is contained in:
Artem Bilan
2019-06-21 16:19:57 -04:00
parent 3ec6b56931
commit f04962301d
5 changed files with 17 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java
View File

@@ -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.
* <p>
* 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<MimeType> METADATA_MIME_TYPES = Arrays.asList(COMPOSITE_METADATA, ROUTING);
private final Function<Message<?>, Mono<Void>> handler;
private final ReactiveMessageHandler handler;
private final RouteMatcher routeMatcher;
@@ -87,7 +88,7 @@ class IntegrationRSocket extends AbstractRSocket {
private final MimeType metadataMimeType;
IntegrationRSocket(Function<Message<?>, Mono<Void>> 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<Flux<DataBuffer>> 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()) {

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

@@ -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
* <p>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());
}

2
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java
View File

@@ -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;

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

@@ -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;

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

@@ -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;