Provide new config options for streams enabling composition

User can switch off source or sink behaviour (the default is to bind
to input and output streams), and then configure the name of a
supplier (for a source) or consumer (for a sink).

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

@@ -17,9 +17,11 @@
 package org.springframework.cloud.function.stream.config;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionOutcome;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.condition.SpringBootCondition;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
@@ -27,40 +29,223 @@ import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
 import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.core.type.AnnotatedTypeMetadata;
 
 /**
  * @author Mark Fisher
  * @author Marius Bogoevici
  */
+@Configuration
 @EnableConfigurationProperties(StreamConfigurationProperties.class)
 @ConditionalOnClass(Binder.class)
 @ConditionalOnBean(FunctionCatalog.class)
 @ConditionalOnProperty(name = "spring.cloud.stream.enabled", havingValue = "true", matchIfMissing = true)
-@EnableBinding(Processor.class)
 public class StreamAutoConfiguration {
 
-	@Autowired
-	private StreamConfigurationProperties properties;
-
-	@Bean
+	@Configuration
 	// Because of the underlying behaviour of Spring AMQP etc., sources do not start
 	// up and fail gracefully if the broker is down. So we need a flag to be able to
 	// switch this off and stop the app failing on startup.
-	// TODO: find a slicker way to do it (e.g. backoff if the broker is down)
-	@ConditionalOnProperty(name = "spring.cloud.function.stream.supplier.enabled", havingValue = "true", matchIfMissing = true)
-	public SupplierInvokingMessageProducer<Object> supplierInvoker(
-			FunctionCatalog registry) {
-		return new SupplierInvokingMessageProducer<Object>(registry);
+	@ConditionalOnProperty(name = "spring.cloud.function.stream.source.enabled", havingValue = "true", matchIfMissing = true)
+	protected static class SourceConfiguration {
+
+		@Autowired
+		private StreamConfigurationProperties properties;
+
+		@Bean
+		public SupplierInvokingMessageProducer<Object> supplierInvoker(
+				FunctionCatalog registry) {
+			return new SupplierInvokingMessageProducer<Object>(registry, properties.getSource().getName());
+		}
+
 	}
 
-	@Bean
-	public StreamListeningFunctionInvoker functionInvoker(FunctionCatalog registry,
-			FunctionInspector functionInspector,
-			@Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) {
-		return new StreamListeningFunctionInvoker(registry, functionInspector,
-				compositeMessageConverterFactory, properties.getDefaultRoute());
+	@Configuration
+	@ConditionalOnProperty(name = "spring.cloud.function.stream.processor.enabled", havingValue = "true", matchIfMissing = true)
+	@Conditional(SourceAndSinkCondition.class)
+	protected static class ProcessorConfiguration {
+
+		@Autowired
+		private StreamConfigurationProperties properties;
+
+		@Bean
+		public StreamListeningFunctionInvoker functionInvoker(FunctionCatalog registry,
+				FunctionInspector functionInspector,
+				@Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) {
+			return new StreamListeningFunctionInvoker(registry, functionInspector,
+					compositeMessageConverterFactory, properties.getDefaultRoute());
+		}
+
+	}
+
+	@Configuration
+	@Conditional(SinkOnlyCondition.class)
+	protected static class SinkConfiguration {
+
+		@Autowired
+		private StreamConfigurationProperties properties;
+		
+		public SinkConfiguration() {
+		}
+
+		@Bean
+		public StreamListeningConsumerInvoker consumerInvoker(FunctionCatalog registry,
+				FunctionInspector functionInspector,
+				@Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) {
+			return new StreamListeningConsumerInvoker(registry, functionInspector,
+					compositeMessageConverterFactory, properties.getSink().getName());
+		}
+
+	}
+
+	@Configuration
+	@EnableBinding(Processor.class)
+	@Conditional(ProcessorCondition.class)
+	protected class ProcessorBindingConfiguration {
+	}
+
+	@Configuration
+	@EnableBinding(Source.class)
+	@Conditional(SourceCondition.class)
+	protected class SourceBindingConfiguration {
+	}
+
+	@Configuration
+	@EnableBinding(Sink.class)
+	@Conditional(SinkCondition.class)
+	protected class SinkBindingConfiguration {
+	}
+
+	private static class SinkOnlyCondition extends SpringBootCondition {
+		private SourceAndSinkCondition processor = new SourceAndSinkCondition();
+		
+		private SinkCondition sink = new SinkCondition();
+
+		@Override
+		public ConditionOutcome getMatchOutcome(ConditionContext context,
+				AnnotatedTypeMetadata metadata) {
+			if (processor.matches(context, metadata)) {
+				return ConditionOutcome.noMatch("Source is provided by Processor");
+			}
+			if (sink.matches(context, metadata)) {
+				return ConditionOutcome.match("Sink is explicitly enabled");
+			}
+			return ConditionOutcome.noMatch("Sink is not enabled and not available through Processor");
+		}
+	}
+
+	private static class SourceAndSinkCondition extends SpringBootCondition {
+		private SourceCondition source = new SourceCondition();
+
+		private SinkCondition sink = new SinkCondition();
+
+		private ProcessorCondition processor = new ProcessorCondition();
+
+		@Override
+		public ConditionOutcome getMatchOutcome(ConditionContext context,
+				AnnotatedTypeMetadata metadata) {
+			if (processor.matches(context, metadata)) {
+				return ConditionOutcome.match("Processor is bound");
+			}
+			if (sink.matches(context, metadata) && source.matches(context, metadata)) {
+				return ConditionOutcome.match("Both Source and Sink are bound");
+			}
+			return ConditionOutcome.noMatch("Both Source and Sink are not bound");
+		}
+	}
+
+	private static class ProcessorCondition extends SpringBootCondition {
+
+		private SourceCondition source = new SourceCondition();
+
+		private SinkCondition sink = new SinkCondition();
+
+		@Override
+		public ConditionOutcome getMatchOutcome(ConditionContext context,
+				AnnotatedTypeMetadata metadata) {
+			return (!source.matches(context, metadata)
+					&& !sink.matches(context, metadata))
+							? ConditionOutcome.match(
+									"Neither source nor sink is explicitly disabled")
+							: ConditionOutcome.noMatch(
+									"Either sink or source was explicitly disabled");
+		}
+
+	}
+
+	private static class SourceCondition extends SpringBootCondition {
+
+		@Override
+		public ConditionOutcome getMatchOutcome(ConditionContext context,
+				AnnotatedTypeMetadata metadata) {
+			Boolean enabled = context.getEnvironment().getProperty(
+					"spring.cloud.function.stream.source.enabled", Boolean.class);
+			Boolean sink = context.getEnvironment().getProperty(
+					"spring.cloud.function.stream.sink.enabled", Boolean.class, true);
+			if (enabled != null && enabled) {
+				if (!sink) {
+					return ConditionOutcome
+							.match("Source explicitly enabled and sink disabled");
+				}
+				else {
+					return ConditionOutcome
+							.noMatch("Source explicitly enabled and sink enabled");
+				}
+			}
+			if (enabled == null) {
+				if (!sink) {
+					return ConditionOutcome
+							.match("Source implicitly enabled and sink disabled");
+				}
+				else {
+					return ConditionOutcome
+							.noMatch("Source not explicitly enabled and sink enabled");
+				}
+			}
+			return ConditionOutcome.noMatch("Source explicitly disabled");
+		}
+
+	}
+
+	private static class SinkCondition extends SpringBootCondition {
+
+		@Override
+		public ConditionOutcome getMatchOutcome(ConditionContext context,
+				AnnotatedTypeMetadata metadata) {
+			Boolean enabled = context.getEnvironment().getProperty(
+					"spring.cloud.function.stream.sink.enabled", Boolean.class);
+			Boolean source = context.getEnvironment().getProperty(
+					"spring.cloud.function.stream.source.enabled", Boolean.class, true);
+			if (enabled != null && enabled) {
+				if (!source) {
+					return ConditionOutcome
+							.match("Sink explicitly enabled and source disabled");
+				}
+				else {
+					return ConditionOutcome
+							.noMatch("Sink explicitly enabled and source enabled");
+				}
+			}
+			if (enabled == null) {
+				if (!source) {
+					return ConditionOutcome
+							.match("Sink implicitly enabled and source disabled");
+				}
+				else {
+					return ConditionOutcome
+							.noMatch("Sink not explicitly enabled and source enabled");
+				}
+			}
+			return ConditionOutcome.noMatch("Sink explicitly disabled");
+		}
+
 	}
 
 }

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

@@ -24,33 +24,135 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 @ConfigurationProperties(prefix = "spring.cloud.function.stream")
 public class StreamConfigurationProperties {
 
-	/**
-	 * The default route for a message if more than one is available and no explicit route
-	 * key is provided.
-	 */
-	private String defaultRoute;
+	private Source source = new Source();
 
-	/**
-	 * Interval to be used for the Duration (in milliseconds) of a non-Flux producing
-	 * Supplier. Default is 0, which means the Supplier will only be invoked once.
-	 */
-	private long interval = 0L;
+	private Sink sink = new Sink();
+
+	private Processor processor = new Processor();
 
 	public static final String ROUTE_KEY = "stream_routekey";
 
+	public Sink getSink() {
+		return this.sink;
+	}
+
+	public Source getSource() {
+		return this.source;
+	}
+
+	public Processor getProcessor() {
+		return this.processor;
+	}
+
+	public static class Sink {
+
+		/**
+		 * The name of a single consumer to wire up to the input channel. Default is null,
+		 * which means all consumers are bound.
+		 */
+		private String name;
+
+		/**
+		 * Flag to be able to switch off binding consumers to input streams.
+		 */
+		private boolean enabled;
+
+		public String getName() {
+			return this.name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public boolean isEnabled() {
+			return this.enabled;
+		}
+
+		public void setEnabled(boolean enabled) {
+			this.enabled = enabled;
+		}
+	}
+
+	public static class Source {
+
+		/**
+		 * The name of a single supplier to wire up to the output channel. Default is
+		 * null, which means all suppliers are bound.
+		 */
+		private String name;
+
+		/**
+		 * Flag to be able to switch off binding suppliers to output streams. Because of
+		 * the underlying behaviour of Spring AMQP etc., sources do not start up and fail
+		 * gracefully if the broker is down. So this flag is needed to control the
+		 * behaviour if you know the broker is not available and there are suppliers.
+		 */
+		private boolean enabled;
+
+		/**
+		 * Interval to be used for the Duration (in milliseconds) of a non-Flux producing
+		 * Supplier. Default is 0, which means the Supplier will only be invoked once.
+		 */
+		private long interval = 0L;
+
+		public long getInterval() {
+			return interval;
+		}
+
+		public void setInterval(long interval) {
+			this.interval = interval;
+		}
+
+		public boolean isEnabled() {
+			return this.enabled;
+		}
+
+		public void setEnabled(boolean enabled) {
+			this.enabled = enabled;
+		}
+
+		public String getName() {
+			return this.name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+
+	public static class Processor {
+
+		/**
+		 * The name of a single processor to wire up to the input and output channels.
+		 * Default is null, which means all functions are bound.
+		 */
+		private String name;
+
+		/**
+		 * Flag to be able to switch off binding consumers to input streams.
+		 */
+		private boolean enabled;
+
+		public String getName() {
+			return this.name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public boolean isEnabled() {
+			return this.enabled;
+		}
+
+		public void setEnabled(boolean enabled) {
+			this.enabled = enabled;
+		}
+	}
+
 	public String getDefaultRoute() {
-		return defaultRoute;
+		return processor.getName()!=null ? processor.getName() : sink.getName();
 	}
 
-	public void setDefaultRoute(String defaultRoute) {
-		this.defaultRoute = defaultRoute;
-	}
-
-	public long getInterval() {
-		return interval;
-	}
-
-	public void setInterval(long interval) {
-		this.interval = interval;
-	}
 }

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

@@ -0,0 +1,194 @@
+/*
+ * Copyright 2016 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.function.stream.config;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.cloud.function.context.FunctionCatalog;
+import org.springframework.cloud.function.context.catalog.FunctionInspector;
+import org.springframework.cloud.function.context.message.MessageUtils;
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.MessageConverter;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ */
+public class StreamListeningConsumerInvoker implements SmartInitializingSingleton {
+
+	private final FunctionInspector functionInspector;
+
+	private final FunctionCatalog functionCatalog;
+
+	private final CompositeMessageConverterFactory converterFactory;
+
+	private MessageConverter converter;
+
+	private final String defaultRoute;
+
+	private final Map<String, FluxMessageProcessor> processors = new HashMap<>();
+
+	private int count = -1;
+
+	private static final FluxMessageProcessor NOENDPOINT = flux -> Flux.empty();
+
+	private static final Object UNCONVERTED = new Object();
+
+	public StreamListeningConsumerInvoker(FunctionCatalog functionCatalog,
+			FunctionInspector functionInspector,
+			CompositeMessageConverterFactory converterFactory, String defaultRoute) {
+		this.functionCatalog = functionCatalog;
+		this.functionInspector = functionInspector;
+		this.converterFactory = converterFactory;
+		this.defaultRoute = defaultRoute;
+	}
+
+	@Override
+	public void afterSingletonsInstantiated() {
+		this.converter = this.converterFactory.getMessageConverterForAllRegistered();
+	}
+
+	@StreamListener
+	public void handle(@Input(Processor.INPUT) Flux<Message<?>> input) {
+		input.groupBy(this::select).flatMap(group -> group.key().process(group)).subscribe();
+	}
+
+	private Flux<Message<?>> consumer(String name, Flux<Message<?>> flux) {
+		Consumer<Object> consumer = functionCatalog.lookup(Consumer.class, name);
+		consumer.accept(flux.map(message -> convertInput(consumer).apply(message))
+				.filter(transformed -> transformed != UNCONVERTED));
+		return Flux.empty();
+	}
+
+	private Flux<Message<?>> balance(List<String> names, Flux<Message<?>> flux) {
+		if (names.isEmpty()) {
+			return Flux.empty();
+		}
+		String name = choose(names);
+		if (functionCatalog.lookup(Consumer.class, name) != null) {
+			return consumer(name, flux);
+		}
+		return Flux.empty();
+	}
+
+	private synchronized String choose(List<String> names) {
+		if (++count >= names.size() || count < 0) {
+			count = 0;
+		}
+		return names.get(count);
+	}
+
+	private FluxMessageProcessor select(Message<?> input) {
+		String name = null;
+		if (input.getHeaders().containsKey(StreamConfigurationProperties.ROUTE_KEY)) {
+			String key = (String) input.getHeaders()
+					.get(StreamConfigurationProperties.ROUTE_KEY);
+			name = stash(key);
+		}
+		if (name == null && defaultRoute != null) {
+			name = stash(defaultRoute);
+		}
+		if (name == null) {
+			Set<String> names = new LinkedHashSet<>(
+					functionCatalog.getNames(Consumer.class));
+			List<String> matches = new ArrayList<>();
+			if (names.size() == 1) {
+				String key = names.iterator().next();
+				name = stash(key);
+			}
+			else {
+				for (String candidate : names) {
+					Object function = functionCatalog.lookup(Consumer.class, candidate);
+					if (function == null) {
+						continue;
+					}
+					Class<?> inputType = functionInspector.getInputType(function);
+					Object value = this.converter.fromMessage(input, inputType);
+					if (value != null && inputType.isInstance(value)) {
+						matches.add(candidate);
+					}
+				}
+				if (matches.size() == 1) {
+					name = stash(matches.iterator().next());
+				}
+				else {
+					// TODO: do we really want this? Or maybe warn that it is happening?
+					return flux -> balance(matches, flux);
+				}
+			}
+		}
+		if (name == null) {
+			return NOENDPOINT;
+		}
+		return processors.get(name);
+	}
+
+	private String stash(String key) {
+		if (functionCatalog.lookup(Consumer.class, key) != null) {
+			if (!processors.containsKey(key)) {
+				processors.put(key, flux -> consumer(key, flux));
+			}
+			return key;
+		}
+		return null;
+	}
+
+	private Function<Message<?>, Object> convertInput(Object function) {
+		Class<?> inputType = functionInspector.getInputType(function);
+		return m -> {
+			if (functionInspector.isMessage(function)) {
+				return MessageUtils.create(function, convertPayload(inputType, m),
+						m.getHeaders());
+			}
+			else {
+				return convertPayload(inputType, m);
+			}
+		};
+	}
+
+	private Object convertPayload(Class<?> inputType, Message<?> m) {
+		Object result;
+		if (inputType.isAssignableFrom(m.getPayload().getClass())) {
+			result = m.getPayload();
+		}
+		else {
+			result = this.converter.fromMessage(m, inputType);
+		}
+		if (result == null) {
+			result = UNCONVERTED;
+		}
+		return result;
+	}
+
+	interface FluxMessageProcessor {
+		Flux<Message<?>> process(Flux<Message<?>> flux);
+	}
+
+}

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

@@ -28,6 +28,7 @@ import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.StringUtils;
 
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
@@ -44,15 +45,24 @@ public class SupplierInvokingMessageProducer<T> extends MessageProducerSupport {
 
 	private final Map<String, Disposable> disposables = new HashMap<>();
 
-	public SupplierInvokingMessageProducer(FunctionCatalog registry) {
+	private String defaultRoute;
+
+	public SupplierInvokingMessageProducer(FunctionCatalog registry,
+			String defaultRoute) {
 		this.functionCatalog = registry;
+		this.defaultRoute = defaultRoute;
 		this.setOutputChannelName(Source.OUTPUT);
 	}
 
 	@Override
 	protected void doStart() {
-		for (String name : functionCatalog.getNames(Supplier.class)) {
-			start(name);
+		if (StringUtils.hasText(this.defaultRoute)) {
+			start(this.defaultRoute);
+		}
+		else {
+			for (String name : functionCatalog.getNames(Supplier.class)) {
+				start(name);
+			}
 		}
 	}
 
@@ -83,7 +93,8 @@ public class SupplierInvokingMessageProducer<T> extends MessageProducerSupport {
 		if (!disposables.containsKey(name)) {
 			synchronized (disposables) {
 				if (!disposables.containsKey(name)) {
-					Supplier<Flux<?>> supplier = functionCatalog.lookup(Supplier.class, name);
+					Supplier<Flux<?>> supplier = functionCatalog.lookup(Supplier.class,
+							name);
 					if (supplier != null) {
 						suppliers.add(name);
 						disposables.put(name,

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

@@ -43,17 +43,15 @@ import static org.assertj.core.api.Assertions.assertThat;
  * @author Marius Bogoevici
  */
 @RunWith(SpringRunner.class)
-@SpringBootTest(classes = PojoStreamingExplicitEndpointTests.StreamingFunctionApplication.class, properties = {
-		"spring.cloud.function.stream.default-route=uppercase",
-		"logging.level.org.springframework.integration=DEBUG", "debug=TRUE" })
-public class PojoStreamingExplicitEndpointTests {
+@SpringBootTest(classes = PojoStreamingExplicitProcessorEnabledTests.StreamingFunctionApplication.class, properties = "spring.cloud.function.stream.processor.name=uppercase")
+public class PojoStreamingExplicitProcessorEnabledTests {
 
 	@Autowired
 	Processor processor;
 
 	@Autowired
 	MessageCollector messageCollector;
-	
+
 	@Autowired
 	StreamingFunctionApplication app;
 
@@ -68,14 +66,15 @@ public class PojoStreamingExplicitEndpointTests {
 
 	@Test
 	public void testRoutingBeatsDefaultEndpoint() throws Exception {
-		processor.input()
-				.send(MessageBuilder.withPayload("{\"name\":\"hello\"}").setHeader(StreamConfigurationProperties.ROUTE_KEY, "sink").build());
+		processor.input().send(MessageBuilder.withPayload("{\"name\":\"hello\"}")
+				.setHeader(StreamConfigurationProperties.ROUTE_KEY, "sink").build());
 		assertThat(app.foos).hasSize(1);
+		assertThat(app.foos.get(0).getName()).isEqualTo("hello");
 	}
 
 	@SpringBootApplication
 	public static class StreamingFunctionApplication {
-		
+
 		private List<Foo> foos = new ArrayList<>();
 
 		@Bean

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

@@ -0,0 +1,122 @@
+/*
+ * Copyright 2017 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.function.stream.mixed;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Marius Bogoevici
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = PojoStreamingExplicitSinkEnabledTests.StreamingFunctionApplication.class, properties = {
+		"spring.cloud.function.stream.source.enabled=false",
+		"spring.cloud.function.stream.sink.name=uppercase,sink" })
+public class PojoStreamingExplicitSinkEnabledTests {
+
+	@Autowired
+	Sink sink;
+
+	@Autowired
+	List<Bar> collector;
+
+	@Before
+	public void init() {
+		collector.clear();
+	}
+
+	@Test
+	public void routing() throws Exception {
+		sink.input().send(MessageBuilder.withPayload("{\"name\":\"hello\"}").build());
+		assertThat(collector).hasSize(1);
+		assertThat(collector.get(0).getName()).isEqualTo("HELLO");
+	}
+
+	@SpringBootApplication
+	public static class StreamingFunctionApplication {
+
+		@Bean
+		public Function<Foo, Bar> uppercase() {
+			return f -> new Bar(f.getName().toUpperCase());
+		}
+
+		@Bean
+		public List<Bar> collector() {
+			return new ArrayList<>();
+		}
+
+		@Bean
+		public Consumer<Bar> sink(final List<Bar> list) {
+			return s -> list.add(s);
+		}
+
+	}
+
+	protected static class Foo {
+		private String name;
+
+		Foo() {
+		}
+
+		public Foo(String name) {
+			this.name = name;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+
+	protected static class Bar {
+		private String name;
+
+		Bar() {
+		}
+
+		public Bar(String name) {
+			this.name = name;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+}

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

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2017 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.function.stream.mixed;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.test.binder.MessageCollector;
+import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.Message;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = PojoStreamingExplicitSourceEnabledTests.StreamingFunctionApplication.class, properties = {
+		"spring.cloud.function.stream.sink.enabled=false",
+		"spring.cloud.function.stream.source.name=words,uppercase" })
+public class PojoStreamingExplicitSourceEnabledTests {
+
+	@Autowired
+	Source source;
+
+	@Autowired
+	MessageCollector messageCollector;
+
+	@Test
+	public void routing() throws Exception {
+		Message<?> message = messageCollector.forChannel(source.output()).poll(1000,
+				TimeUnit.MILLISECONDS);
+		assertThat(((Bar) message.getPayload()).getName()).isEqualTo("FOO");
+	}
+
+	@SpringBootApplication
+	public static class StreamingFunctionApplication {
+
+		@Bean
+		public Function<Foo, Bar> uppercase() {
+			return f -> new Bar(f.getName().toUpperCase());
+		}
+
+		@Bean
+		public List<Bar> collector() {
+			return new ArrayList<>();
+		}
+
+		@Bean
+		public Supplier<Flux<Foo>> words(final List<Bar> list) {
+			return () -> Flux.just(new Foo("foo"), new Foo("bar"));
+		}
+
+	}
+
+	protected static class Foo {
+		private String name;
+
+		Foo() {
+		}
+
+		public Foo(String name) {
+			this.name = name;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+
+	protected static class Bar {
+		private String name;
+
+		Bar() {
+		}
+
+		public Bar(String name) {
+			this.name = name;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+}