Support single-value reactive types in @MessageMapping

This commit adds support for single-value reactive types in
@MessageMapping by converting them using ReactiveAdapterRegistry
and MonoToListenableFutureAdapter.

MonoToListenableFutureAdapter previously package private and used only
in org.springframework.messaging.tcp.reactor has been moved to
org.springframework.messaging.support and made public in order to be
used by ReactiveReturnValueHandler as well.

Issue: SPR-16634

Author: sdeleuze

spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/ReactiveReturnValueHandler.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2002-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.handler.invocation;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ReactiveAdapter;
+import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.messaging.support.MonoToListenableFutureAdapter;
+import org.springframework.util.concurrent.ListenableFuture;
+
+/**
+ * Support for single-value reactive types (like {@code Mono} or {@code Single}) as a
+ * return value type.
+ *
+ * @author Sebastien Deleuze
+ * @since 5.1
+ */
+public class ReactiveReturnValueHandler extends AbstractAsyncReturnValueHandler {
+
+	private final ReactiveAdapterRegistry adapterRegistry;
+
+
+	public ReactiveReturnValueHandler() {
+		this(ReactiveAdapterRegistry.getSharedInstance());
+	}
+
+	public ReactiveReturnValueHandler(ReactiveAdapterRegistry adapterRegistry) {
+		this.adapterRegistry = adapterRegistry;
+	}
+
+
+	@Override
+	public boolean supportsReturnType(MethodParameter returnType) {
+		return this.adapterRegistry.getAdapter(returnType.getParameterType()) != null;
+	}
+
+	@Override
+	public boolean isAsyncReturnValue(Object returnValue, MethodParameter returnType) {
+		ReactiveAdapter adapter = this.adapterRegistry.getAdapter(returnType.getParameterType(), returnValue);
+		return !adapter.isMultiValue() && !adapter.isNoValue();
+	}
+
+	@Override
+	public ListenableFuture<?> toListenableFuture(Object returnValue, MethodParameter returnType) {
+		ReactiveAdapter adapter = this.adapterRegistry.getAdapter(returnType.getParameterType(), returnValue);
+		return new MonoToListenableFutureAdapter<>(Mono.from(adapter.toPublisher(returnValue)));
+	}
+
+}

spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/SimpAnnotationMethodMessageHandler.java

@@ -60,6 +60,7 @@
 import org.springframework.messaging.handler.invocation.HandlerMethodReturnValueHandler;
 import org.springframework.messaging.handler.invocation.HandlerMethodReturnValueHandlerComposite;
 import org.springframework.messaging.handler.invocation.ListenableFutureReturnValueHandler;
+import org.springframework.messaging.handler.invocation.ReactiveReturnValueHandler;
 import org.springframework.messaging.simp.SimpAttributesContextHolder;
 import org.springframework.messaging.simp.SimpLogging;
 import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
@@ -337,6 +338,7 @@ protected List<? extends HandlerMethodReturnValueHandler> initReturnValueHandler
 
 		handlers.add(new ListenableFutureReturnValueHandler());
 		handlers.add(new CompletableFutureReturnValueHandler());
+		handlers.add(new ReactiveReturnValueHandler());
 
 		// Annotation-based return value types
 

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/AbstractMonoToListenableFutureAdapter.java renamed to spring-messaging/src/main/java/org/springframework/messaging/support/AbstractMonoToListenableFutureAdapter.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.messaging.tcp.reactor;
+package org.springframework.messaging.support;
 
 import java.time.Duration;
 import java.util.concurrent.ExecutionException;

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/MonoToListenableFutureAdapter.java renamed to spring-messaging/src/main/java/org/springframework/messaging/support/MonoToListenableFutureAdapter.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.messaging.tcp.reactor;
+package org.springframework.messaging.support;
 
 import reactor.core.publisher.Mono;
 
@@ -29,7 +29,7 @@
  * @since 5.0
  * @param <T> the object type
  */
