GH-1742 Support Batch Listeners in Core (Function)

- Support `batchMode` in function properties so, for binders that support batch listeners, a message payload of `List<byte[]>` can be properly converted. - In batch mode, handle `public Function<List<Person>, Person> func()` `public Function<List<List<Person>>, Person> func()` `Function<Message<List<Person>>, Person> func()` - Also use `ParameterizedType` as a conversion hint for `public Function<List<Person>, Person> func()` when not in batch mode (json: `[{\"name\":\"bob\"},{\"name\":\"jill\"}]`). Previously, just the' class was used without generic type information. Move batchMode to ConsumerProperties. Fix imports Remove duplicate test from if test for message conversion.
2019-06-19 14:47:50 -04:00
parent 562a025cda
commit 9fd674909a
8 changed files with 1255 additions and 30 deletions
--- a/docs/src/main/asciidoc/spring-cloud-stream.adoc
+++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc
@@ -587,6 +587,8 @@ support.
 [[spring_cloud_function]]
 ==== Spring Cloud Function support

+===== Overview
+
 Since Spring Cloud Stream v2.1, another alternative for defining _stream handlers_ and _sources_ is to use build-in
 support for https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] where they can be expressed as beans of
 type `java.util.function.[Supplier/Function/Consumer]`.
@@ -712,7 +714,19 @@ For example, the above composition could be defined as such (if both functions p
 --spring.cloud.stream.function.definition=reactiveUpperCase|wrapInQuotes
 ----

+===== Batch Consumers

+When using a `MessageChannelBinder` that supports batch listeners, and the feature is enabled for the consumer binding, you can set `spring.cloud.stream.function.definition` to `true` to enable the entire batch of messages to be passed to the function in a `List`.
+
+====
+[source, java]
+----
+@Bean
+public Function<List<Person>, Person> findFirstPerson() {
+    return persons -> persons.get(0);
+}
+----
+====

 [[spring-cloud-streams-overview-using-polled-consumers]]
 ==== Using Polled Consumers
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java
@@ -17,12 +17,15 @@
 package org.springframework.cloud.stream.binder;

 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;

 import javax.validation.constraints.Min;

 import com.fasterxml.jackson.annotation.JsonInclude;

+import org.springframework.messaging.Message;
+
 /**
 * Common consumer properties - spring.cloud.stream.bindings.[destinationName].consumer.
 *
@@ -157,6 +160,15 @@ public class ConsumerProperties {
 	 */
 	private boolean multiplex;

+	/**
+	 * When set to true, if the binder supports it, the messages emitted will have a {@link List}
+	 * payload; When used in conjunction with functions, the function can receive a list of
+	 * objects (or {@link Message}s) with the payloads converted if necessary.
+	 *
+	 * @since 3.0
+	 */
+	private boolean batchMode;
+
 	public String getRetryTemplateName() {
 		return retryTemplateName;
 	}
@@ -285,4 +297,12 @@ public class ConsumerProperties {
 		this.autoStartup = autoStartup;
 	}

+	public boolean isBatchMode() {
+		return this.batchMode;
+	}
+
+	public void setBatchMode(boolean batchMode) {
+		this.batchMode = batchMode;
+	}
+
 }
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java
@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.converter;

 import java.io.IOException;
 import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -44,7 +45,7 @@ import org.springframework.messaging.converter.MessageConversionException;
 */
 class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter {

-	private final Map<ParameterizedTypeReference<?>, JavaType> typeCache = new ConcurrentHashMap<>();
+	private final Map<Type, JavaType> typeCache = new ConcurrentHashMap<>();

 	ApplicationJsonMessageMarshallingConverter(@Nullable ObjectMapper objectMapper) {
 		if (objectMapper != null) {
@@ -67,8 +68,8 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 	}

 	@Override
-	protected Object convertFromInternal(Message<?> message, Class<?> targetClass,
-			@Nullable Object conversionHint) {
+	protected Object convertFromInternal(Message<?> message, Class<?> targetClass, @Nullable Object hint) {
+		Object conversionHint = hint;
 		Object result = null;
 		if (conversionHint instanceof MethodParameter) {
 			Class<?> conversionHintType = ((MethodParameter) conversionHint)
@@ -87,12 +88,14 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 				ParameterizedTypeReference<Object> forType = ParameterizedTypeReference
 						.forType(((MethodParameter) conversionHint)
 								.getGenericParameterType());
-				result = convertParameterizedType(message, targetClass, forType);
+				result = convertParameterizedType(message, forType.getType());
 			}
 		}
 		else if (conversionHint instanceof ParameterizedTypeReference) {
-			result = convertParameterizedType(message, targetClass,
-					(ParameterizedTypeReference<?>) conversionHint);
+			result = convertParameterizedType(message, ((ParameterizedTypeReference<?>) conversionHint).getType());
+		}
+		else if (conversionHint instanceof ParameterizedType) {
+			result = convertParameterizedType(message, (Type) conversionHint);
 		}

 		if (result == null) {
@@ -109,15 +112,14 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 		return result;
 	}

-	private Object convertParameterizedType(Message<?> message, Class<?> targetClass,
-			ParameterizedTypeReference<?> conversionHint) {
+	private Object convertParameterizedType(Message<?> message, Type conversionHint) {
 		ObjectMapper objectMapper = this.getObjectMapper();
 		Object payload = message.getPayload();
 		try {
 			JavaType type = this.typeCache.get(conversionHint);
 			if (type == null) {
 				type = objectMapper.getTypeFactory()
-						.constructType((conversionHint).getType());
+						.constructType(conversionHint);
 				this.typeCache.put(conversionHint, type);
 			}
 			if (payload instanceof byte[]) {
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
@@ -17,23 +17,26 @@
 package org.springframework.cloud.stream.function;

 import java.lang.reflect.Field;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
+import java.util.stream.Collectors;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;

-import org.springframework.beans.factory.BeanFactory;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.FunctionType;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ProducerProperties;
-import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
 import org.springframework.integration.support.MessageBuilder;
@@ -52,6 +55,7 @@ import org.springframework.util.ReflectionUtils;
 * @author Oleg Zhurakousky
 * @author David Turanski
 * @author Tolga Kavukcu
+ * @author Gary Russell
 * @since 2.1
 */
 class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O>>> {
@@ -68,16 +72,22 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O

 	private final Class<?> inputClass;

+	private final ParameterizedType inputParameterizedType;
+
 	private final Class<?> outputClass;

 	private final Function<Flux<?>, Flux<?>> userFunction;

 	private final CompositeMessageConverter messageConverter;

-	private final BeanFactory beanFactory;
+	private final MessageChannel errorChannel;

 	private final boolean isInputArgumentMessage;

+	private final Class<?> messagePayloadClass;
+
+	private final Type messagePayloadType;
+
 	private final ConsumerProperties consumerProperties;

 	private final ProducerProperties producerProperties;
@@ -86,6 +96,12 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O

 	private final StreamFunctionProperties functionProperties;

+	private final boolean batchMode;
+
+	private final Type listContentParameterizedType;
+
+	private final Class<?> listContentClass;
+
 	FunctionInvoker(StreamFunctionProperties functionProperties,
 			FunctionCatalog functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory) {
@@ -97,7 +113,7 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 	FunctionInvoker(StreamFunctionProperties functionProperties,
 			FunctionCatalog functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory,
-			BeanFactory beanFactory) {
+			MessageChannel errorChannel) {

 		this.functionProperties = functionProperties;
 		Object originalUserFunction = functionCatalog
@@ -113,12 +129,63 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 		this.isInputArgumentMessage = functionType.isMessage();
 		this.inputClass = functionType.getInputType();
 		this.outputClass = functionType.getOutputType();
-		this.beanFactory = beanFactory;
+		this.errorChannel = errorChannel;
 		this.bindingServiceProperties = functionProperties.getBindingServiceProperties();
 		this.consumerProperties = this.bindingServiceProperties
 				.getConsumerProperties(functionProperties.getInputDestinationName());
 		this.producerProperties = this.bindingServiceProperties
 				.getProducerProperties(functionProperties.getOutputDestinationName());
+		this.batchMode = this.consumerProperties.isBatchMode();
+		Type type = functionType.getType();
+		ParameterizedType functionInputParameterizedType = null;
+		Type listContainsType = null;
+		Type payloadType = null;
+		if (type instanceof ParameterizedType) {
+			Type functionInputType = ((ParameterizedType) type).getActualTypeArguments()[0];
+			if (functionInputType instanceof ParameterizedType) {
+				functionInputParameterizedType = (ParameterizedType) functionInputType;
+				Type rawType = ((ParameterizedType) functionInputType).getRawType();
+				if (rawType.equals(List.class)) {
+					listContainsType = ((ParameterizedType) functionInputType).getActualTypeArguments()[0];
+				}
+				else if (rawType.equals(Message.class)) {
+					payloadType = determinePayloadType(functionInputType);
+				}
+			}
+		}
+		if (listContainsType instanceof Class) {
+			this.listContentClass = (Class<?>) listContainsType;
+			this.listContentParameterizedType = null;
+		}
+		else {
+			this.listContentClass = Object.class;
+			this.listContentParameterizedType = listContainsType;
+		}
+		if ((functionInputParameterizedType != null && functionInputParameterizedType.getRawType().equals(Flux.class))
+				|| payloadType != null) {
+			functionInputParameterizedType = null;
+		}
+		this.inputParameterizedType = functionInputParameterizedType;
+		if (payloadType instanceof Class) {
+			this.messagePayloadClass = (Class<?>) payloadType;
+			this.messagePayloadType = null;
+		}
+		else {
+			this.messagePayloadClass = Object.class;
+			this.messagePayloadType = payloadType;
+		}
+	}
+
+	private Type determinePayloadType(Type functionInputType) {
+		Type payloadType;
+		payloadType = ((ParameterizedType) functionInputType).getActualTypeArguments()[0];
+		if (payloadType instanceof ParameterizedType) {
+			Type payloadRawType = ((ParameterizedType) payloadType).getRawType();
+			if (payloadRawType.equals(List.class)) {
+				payloadType = ((ParameterizedType) payloadType).getActualTypeArguments()[0];
+			}
+		}
+		return payloadType;
 	}

 	@Override
@@ -127,7 +194,8 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O

 		return input.concatMap(message -> {
 			return Flux.just(message).doOnNext(originalMessageRef::set)
-					.map(this::resolveArgument).transform(this.userFunction::apply)
+					.map(this::resolveArgument)
+					.transform(this.userFunction::apply)
 					.retryBackoff(this.consumerProperties.getMaxAttempts(),
 							Duration.ofMillis(
 									this.consumerProperties.getBackOffInitialInterval()),
@@ -143,19 +211,10 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 	}

 	private void onError(Throwable t, Message<I> originalMessage) {
-		String inputDestinationName = functionProperties.getInputDestinationName();
-		BindingProperties bindingProperties = functionProperties.getBindingServiceProperties().getBindings().get(inputDestinationName);
-		String destinationName = bindingProperties.getDestination();
-		String groupName = bindingProperties.getGroup();
-		String bindingErrorChannelName = destinationName + "." + groupName + ".errors";
-
-		if (beanFactory != null) {
-			MessageChannel errorChannel = beanFactory.containsBean(bindingErrorChannelName)
-					? beanFactory.getBean(bindingErrorChannelName, MessageChannel.class)
-							:  beanFactory.getBean("errorChannel", MessageChannel.class);
-			ErrorMessage em = new ErrorMessage(t, originalMessage.getHeaders(), (Message<?>) originalMessage);
+		if (this.errorChannel != null) {
+			ErrorMessage em = new ErrorMessage(t, originalMessage);
 			logger.error(em);
-			errorChannel.send(em);
+			this.errorChannel.send(em);
 		}
 		else {
 			logger.error(t);
@@ -221,19 +280,48 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 		}

 		T argument = (T) (shouldConvertFromMessage(message)
-				? this.messageConverter.fromMessage(message, this.inputClass) : message);
+				? this.messageConverter.fromMessage(message, this.inputClass, this.inputParameterizedType) : message);
 		Assert.notNull(argument, "Failed to resolve argument type '" + this.inputClass
 				+ "' from message: " + message);
-		if (this.isInputArgumentMessage && !(argument instanceof Message)) {
+		if (this.batchMode
+				&& this.messagePayloadClass != null
+				&& this.isInputArgumentMessage
+				&& argument instanceof Message
+				&& ((Message<?>) argument).getPayload() instanceof List
+				&& !this.messagePayloadClass.isAssignableFrom(((Message<?>) argument).getPayload().getClass())) {
+			argument = (T) MessageBuilder
+					.withPayload(convertListContents(message.getPayload(), this.messagePayloadClass,
+							this.messagePayloadType))
+					.build();
+		}
+		else if (this.isInputArgumentMessage && !(argument instanceof Message)) {
+			if (shouldBatchConvert(argument)) {
+				argument = convertListContents(argument, this.messagePayloadClass, this.messagePayloadType);
+			}
 			argument = (T) MessageBuilder.withPayload(argument)
 					.copyHeaders(message.getHeaders()).build();
 		}
 		else if (!this.isInputArgumentMessage && argument instanceof Message) {
 			argument = ((Message<T>) argument).getPayload();
+			if (shouldBatchConvert(argument)) {
+				argument = convertListContents(argument, this.listContentClass, this.listContentParameterizedType);
+			}
 		}
 		return argument;
 	}

+	private <T> boolean shouldBatchConvert(T argument) {
+		return this.batchMode && argument instanceof List && this.listContentClass != null;
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T> T convertListContents(T argument, Class<?> targetClass, Type hint) {
+		return (T) ((List<?>) argument).stream()
+				.map(payload -> this.messageConverter.fromMessage(MessageBuilder.withPayload(payload).build(),
+						targetClass, hint))
+				.collect(Collectors.toList());
+	}
+
 	private boolean shouldConvertFromMessage(Message<?> message) {
 		return !this.inputClass.isAssignableFrom(Message.class)
 				&& !this.inputClass.isAssignableFrom(message.getPayload().getClass())
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java
@@ -49,6 +49,8 @@ public class StreamFunctionProperties {

 	private Map<String, List<String>> outputBindings = new HashMap<>();

+	private boolean batchMode;
+
 	private boolean composeTo;

 	private boolean composeFrom;
@@ -110,7 +112,13 @@ public class StreamFunctionProperties {

 	public void setInputBindings(Map<String, List<String>> inputBindings) {
 		this.inputBindings = inputBindings;
+	}

+	public boolean isBatchMode() {
+		return this.batchMode;
+	}

+	public void setBatchMode(boolean batchMode) {
+		this.batchMode = batchMode;
 	}
 }
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java
@@ -33,7 +33,7 @@ public class InputDestination extends AbstractDestination {
 	 * destination (e.g., Processor.INPUT).
 	 * @param message message to send
 	 */
-	public void send(Message<byte[]> message) {
+	public void send(Message<?> message) {
 		this.getChannel().send(message);
 	}

--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java
@@ -0,0 +1,257 @@
+/*
+ * Copyright 2019-2019 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.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+
+import org.junit.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.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 static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Gary Russel
+ * @author Oleg Zhurakousky
+ *
+ * @since 3.0
+ */
+public class FunctionBatchingTests {
+
+	@Test
+	public void testMessageBatchConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						MessageBatchConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func",
+								"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			List<byte[]> list = new ArrayList<>();
+			list.add("{\"name\":\"bob\"}".getBytes());
+			list.add("{\"name\":\"jill\"}".getBytes());
+			Message<List<byte[]>> inputMessage = MessageBuilder
+					.withPayload(list)
+					.build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+
+			context.stop();
+		}
+	}
+
+	@Test
+	public void testListPayloadConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						ListPayloadNotBatchConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes())
+					.build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+
+			context.stop();
+		}
+	}
+
+	@Test
+	public void testSimpleBatchConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						SimpleBatchConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func",
+								"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			List<byte[]> list = new ArrayList<>();
+			list.add("{\"name\":\"bob\"}".getBytes());
+			list.add("{\"name\":\"jill\"}".getBytes());
+			Message<List<byte[]>> inputMessage = MessageBuilder
+					.withPayload(list)
+					.build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+			context.stop();
+		}
+	}
+
+	@Test
+	public void testNestedBatchConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						NestedBatchConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func",
+								"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			List<byte[]> list = new ArrayList<>();
+			list.add("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes());
+			Message<List<byte[]>> inputMessage = MessageBuilder
+					.withPayload(list)
+					.build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+			context.stop();
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class SimpleBatchConfiguration {
+
+		@Bean
+		public Function<List<Person>, Person> func() {
+			return x -> x.get(0);
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+
+	@EnableAutoConfiguration
+	public static class ListPayloadNotBatchConfiguration {
+
+		@Bean
+		public Function<List<Person>, Person> func() {
+			return x -> x.get(0);
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	public static class NestedBatchConfiguration {
+
+		@Bean
+		public Function<List<List<Person>>, Person> func() {
+			return x -> x.get(0).get(0);
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	public static class MessageBatchConfiguration {
+
+		@Bean
+		public Function<Message<List<Person>>, Person> func() {
+			return x -> x.getPayload().get(0);
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+}
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java
@@ -0,0 +1,836 @@
+/*
+ * Copyright 2018-2019 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.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+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.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamMessageConverter;
+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.config.BindingServiceProperties;
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.cloud.stream.function.pojo.Baz;
+import org.springframework.cloud.stream.function.pojo.ErrorBaz;
+import org.springframework.cloud.stream.messaging.Processor;
+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.MessageHeaders;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.util.ReflectionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Oleg Zhurakousky
+ * @author Tolga Kavukcu
+ * @author Gary Russell
+ *
+ */
+public class FunctionInvokerTests {
+
+	private static String testWithFluxedConsumerValue;
+
+	@Test
+	public void testSimpleEchoConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						SimpleEchoConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("{\"name\":\"bob\"}".getBytes()).build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+
+		}
+	}
+
+	@Test
+	public void testFluxPojoFunction() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration
+						.getCompleteConfiguration(SimpleFluxFunctionConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false",
+										"--spring.cloud.stream.function.definition=func")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("{\"name\":\"bob\"}".getBytes()).build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload()).isEqualTo("Person: bob".getBytes());
+
+		}
+	}
+
+	@Test
+	public void testFluxMessagePojoFunction() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						SimpleFluxMessageFunctionConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false",
+										"--spring.cloud.stream.function.definition=func")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("{\"name\":\"bob\"}".getBytes()).build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload()).isEqualTo("Person: bob".getBytes());
+		}
+	}
+
+	@Test
+	public void testFunctionHonorsOutboundBindingContentType() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						ConverterDoesNotProduceCTConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false",
+										"--spring.cloud.stream.function.definition=func",
+										"--spring.cloud.stream.bindings.output.contentType=text/plain")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("{\"name\":\"bob\"}".getBytes())
+					.setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)
+					.toString()).isEqualTo("text/plain");
+
+		}
+	}
+
+	@Test
+	public void testFunctionHonorsConverterSetContentType() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						ConverterInjectingCTConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false",
+										"--spring.cloud.stream.function.definition=func",
+										"--spring.cloud.stream.bindings.output.contentType=text/plain")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("{\"name\":\"bob\"}".getBytes())
+					.setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)
+					.toString()).isEqualTo("ping/pong");
+
+		}
+	}
+
+	@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());
+
+			StreamFunctionProperties functionProperties = createStreamFunctionProperties();
+
+			functionProperties.setDefinition("messageToMessageSameType");
+			FunctionInvoker<Foo, Foo> messageToMessageSameType = new FunctionInvoker<>(
+					functionProperties,
+					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);
+
+			functionProperties.setDefinition("pojoToPojoSameType");
+			FunctionInvoker<Foo, Foo> pojoToPojoSameType = new FunctionInvoker<>(
+					functionProperties,
+					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());
+
+			functionProperties.setDefinition("messageToMessageNoType");
+			FunctionInvoker<Foo, Foo> messageToMessageNoType = new FunctionInvoker<>(
+					functionProperties,
+					context.getBean(FunctionCatalog.class),
+					context.getBean(FunctionInspector.class),
+					context.getBean(CompositeMessageConverterFactory.class));
+			outputMessage = messageToMessageNoType.apply(Flux.just(inputMessage))
+					.blockFirst();
+			assertThat(outputMessage).isInstanceOf(Message.class);
+
+			functionProperties.setDefinition("withException");
+			FunctionInvoker<Foo, Foo> withException = new FunctionInvoker<>(
+					functionProperties,
+					context.getBean(FunctionCatalog.class),
+					context.getBean(FunctionInspector.class),
+					context.getBean(CompositeMessageConverterFactory.class));
+
+			Flux<Message<Foo>> fluxOfMessages = Flux
+					.just(new GenericMessage<>(new ErrorFoo()), inputMessage);
+			Message<Foo> resultMessage = withException.apply(fluxOfMessages).blockFirst();
+			assertThat(resultMessage.getPayload()).isNotInstanceOf(ErrorFoo.class);
+		}
+	}
+
+	@Test
+	public void testNativeEncodingEnabled() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration
+						.getCompleteConfiguration(MyFunctionsConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false")) {
+
+			Message<Baz> inputMessage = new GenericMessage<>(new Baz());
+
+			StreamFunctionProperties functionProperties = createStreamFunctionPropertiesWithNativeEncoding();
+
+			functionProperties.setDefinition("pojoToPojoNonEmptyPojo");
+			FunctionInvoker<Baz, Baz> pojoToPojoSameType = new FunctionInvoker<>(
+					functionProperties,
+					context.getBean(FunctionCatalog.class),
+					context.getBean(FunctionInspector.class),
+					context.getBean(CompositeMessageConverterFactory.class));
+			Message<Baz> outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage))
+					.blockFirst();
+			assertThat(inputMessage.getPayload()).isEqualTo(outputMessage.getPayload());
+
+			Message<Baz> inputMessageWithBaz = new GenericMessage<>(new Baz());
+
+			functionProperties.setDefinition("messageToMessageNoType");
+			FunctionInvoker<Baz, Baz> messageToMessageNoType = new FunctionInvoker<>(
+					functionProperties,
+					context.getBean(FunctionCatalog.class),
+					context.getBean(FunctionInspector.class),
+					context.getBean(CompositeMessageConverterFactory.class));
+			outputMessage = messageToMessageNoType.apply(Flux.just(inputMessageWithBaz))
+					.blockFirst();
+			assertThat(outputMessage).isInstanceOf(Message.class);
+
+			functionProperties.setDefinition("withExceptionNativeEncodingEnabled");
+			FunctionInvoker<Baz, Baz> withException = new FunctionInvoker<>(
+					functionProperties,
+					context.getBean(FunctionCatalog.class),
+					context.getBean(FunctionInspector.class),
+					context.getBean(CompositeMessageConverterFactory.class));
+
+			Flux<Message<Baz>> fluxOfMessages = Flux
+					.just(new GenericMessage<>(new ErrorBaz()), inputMessage);
+			Message<Baz> resultMessage = withException.apply(fluxOfMessages).blockFirst();
+			assertThat(resultMessage.getPayload()).isNotInstanceOf(ErrorFoo.class);
+		}
+	}
+
+	@Test
+	public void testWithOutNativeEncodingEnabled() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration
+						.getCompleteConfiguration(MyFunctionsConfiguration.class))
+								.web(WebApplicationType.NONE)
+								.run("--spring.jmx.enabled=false")) {
+
+			Message<Baz> inputMessage = new GenericMessage<>(new Baz());
+
+			StreamFunctionProperties functionProperties = createStreamFunctionProperties();
+
+			functionProperties.setDefinition("pojoToPojoNonEmptyPojo");
+			FunctionInvoker<Baz, Baz> pojoToPojoSameType = new FunctionInvoker<>(
+					functionProperties,
+					context.getBean(FunctionCatalog.class),
+					context.getBean(FunctionInspector.class),
+					context.getBean(CompositeMessageConverterFactory.class));
+			Message<Baz> outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage))
+					.blockFirst();
+			assertThat(outputMessage).isNotNull();
+			assertThat(inputMessage.getPayload())
+					.isNotEqualTo(outputMessage.getPayload());
+
+		}
+	}
+
+	@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,
+					context.getBean(FunctionCatalog.class),
+					context.getBean(FunctionInspector.class),
+					context.getBean(CompositeMessageConverterFactory.class));
+
+			fluxedConsumer.apply(Flux.just(inputMessage)).blockFirst();
+
+			assertThat(testWithFluxedConsumerValue).isEqualTo(value);
+		}
+	}
+
+	@Test
+	public void testListPayloadConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						ListPayloadNotBatchConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder
+					.withPayload("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes())
+					.build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+
+		}
+	}
+
+	@Test
+	public void testSimpleBatchConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						SimpleBatchConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func",
+								"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			List<byte[]> list = new ArrayList<>();
+			list.add("{\"name\":\"bob\"}".getBytes());
+			list.add("{\"name\":\"jill\"}".getBytes());
+			Message<List<byte[]>> inputMessage = MessageBuilder
+					.withPayload(list)
+					.build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+
+		}
+	}
+
+	@Test
+	public void testNestedBatchConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						NestedBatchConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func",
+								"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			List<byte[]> list = new ArrayList<>();
+			list.add("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes());
+			Message<List<byte[]>> inputMessage = MessageBuilder
+					.withPayload(list)
+					.build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+
+		}
+	}
+
+	@Test
+	public void testMessageBatchConfiguration() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(
+						MessageBatchConfiguration.class)).web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.function.definition=func",
+								"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context
+					.getBean(OutputDestination.class);
+
+			List<byte[]> list = new ArrayList<>();
+			list.add("{\"name\":\"bob\"}".getBytes());
+			list.add("{\"name\":\"jill\"}".getBytes());
+			Message<List<byte[]>> inputMessage = MessageBuilder
+					.withPayload(list)
+					.build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage).isNotNull();
+			assertThat(outputMessage.getPayload())
+					.isEqualTo("{\"name\":\"bob\"}".getBytes());
+
+		}
+	}
+
+	private StreamFunctionProperties createStreamFunctionProperties() {
+		StreamFunctionProperties functionProperties = new StreamFunctionProperties();
+		functionProperties.setInputDestinationName("input");
+		functionProperties.setOutputDestinationName("output");
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3);
+		try {
+			Field f = ReflectionUtils.findField(StreamFunctionProperties.class,
+					"bindingServiceProperties");
+			f.setAccessible(true);
+			f.set(functionProperties, bindingServiceProperties);
+			return functionProperties;
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+	}
+
+	private StreamFunctionProperties createStreamFunctionPropertiesWithNativeEncoding() {
+		StreamFunctionProperties functionProperties = new StreamFunctionProperties();
+		functionProperties.setInputDestinationName("input");
+		functionProperties.setOutputDestinationName("output");
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3);
+		bindingServiceProperties.getProducerProperties("output")
+				.setUseNativeEncoding(true);
+		try {
+			Field bspField = ReflectionUtils.findField(StreamFunctionProperties.class,
+					"bindingServiceProperties");
+			bspField.setAccessible(true);
+			bspField.set(functionProperties, bindingServiceProperties);
+			return functionProperties;
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class SimpleEchoConfiguration {
+
+		@Bean
+		public Function<Person, Person> func() {
+			return x -> x;
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class SimpleFluxFunctionConfiguration {
+
+		@Bean
+		public Function<Flux<Person>, Flux<String>> func() {
+			return x -> x.map(person -> person.toString());
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+			@Override
+			public String toString() {
+				return "Person: " + name;
+			}
+
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class SimpleFluxMessageFunctionConfiguration {
+
+		@Bean
+		public Function<Flux<Message<Person>>, Flux<Message<String>>> func() {
+			return x -> x.map(personMessage -> {
+				Person person = personMessage.getPayload();
+				Message<String> message = MessageBuilder.withPayload(person.toString())
+						.copyHeaders(personMessage.getHeaders()).build();
+				return message;
+			});
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+			@Override
+			public String toString() {
+				return "Person: " + name;
+			}
+
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class ConverterDoesNotProduceCTConfiguration {
+
+		@Bean
+		public Function<String, String> func() {
+			return x -> x;
+		}
+
+		@StreamMessageConverter
+		public MessageConverter customConverter() {
+			return new MessageConverter() {
+
+				@Override
+				public Message<?> toMessage(Object payload, MessageHeaders headers) {
+					return new GenericMessage<byte[]>(((String) payload).getBytes());
+				}
+
+				@Override
+				public Object fromMessage(Message<?> message, Class<?> targetClass) {
+					String contentType = message.getHeaders()
+							.get(MessageHeaders.CONTENT_TYPE).toString();
+					if (contentType.equals("foo/bar")) {
+						return new String((byte[]) message.getPayload());
+					}
+					return null;
+				}
+			};
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class ConverterInjectingCTConfiguration {
+
+		@Bean
+		public Function<String, String> func() {
+			return x -> x;
+		}
+
+		@StreamMessageConverter
+		public MessageConverter customConverter() {
+			return new MessageConverter() {
+
+				@Override
+				public Message<?> toMessage(Object payload, MessageHeaders headers) {
+					return MessageBuilder.withPayload(((String) payload).getBytes())
+							.setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build();
+				}
+
+				@Override
+				public Object fromMessage(Message<?> message, Class<?> targetClass) {
+					String contentType = message.getHeaders()
+							.get(MessageHeaders.CONTENT_TYPE).toString();
+					if (contentType.equals("foo/bar")) {
+						return new String((byte[]) message.getPayload());
+					}
+					return null;
+				}
+			};
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class ListPayloadNotBatchConfiguration {
+
+		@Bean
+		public Function<List<Person>, Person> func() {
+			return x -> x.get(0);
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class SimpleBatchConfiguration {
+
+		@Bean
+		public Function<List<Person>, Person> func() {
+			return x -> x.get(0);
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class NestedBatchConfiguration {
+
+		@Bean
+		public Function<List<List<Person>>, Person> func() {
+			return x -> x.get(0).get(0);
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@EnableBinding(Processor.class)
+	public static class MessageBatchConfiguration {
+
+		@Bean
+		public Function<Message<List<Person>>, Person> func() {
+			return x -> x.getPayload().get(0);
+		}
+
+		public static class Person {
+
+			private String name;
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+		}
+
+	}
+
+	@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();
+		}
+
+		@Bean
+		public Function<Message<?>, Message<?>> messageToMessageAnyType() {
+			return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders())
+					.build();
+		}
+
+		@Bean
+		public Function<Message<?>, Message<?>> messageToMessageNoType() {
+			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;
+		}
+
+		@Bean
+		public Function<Baz, Baz> pojoToPojoNonEmptyPojo() {
+			return x -> x;
+		}
+
+		@Bean
+		public Function<Foo, Foo> withException() {
+			return x -> {
+				if (x instanceof ErrorFoo) {
+					System.out.println("Throwing exception ");
+					throw new RuntimeException("Boom!");
+				}
+				else {
+					System.out.println("All is good ");
+					return x;
+				}
+			};
+		}
+
+		@Bean
+		public Function<Baz, Baz> withExceptionNativeEncodingEnabled() {
+			return x -> {
+				if (x instanceof ErrorBaz) {
+					System.out.println("Throwing exception ");
+					throw new RuntimeException("Boom!");
+				}
+				else {
+					System.out.println("All is good ");
+					return x;
+				}
+			};
+		}
+
+	}
+
+	private static class Foo {
+
+	}
+
+	private static class ErrorFoo extends Foo {
+
+	}
+
+	private static class Bar {
+
+	}
+
+}