GH-1441 Aded initial support for Functions and Consumers

Resolves #1441

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

@@ -27,8 +27,10 @@ import reactor.core.publisher.Flux;
 import org.springframework.cloud.function.context.FunctionType;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.util.Assert;
@@ -49,12 +51,16 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 
 	private final Class<?> inputClass;
 
+	private final Class<?> outputClass;
+
 	private final Function<Flux<?>, Flux<?>> userFunction;
 
 	private final CompositeMessageConverter messageConverter;
 
 	private final MessageChannel errorChannel;
 
+	private final boolean isInputArgumentMessage;
+
 	FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory) {
 		this(functionName, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
@@ -66,39 +72,51 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 		Assert.isInstanceOf(Function.class, this.userFunction);
 		this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
 		FunctionType functionType = functionInspector.getRegistration(this.userFunction).getType();
+		this.isInputArgumentMessage = functionType.isMessage();
 		this.inputClass = functionType.getInputType();
+		this.outputClass = functionType.getOutputType();
 		this.errorChannel = errorChannel;
 	}
 
+	@SuppressWarnings("unchecked")
 	@Override
 	public Flux<Message<O>> apply(Flux<Message<I>> input) {
 		AtomicReference<Message<I>> originalMessageRef = new AtomicReference<>();
 		return input
 				.doOnNext(originalMessageRef::set)        // to preserve the original message
 				.map(this::resolveArgument)                // resolves argument type before invocation of user function
-				.onErrorContinue((exception, originalMessage) -> {
-					if (this.errorChannel != null) {
-						ErrorMessage em = new ErrorMessage(exception, (Message<?>) originalMessage);
-						logger.error(em);
-						this.errorChannel.send(em);
-					}
-					else {
-						exception.printStackTrace();
-					}
-				})
+				.onErrorContinue((x, y) -> onError(x, (Message<I>) y))
 				.transform(this.userFunction::apply)    // invoke user function
 				.map(resultMessage -> toMessage(resultMessage, originalMessageRef.get())); // create output message
 	}
 
+	private void onError(Throwable t, Message<I> originalMessage) {
+		if (this.errorChannel != null) {
+			ErrorMessage em = new ErrorMessage(t, (Message<?>) originalMessage);
+			logger.error(em);
+			this.errorChannel.send(em);
+		}
+		else {
+			logger.error(t);
+		}
+	}
+
 	@SuppressWarnings("unchecked")
 	private <T> Message<O> toMessage(T value, Message<I> originalMessage) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Converting result back to message using the original message: " + originalMessage);
 		}
-		return (Message<O>)
+		Message<O> returnMessage = (Message<O>)
 				(value instanceof Message
 						? value
-						: this.messageConverter.toMessage(value, originalMessage.getHeaders()));
+						: this.messageConverter.toMessage(value, originalMessage.getHeaders(), this.outputClass));
+		if (returnMessage == null) {
+			if (value.getClass().isAssignableFrom(this.outputClass)) {
+				returnMessage = (Message<O>) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders()).removeHeader(MessageHeaders.CONTENT_TYPE).build();
+			}
+		}
+		Assert.notNull(returnMessage, "Failed to convert result value '" + value + "' to message.");
+		return returnMessage;
 	}
 
 	@SuppressWarnings("unchecked")
@@ -110,13 +128,15 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 		T argument = (T) (shouldConvertFromMessage(message)
 				? this.messageConverter.fromMessage(message, this.inputClass)
 				: message);
-
 		Assert.notNull(argument, "Failed to resolve argument type '" + this.inputClass + "' from message: " + message );
+		if (!this.isInputArgumentMessage && argument instanceof Message) {
+			argument = ((Message<T>)argument).getPayload();
+		}
 		return argument;
 	}
 
 	private boolean shouldConvertFromMessage(Message<?> message) {
-		return !this.inputClass.isAssignableFrom(byte[].class) &&
+		return !message.getPayload().getClass().isAssignableFrom(this.inputClass) &&
 				!this.inputClass.isAssignableFrom(Object.class);
 	}
 

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

