Add additional tests to validate edge scenarious

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

@@ -50,6 +50,7 @@ import org.springframework.cloud.stream.binder.test.TestChannelBinderConfigurati
 import org.springframework.cloud.stream.messaging.Sink;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.handler.LoggingHandler;
@@ -57,6 +58,7 @@ import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.support.PeriodicTrigger;
@@ -942,6 +944,37 @@ public class ImplicitFunctionBindingTests {
 			assertThat(result.getPayload()).isInstanceOf(String.class); // no output conversion to byte[] has happened.
 			assertThat(result.getPayload()).isEqualTo("byte[]");
 		}
+
+		//Consumer reactiveConsumer
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
+								"--spring.cloud.function.definition=reactiveConsumer",
+								"--spring.cloud.stream.bindings.reactiveConsumer-in-0.consumer.useNativeDecoding=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			inputDestination.send(new GenericMessage<byte[]>("hello".getBytes()));
+
+			QueueChannel testChannel = context.getBean("testChannel", QueueChannel.class);
+
+			Message result = testChannel.receive(2000);
+			assertThat(result.getPayload()).isEqualTo(byte[].class.getName());
+		}
+
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
+								"--spring.cloud.function.definition=reactiveFunctionConsumer",
+								"--spring.cloud.stream.bindings.reactiveFunctionConsumer-in-0.consumer.useNativeDecoding=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			inputDestination.send(new GenericMessage<byte[]>("hello".getBytes()));
+
+			QueueChannel testChannel = context.getBean("testChannel", QueueChannel.class);
+
+			Message result = testChannel.receive(2000);
+			assertThat(result.getPayload()).isEqualTo(byte[].class.getName());
+		}
 	}
 
 	@Test
@@ -1036,6 +1069,11 @@ public class ImplicitFunctionBindingTests {
 	@EnableAutoConfiguration
 	public static class SingleFunctionConfiguration2 {
 
+		@Bean
+		public QueueChannel testChannel() {
+			return new QueueChannel();
+		}
+
 		@Bean
 		public Function<Object, String> imperative() {
 			return x -> {
@@ -1049,6 +1087,20 @@ public class ImplicitFunctionBindingTests {
 				return x.getClass().getSimpleName();
 			});
 		}
+
+		@Bean
+		public Consumer<Flux<Message<?>>> reactiveConsumer(MessageChannel testChannel) {
+			return flux -> flux.subscribe(v -> {
+				testChannel.send(new GenericMessage<String>(((Message<?>) v).getPayload().getClass().getName()));
+			});
+		}
+
+		@Bean
+		public Function<Flux<Message<?>>, Mono<Void>> reactiveFunctionConsumer(MessageChannel testChannel) {
+			return flux -> flux.doOnNext(x -> {
+				testChannel.send(new GenericMessage<String>(((Message<?>) x).getPayload().getClass().getName()));
+			}).then();
+		}
 	}
 
 	@EnableAutoConfiguration

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

@@ -0,0 +1,71 @@
+/*
+ * 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.util.function.Function;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+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.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class ScenarioTests {
+
+	@Test
+	public void testComposingSupplierWuthTypelessMessageFunction() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(TestConfiguration.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.jmx.enabled=false",
+								"--spring.cloud.function.definition=messageSupplier|messageFunction")) {
+
+			OutputDestination output = context.getBean(OutputDestination.class);
+			assertThat(output.receive(1000)).isNotNull();
+			assertThat(output.receive(1100)).isNotNull();
+			assertThat(output.receive(1200)).isNotNull();
+		}
+	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	public static class TestConfiguration {
+		@Bean
+		public Supplier<Message<?>> messageSupplier() {
+			return () -> new GenericMessage<>("10/27/20 07:20:01");
+		}
+		@Bean
+		public Function<Message<?>, Message<?>> messageFunction() {
+			return message -> message;
+		}
+	}
+}