Upgrade to the latest Reactor & SF

Related to spring-projects/spring-framework#25884

* Don't use `MonoProcessor` & `FluxProcessor` since they are deprecated now
* Fix RSocket components to use an `AtomicReference` header instead of deprecated
`MonoProcessor`
* Introduce an internal `IntegrationRSocketPayloadReturnValueHandler` to handle
properly a Spring Integration case for inbound requests with its
`IntegrationRSocketEndpoint` implementation
* Don't do response handling in the `RSocketInboundGateway` any more
since it is deferred now to the `IntegrationRSocketPayloadReturnValueHandler`
* Remove internal `ChannelSendOperator` since it is out of use from now on

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

@@ -29,7 +29,6 @@
 import reactor.core.Disposable;
 import reactor.core.Disposables;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxProcessor;
 import reactor.core.publisher.Sinks;
 import reactor.core.scheduler.Schedulers;
 
@@ -46,24 +45,17 @@
 public class FluxMessageChannel extends AbstractMessageChannel
 		implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
 
-	private final Sinks.Many<Message<?>> sink;
-
-	private final FluxProcessor<Message<?>, Message<?>> processor;
+	private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
 
 	private final Sinks.Many<Boolean> subscribedSignal = Sinks.many().replay().limit(1);
 
 	private final Disposable.Composite upstreamSubscriptions = Disposables.composite();
 
 	private volatile boolean active = true;
 
-	public FluxMessageChannel() {
-		this.sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
-		this.processor = FluxProcessor.fromSink(this.sink);
-	}
-
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
-		Assert.state(this.active && this.processor.hasDownstreams(),
+		Assert.state(this.active && this.sink.currentSubscriberCount() > 0,
 				() -> "The [" + this + "] doesn't have subscribers to accept messages");
 		long remainingTime = 0;
 		if (timeout > 0) {
@@ -101,11 +93,12 @@ private boolean tryEmitMessage(Message<?> message) {
 
 	@Override
 	public void subscribe(Subscriber<? super Message<?>> subscriber) {
-		this.processor
-				.doFinally((s) -> this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams()))
+		this.sink.asFlux()
+				.doFinally((s) -> this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0))
 				.share()
 				.subscribe(subscriber);
-		this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams());
+
+		this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0);
 	}
 
 	@Override
@@ -131,9 +124,9 @@ public void subscribeTo(Publisher<? extends Message<?>> publisher) {
 	@Override
 	public void destroy() {
 		this.active = false;
-		this.subscribedSignal.tryEmitNext(false);
 		this.upstreamSubscriptions.dispose();
-		this.processor.onComplete();
+		this.subscribedSignal.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
+		this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
 		super.destroy();
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java

@@ -67,7 +67,6 @@
 import org.springframework.util.Assert;
 
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.Sinks;
 
 /**
@@ -899,7 +898,10 @@ public boolean send(Message<?> message, long timeout) {
 
 		@Override
 		public void subscribeTo(Publisher<? extends Message<?>> publisher) {
-			publisher.subscribe(MonoProcessor.fromSink(this.replyMono));
+			Mono.from(publisher)
+					.subscribe(
+							(value) -> this.replyMono.emitValue(value, Sinks.EmitFailureHandler.FAIL_FAST),
+							this.replyMono::tryEmitError, this.replyMono::tryEmitEmpty);
 		}
 
 	}

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java

@@ -51,7 +51,6 @@
 
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxProcessor;
 
 /**
  * @author Artem Bilan
@@ -141,7 +140,7 @@ void testFluxMessageChannelCleanUp() throws InterruptedException {
 
 		flowRegistration.destroy();
 
-		assertThat(TestUtils.getPropertyValue(flux, "processor", FluxProcessor.class).isTerminated()).isTrue();
+		assertThat(TestUtils.getPropertyValue(flux, "sink.sink.done", Boolean.class)).isTrue();
 	}
 
 	@Configuration

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

@@ -17,24 +17,35 @@
 package org.springframework.integration.rsocket;
 
 import java.lang.reflect.Method;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.core.MethodParameter;
+import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.core.codec.Encoder;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.ReactiveMessageHandler;
 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.HandlerMethodReturnValueHandler;
 import org.springframework.messaging.handler.invocation.reactive.SyncHandlerMethodArgumentResolver;
 import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition;
 import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
+import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler;
+import org.springframework.util.Assert;
 import org.springframework.util.ReflectionUtils;
 
+import io.rsocket.Payload;
 import io.rsocket.frame.FrameType;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 /**
  * The {@link RSocketMessageHandler} extension for Spring Integration needs.
@@ -109,6 +120,23 @@ protected List<? extends HandlerMethodArgumentResolver> initArgumentResolvers()
 		}
 	}
 
+	@Override
+	@SuppressWarnings("unchecked")
+	protected List<? extends HandlerMethodReturnValueHandler> initReturnValueHandlers() {
+		HandlerMethodReturnValueHandler integrationRSocketPayloadReturnValueHandler =
+				new IntegrationRSocketPayloadReturnValueHandler((List<Encoder<?>>) getEncoders(),
+						getReactiveAdapterRegistry());
+		if (this.messageMappingCompatible) {
+			List<HandlerMethodReturnValueHandler> handlers = new ArrayList<>();
+			handlers.add(integrationRSocketPayloadReturnValueHandler);
+			handlers.addAll(getReturnValueHandlerConfigurer().getCustomHandlers());
+			return handlers;
+		}
+		else {
+			return Collections.singletonList(integrationRSocketPayloadReturnValueHandler);
+		}
+	}
+
 	protected static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {
 
 		@Override
@@ -123,4 +151,35 @@ public Object resolveArgumentValue(MethodParameter parameter, Message<?> message
 
 	}
 
+	protected static final class IntegrationRSocketPayloadReturnValueHandler extends RSocketPayloadReturnValueHandler {
+
+		protected IntegrationRSocketPayloadReturnValueHandler(List<Encoder<?>> encoders,
+				ReactiveAdapterRegistry registry) {
+
+			super(encoders, registry);
+		}
+
+		@Override public Mono<Void> handleReturnValue(@Nullable Object returnValue, MethodParameter returnType,
+				Message<?> message) {
+
+			AtomicReference<Flux<Payload>> responseReference = getResponseReference(message);
+
+			if (returnValue == null && responseReference != null) {
+				return super.handleReturnValue(responseReference.get(), returnType, message);
+			}
+			else {
+				return super.handleReturnValue(returnValue, returnType, message);
+			}
+		}
+
+		@Nullable
+		@SuppressWarnings("unchecked")
+		private static AtomicReference<Flux<Payload>> getResponseReference(Message<?> message) {
+			Object headerValue = message.getHeaders().get(RESPONSE_HEADER);
+			Assert.state(headerValue == null || headerValue instanceof AtomicReference, "Expected AtomicReference");
+			return (AtomicReference<Flux<Payload>>) headerValue;
+		}
+
+	}
+
 }