-class MonoToListenableFutureAdapter<T> extends AbstractMonoToListenableFutureAdapter<T, T> {
+public class MonoToListenableFutureAdapter<T> extends AbstractMonoToListenableFutureAdapter<T, T> {
 
 	public MonoToListenableFutureAdapter(Mono<T> mono) {
 		super(mono);

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java

@@ -49,6 +49,7 @@
 
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MonoToListenableFutureAdapter;
 import org.springframework.messaging.tcp.ReconnectStrategy;
 import org.springframework.messaging.tcp.TcpConnection;
 import org.springframework.messaging.tcp.TcpConnectionHandler;

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpConnection.java

@@ -23,6 +23,7 @@
 import reactor.netty.NettyOutbound;
 
 import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MonoToListenableFutureAdapter;
 import org.springframework.messaging.tcp.TcpConnection;
 import org.springframework.util.concurrent.ListenableFuture;
 

spring-messaging/src/test/java/org/springframework/messaging/simp/annotation/support/SimpAnnotationMethodMessageHandlerTests.java

@@ -31,12 +31,18 @@
 import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxProcessor;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
 
 import org.springframework.context.support.StaticApplicationContext;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.handler.HandlerMethod;
@@ -336,6 +342,61 @@ public void completableFutureFailure() {
 		assertTrue(controller.exceptionCaught);
 	}
 
+	@Test
+	public void monoSuccess() {
+		Message emptyMessage = MessageBuilder.withPayload(new byte[0]).build();
+		given(this.channel.send(any(Message.class))).willReturn(true);
+		given(this.converter.toMessage(any(), any(MessageHeaders.class))).willReturn(emptyMessage);
+
+		ReactiveController controller = new ReactiveController();
+		this.messageHandler.registerHandler(controller);
+		this.messageHandler.setDestinationPrefixes(Arrays.asList("/app1", "/app2/"));
+
+		Message<?> message = createMessage("/app1/mono");
+		this.messageHandler.handleMessage(message);
+
+		assertNotNull(controller.mono);
+		controller.mono.onNext("foo");
+		verify(this.converter).toMessage(this.payloadCaptor.capture(), any(MessageHeaders.class));
+		assertEquals("foo", this.payloadCaptor.getValue());
+	}
+
+	@Test
+	public void monoFailure() {
+		Message emptyMessage = MessageBuilder.withPayload(new byte[0]).build();
+		given(this.channel.send(any(Message.class))).willReturn(true);
+		given(this.converter.toMessage(any(), any(MessageHeaders.class))).willReturn(emptyMessage);
+
+		ReactiveController controller = new ReactiveController();
+		this.messageHandler.registerHandler(controller);
+		this.messageHandler.setDestinationPrefixes(Arrays.asList("/app1", "/app2/"));
+
+		Message<?> message = createMessage("/app1/mono");
+		this.messageHandler.handleMessage(message);
+
+		controller.mono.onError(new IllegalStateException());
+		assertTrue(controller.exceptionCaught);
+	}
+
+	@Test
+	public void fluxNotHandled() {
+		Message emptyMessage = MessageBuilder.withPayload(new byte[0]).build();
+		given(this.channel.send(any(Message.class))).willReturn(true);
+		given(this.converter.toMessage(any(), any(MessageHeaders.class))).willReturn(emptyMessage);
+
+		ReactiveController controller = new ReactiveController();
+		this.messageHandler.registerHandler(controller);
+		this.messageHandler.setDestinationPrefixes(Arrays.asList("/app1", "/app2/"));
+
+		Message<?> message = createMessage("/app1/flux");
+		this.messageHandler.handleMessage(message);
+
+		assertNotNull(controller.flux);
+		controller.flux.onNext("foo");
+
+		verify(this.converter, never()).toMessage(any(), any(MessageHeaders.class));
+	}
+
 	@Test
 	public void placeholder() throws Exception {
 		Message<?> message = createMessage("/pre/myValue");
@@ -542,6 +603,33 @@ public void handleValidationException() {
 		}
 	}
 
+	@Controller
+	private static class ReactiveController {
+
+		private MonoProcessor<String> mono;
+
+		private FluxProcessor<String, String> flux;
+
+		private boolean exceptionCaught = false;
+
+		@MessageMapping("mono")
+		public Mono<String> handleMono() {
+			this.mono = MonoProcessor.create();
+			return this.mono;
+		}
+
+		@MessageMapping("flux")
+		public Flux<String> handleFlux() {
+			this.flux = EmitterProcessor.create();
+			return this.flux;
+		}
+
+		@MessageExceptionHandler(IllegalStateException.class)
+		public void handleValidationException() {
+			this.exceptionCaught = true;
+		}
+	}
+
 
 	private static class StringTestValidator implements Validator {