From e3393b98ca2d4db83141cdceeef8f671277d873e Mon Sep 17 00:00:00 2001
From: jmaxwell <jmaxwell@theice.com>
Date: Wed, 14 Mar 2018 22:13:35 -0500
Subject: [PATCH] GH-1301 Add conversion to Flux Message Payload.

Resolves #1301
Resolves #1302
---
 ...ageChannelToInputFluxParameterAdapter.java | 34 ++++++++++++++-----
 1 file changed, 26 insertions(+), 8 deletions(-)

diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java
index b16e7c7a2..d809a985f 100644
--- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java
+++ b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java
@@ -26,6 +26,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
 /**
@@ -53,15 +54,33 @@ public class MessageChannelToInputFluxParameterAdapter
 
 	@Override
 	public Flux<?> adapt(final SubscribableChannel bindingTarget, MethodParameter parameter) {
-		ResolvableType resolvableType = ResolvableType.forMethodParameter(parameter);
-		final Class<?> argumentClass = (resolvableType.getGeneric(0).getRawClass() != null) ? (resolvableType
-				.getGeneric(0).getRawClass()) : Object.class;
+		final ResolvableType fluxResolvableType = ResolvableType.forMethodParameter(parameter);
+		final ResolvableType fluxTypeParameter = fluxResolvableType.getGeneric(0);
+		final Class<?> fluxTypeParameterRawClass = fluxTypeParameter.getRawClass();
+		final Class<?> fluxTypeParameterClass = (fluxTypeParameterRawClass != null) ? fluxTypeParameterRawClass
+				: Object.class;
+
 		final Object monitor = new Object();
-		if (Message.class.isAssignableFrom(argumentClass)) {
+
+		if (Message.class.isAssignableFrom(fluxTypeParameterClass)) {
+
+			final ResolvableType payloadTypeParameter = fluxTypeParameter.getGeneric(0);
+			final Class<?> payloadTypeParameterRawClass = payloadTypeParameter.getRawClass();
+			final Class<?> payloadTypeParameterClass = (payloadTypeParameterRawClass != null)
+					? payloadTypeParameterRawClass : Object.class;
+
 			return Flux.create(emitter -> {
 				MessageHandler messageHandler = message -> {
 					synchronized (monitor) {
-						emitter.next(message);
+
+						if (payloadTypeParameterClass.isAssignableFrom(message.getPayload().getClass())) {
+							emitter.next(message);
+						}
+						else {
+							emitter.next(MessageBuilder.createMessage(
+									this.messageConverter.fromMessage(message, payloadTypeParameterClass),
+									message.getHeaders()));
+						}
 					}
 				};
 				bindingTarget.subscribe(messageHandler);
@@ -72,12 +91,11 @@ public class MessageChannelToInputFluxParameterAdapter
 			return Flux.create(emitter -> {
 				MessageHandler messageHandler = message -> {
 					synchronized (monitor) {
-						if (argumentClass.isAssignableFrom(message
-								.getPayload().getClass())) {
+						if (fluxTypeParameterClass.isAssignableFrom(message.getPayload().getClass())) {
 							emitter.next(message.getPayload());
 						}
 						else {
-							emitter.next(this.messageConverter.fromMessage(message, argumentClass));
+							emitter.next(this.messageConverter.fromMessage(message, fluxTypeParameterClass));
 						}
 					}
 				};