Fix Rsocket module according latest SF changes

Author: artembilan

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

@@ -32,11 +32,12 @@
 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 @@
  * 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 @@ private Mono<Void> handle(Payload payload) {
 		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 @@ private Flux<Payload> handleAndReply(Payload firstPayload, Flux<Payload> payload
 						.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()) {

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

@@ -30,9 +30,9 @@
 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 @@ public void setDefaultDataMimeType(@Nullable MimeType defaultDataMimeType) {
 	 * <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 @@ protected IntegrationRSocket createRSocket(ConnectionSetupPayload setupPayload,
 						? 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());
 	}

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

@@ -41,8 +41,8 @@
 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;
 

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

@@ -28,7 +28,7 @@
 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;
 

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

@@ -50,10 +50,10 @@
 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;