Reactive functions and native encoding

When native encoding is enabled with a reactive function,
Spring Cloud Stream is still trying to convert the message
causing ClassCastException and other potential issues downstream.
Fixing this issue by signaling the framework to not perform any
message conversion on the outbound when useNativeEncoding is
enabled with a reactive function.

Resovles https://github.com/spring-cloud/spring-cloud-stream/issues/2037

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -28,6 +28,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -96,6 +97,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -444,12 +446,16 @@ public class FunctionConfiguration {
 					resultPublishers = Collections.singletonList(resultPublishers);
 				}
 				Iterator<String> outputBindingIter = outputBindingNames.iterator();
-
+				AtomicInteger index = new AtomicInteger();
 				((Iterable) resultPublishers).forEach(publisher -> {
 					Flux flux = Flux.from((Publisher) publisher);
 					if (!CollectionUtils.isEmpty(outputBindingNames)) {
-						MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class);
-						flux = flux.doOnNext(message -> {
+						final String outputDestinationName = outputBindingIter.next();
+						this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, index.getAndIncrement());
+						MessageChannel outputChannel = this.applicationContext.getBean(outputDestinationName, MessageChannel.class);
+						flux = flux
+							.map(value -> value instanceof Message ? value : new GenericMessage(value))
+							.doOnNext(message -> {
 							if (message instanceof Message && ((Message<?>) message).getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
 								String destinationName = (String) ((Message<?>) message).getHeaders().get("spring.cloud.stream.sendto.destination");
 								ProducerProperties producerProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()).getProducer();

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ReactiveFunctionWithNativeEncodingTests.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2020-2020 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
+ *
+ *      https://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.cloud.stream.function;
+
+import java.lang.reflect.Field;
+import java.util.function.Function;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+import reactor.core.publisher.Flux;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.util.ReflectionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author Soby Chacko
+ */
+public class ReactiveFunctionWithNativeEncodingTests {
+
+	@Test
+	public void testFunctionIsAppliedToExistingMessageSource() throws Exception {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(
+				FunctionsConfiguration.class)).web(WebApplicationType.NONE).run(
+			"--spring.cloud.stream.function.definition=toUpperCaseReactive",
+			"--spring.jmx.enabled=false",
+			"--spring.cloud.stream.bindings.toUpperCaseReactive-in-0.consumer.useNativeDecoding=true",
+			"--spring.cloud.stream.bindings.toUpperCaseReactive-out-0.producer.useNativeEncoding=true"
+		)) {
+			SimpleFunctionRegistry simpleFunctionRegistry = context.getBean(SimpleFunctionRegistry.class);
+
+			final Field messageConverter = ReflectionUtils
+				.findField(SimpleFunctionRegistry.class, "messageConverter", CompositeMessageConverter.class);
+			messageConverter.setAccessible(true);
+			final CompositeMessageConverter o = (CompositeMessageConverter) messageConverter.get(simpleFunctionRegistry);
+
+			final CompositeMessageConverter spy = Mockito.spy(o);
+			ReflectionUtils.setField(messageConverter, simpleFunctionRegistry, spy);
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+			Message<byte[]> inputMessageOne = MessageBuilder.withPayload("Hello".getBytes()).build();
+			inputDestination.send(inputMessageOne);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+
+			assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes());
+			// Verify that no message conversion done by the framework on the outbound.
+			// If useNativeEncoding is not enabled (see the properties above),
+			// then the following verification should fail.
+			verify(spy, never()).toMessage(any(), any());
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class FunctionsConfiguration {
+
+		@Bean
+		public Function<Flux<byte[]>, Flux<byte[]>> toUpperCaseReactive() {
+			return flux -> flux.map(f -> new String(f).toUpperCase().getBytes());
+		}
+
+	}
+}