@@ -33,6 +33,7 @@ import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.SubscribableChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -103,6 +104,15 @@ public class IntegrationFlowFunctionSupport {
 		return IntegrationFlows.from(supplier);
 	}
 
+	/**
+	 * @param inputChannel
+	 * @return
+	 */
+	public <O> IntegrationFlowBuilder integrationFlowFromChannel(SubscribableChannel inputChannel) {
+		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel).bridge();
+		return flowBuilder;
+	}
+
 	/**
 	 * Add a {@link Function} bean to the end of an integration flow.
 	 * The name of the bean must be provided via `spring.cloud.stream.function.name` property.

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

@@ -0,0 +1,93 @@
+/*
+ * 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.Function;
+
+import org.junit.Test;
+
+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.FunctionCatalog;
+import org.springframework.cloud.function.context.catalog.FunctionInspector;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+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.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class FunctionInvokerTests {
+
+	@Test
+	public void testSameMessageTypesAreNotConverted() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class)).web(
+				WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false")) {
+
+			Message<Foo> inputMessage = new GenericMessage<>(new Foo());
+
+			FunctionInvoker<Foo, Foo> messageToMessageSameType = new FunctionInvoker<>("messageToMessageSameType",
+					new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
+			Message<Foo> outputMessage = messageToMessageSameType.apply(Flux.just(inputMessage)).blockFirst();
+			assertThat(inputMessage).isSameAs(outputMessage);
+
+			FunctionInvoker<Foo, Foo> pojoToPojoSameType = new FunctionInvoker<>("pojoToPojoSameType",
+					new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
+			outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage)).blockFirst();
+			assertThat(inputMessage.getPayload()).isEqualTo(outputMessage.getPayload());
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class MyFunctionsConfiguration {
+
+		@Bean
+		public Function<Message<Foo>, Message<Bar>> messageToMessageDifferentType() {
+			return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders()).build();
+		}
+		@Bean
+		public Function<Message<Foo>, Message<Foo>> messageToMessageSameType() {
+			return x -> x;
+		}
+
+		@Bean
+		public Function<Foo, Foo> pojoToPojoSameType() {
+			return x -> x;
+		}
+	}
+
+	private static class Foo {
+
+	}
+
+	private static class Bar {
+
+	}
+}

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

@@ -0,0 +1,178 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.junit.Test;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+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.cloud.stream.messaging.Processor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlowBuilder;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ * @since 2.1
+ *
+ */
+public class ProcessorToFunctionsSupportTests {
+
+	@Test
+	public void testPathThrough() {
+		ApplicationContext context =
+				new SpringApplicationBuilder(
+						TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.jmx.enabled=false");
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
+		assertThat(target.receive(1000).getPayload()).isEqualTo("hello".getBytes(StandardCharsets.UTF_8));
+	}
+
+	@Test
+	public void testSingleFunction() {
+		ApplicationContext context =
+				new SpringApplicationBuilder(
+						TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.cloud.stream.function.name=toUpperCase", "--spring.jmx.enabled=false");
+
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
+		assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8));
+	}
+
+	@Test
+	public void testComposedFunction() {
+		ApplicationContext context =
+				new SpringApplicationBuilder(
+						TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf", "--spring.jmx.enabled=false");
+
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
+		assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO:HELLO".getBytes(StandardCharsets.UTF_8));
+	}
+
+	@Test
+	public void testConsumer() {
+		ApplicationContext context =
+				new SpringApplicationBuilder(
+						TestChannelBinderConfiguration.getCompleteConfiguration(ConsumerConfiguration.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.cloud.stream.function.name=log", "--spring.jmx.enabled=false");
+
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
+		source.send(new GenericMessage<byte[]>("hello1".getBytes(StandardCharsets.UTF_8)));
+		source.send(new GenericMessage<byte[]>("hello2".getBytes(StandardCharsets.UTF_8)));
+		assertThat(target.receive(1000).getPayload()).isEqualTo("hello".getBytes(StandardCharsets.UTF_8));
+		assertThat(target.receive(1000).getPayload()).isEqualTo("hello1".getBytes(StandardCharsets.UTF_8));
+		assertThat(target.receive(1000).getPayload()).isEqualTo("hello2".getBytes(StandardCharsets.UTF_8));
+	}
+
+	@EnableAutoConfiguration
+	@Import(BaseProcessorConfiguration.class)
+	public static class FunctionsConfiguration {
+
+		@Bean
+		public Function<String, String> toUpperCase() {
+			return String::toUpperCase;
+		}
+
+		@Bean
+		public Function<String, String> concatWithSelf() {
+			return x -> x + ":" + x;
+		}
+	}
+
+	@EnableAutoConfiguration
+	@Import(BaseProcessorConfiguration.class)
+	public static class ConsumerConfiguration {
+
+		@Autowired
+		OutputDestination out;
+
+		@Bean
+		public Consumer<String> log() {
+			return x -> {
+				DirectFieldAccessor dfa = new DirectFieldAccessor(out);
+				MessageChannel channel = (MessageChannel) dfa.getPropertyValue("channel");
+				channel.send(new GenericMessage<byte[]>(x.getBytes()));
+			};
+		}
+	}
+
+	/**
+	 * This configuration essentially emulates our existing app-starters for Processor
+	 * and essentially demonstrates how a function(s) could be applied to an existing
+	 * processor app via {@link IntegrationFlowFunctionSupport} class.
+	 */
+	@EnableBinding(Processor.class)
+	public static class BaseProcessorConfiguration {
+
+		@Autowired
+		private Processor processor;
+
+		@Bean
+		public IntegrationFlow fromChannel(@Nullable IntegrationFlowFunctionSupport functionSupport) {
+
+			IntegrationFlowBuilder flowBuilder = null;
+			if (functionSupport == null) {
+				flowBuilder = IntegrationFlows.from(processor.input()).bridge().channel(processor.output());
+			}
+			else {
+				flowBuilder = functionSupport.integrationFlowFromChannel(processor.input());
+				if (!functionSupport.andThenFunction(flowBuilder, processor.output())) {
+					flowBuilder = flowBuilder.channel(processor.output());
+				}
+			}
+
+			return flowBuilder.get();
+		}
+
+	}
+}

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

@@ -95,7 +95,22 @@ public class SourceToFunctionsSupportTests {
 	public void testFailedInputTypeConversion() {
 		try (ConfigurableApplicationContext context =
 				new SpringApplicationBuilder(
-						TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoContentType.class))
+						TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoConversionPossible.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf",
+								"--spring.jmx.enabled=false")) {
+			PollableChannel errorChannel = context.getBean("errorChannel", PollableChannel.class);
+			OutputDestination target = context.getBean(OutputDestination.class);
+			assertNull(target.receive(1000));
+			assertNotNull(errorChannel.receive(1000));
+		}
+	}
+
+	@Test
+	public void testComposedFunctionIsAppliedToExistingMessageSourceFailedTypeConversion() {
+		try (ConfigurableApplicationContext context =
+				new SpringApplicationBuilder(
+						TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoConversionPossible.class))
 						.web(WebApplicationType.NONE)
 						.run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf",
 								"--spring.jmx.enabled=false")) {
@@ -206,7 +221,7 @@ public class SourceToFunctionsSupportTests {
 
 	@EnableAutoConfiguration
 	@Import(ExistingMessageSourceConfigurationNoContentTypeSet.class)
-	public static class FunctionsConfigurationNoContentType {
+	public static class FunctionsConfigurationNoConversionPossible {
 
 		@Bean
 		public PollableChannel errorChannel() {
@@ -214,13 +229,13 @@ public class SourceToFunctionsSupportTests {
 		}
 
 		@Bean
-		public Function<String, String> toUpperCase() {
-			return String::toUpperCase;
+		public Function<Boolean, Boolean> toUpperCase() {
+			return x -> true;
 		}
 
 		@Bean
-		public Function<String, String> concatWithSelf() {
-			return x -> x + ":" + x;
+		public Function<Boolean, Integer> concatWithSelf() {
+			return x -> 1;
 		}
 
 	}