GH-1539 Fixed support for fluxed Consumer

Fixed support for Consumer<Flux<?>> which currently is not wrapped with FluxWrapper

NOTE: The newly created FluxedConsumerWrapper will be migrated to Spring Cloud Function after SCSt-2.1.0 and SCF-2.0.0 releases

Resolves #1539

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright 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.cloud.stream.function;
+
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.function.core.FluxWrapper;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ * @since 2.1
+ *
+ * Will most likely be moved to SCF
+ */
+class FluxedConsumerWrapper<T> implements Function<Flux<T>, Mono<Void>>, FluxWrapper<Consumer<Flux<T>>> {
+
+	private final Consumer<Flux<T>> consumer;
+
+	FluxedConsumerWrapper(Consumer<Flux<T>> consumer) {
+		this.consumer = consumer;
+	}
+
+	@Override
+	public Consumer<Flux<T>> getTarget() {
+		return consumer;
+	}
+
+	@Override
+	public Mono<Void> apply(Flux<T> t) {
+		return  Mono.fromRunnable(() -> this.consumer.accept(t));
+	}
+
+}

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

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.function;
 
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.apache.commons.logging.Log;
@@ -74,13 +75,15 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 		this(functionProperties, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	FunctionInvoker(StreamFunctionProperties functionProperties, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory, MessageChannel errorChannel) {
 		Object originalUserFunction = functionCatalog.lookup(functionProperties.getDefinition());
 
-		//TODO needs handling for when not a FluxWrapper
-		this.userFunction = (Function<Flux<?>, Flux<?>>) originalUserFunction;
+		this.userFunction = originalUserFunction instanceof Consumer
+				? new FluxedConsumerWrapper<>((Consumer) originalUserFunction)
+						: (Function<Flux<?>, Flux<?>>) originalUserFunction;
+
 		Assert.isInstanceOf(Function.class, this.userFunction);
 		this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
 		FunctionType functionType = functionInspector.getRegistration(originalUserFunction).getType();

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

@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.function;
 
 import java.lang.reflect.Field;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.junit.Test;
@@ -43,6 +44,7 @@ import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.ReflectionUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertEquals;
 
 /**
  *
@@ -148,6 +150,30 @@ public class FunctionInvokerTests {
 		}
 	}
 
+	private static String testWithFluxedConsumerValue;
+
+	@Test
+	public void testWithFluxedConsumer() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
+
+			String value = "Hello";
+			Message<String> inputMessage = new GenericMessage<>(value);
+
+			StreamFunctionProperties functionProperties = createStreamFunctionProperties();
+
+			functionProperties.setDefinition("fluxConsumer");
+			FunctionInvoker<String, Void> fluxedConsumer = new FunctionInvoker<>(functionProperties,
+					new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)),
+					context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
+
+			fluxedConsumer.apply(Flux.just(inputMessage)).blockFirst();
+
+			assertEquals(value, testWithFluxedConsumerValue);
+		}
+	}
+
 	private StreamFunctionProperties createStreamFunctionProperties() {
 		StreamFunctionProperties functionProperties = new StreamFunctionProperties();
 		ConsumerProperties consumerProperties = new ConsumerProperties();
@@ -184,6 +210,14 @@ public class FunctionInvokerTests {
 	@EnableAutoConfiguration
 	public static class MyFunctionsConfiguration {
 
+		@Bean
+		public Consumer<Flux<String>> fluxConsumer() {
+			return f -> f.subscribe(v -> {
+				System.out.println("Consuming flux: " + v);
+				testWithFluxedConsumerValue = v;
+			});
+		}
+
 		@Bean
 		public Function<Message<Foo>, Message<Bar>> messageToMessageDifferentType() {
 			return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders()).build();