add support for simple (non-Flux) types

add objectToStringHttpMessageConverter

CompilerController accepts parameterized types

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

@@ -33,6 +33,8 @@ import org.springframework.boot.autoconfigure.condition.SpringBootCondition;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.function.invoker.AbstractFunctionInvoker;
 import org.springframework.cloud.function.registry.FunctionCatalog;
+import org.springframework.cloud.function.support.FluxFunction;
+import org.springframework.cloud.function.support.FunctionUtils;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.messaging.Processor;
@@ -43,6 +45,7 @@ import org.springframework.context.annotation.ConditionContext;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.ConfigurationCondition;
 import org.springframework.core.type.AnnotatedTypeMetadata;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 import reactor.core.publisher.Flux;
@@ -82,7 +85,8 @@ public class StreamConfiguration {
 		@ConditionalOnProperty("spring.cloud.stream.bindings.input.destination")
 		public AbstractFunctionInvoker<?, ?> invoker(FunctionCatalog registry) {
 			String name = properties.getEndpoint();
-			Function<Flux<Object>, Flux<Object>> function = registry.lookupFunction(name);
+			Function<Object, Object> function = registry.lookupFunction(name);
+			Assert.notNull(function, "no such function: " + name);
 			return new StreamListeningFunctionInvoker(function);
 		}		
 	}

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

@@ -19,25 +19,36 @@ package org.springframework.cloud.function.stream;
 import java.util.function.Function;
 
 import org.springframework.cloud.function.invoker.AbstractFunctionInvoker;
+import org.springframework.cloud.function.support.FluxFunction;
+import org.springframework.cloud.function.support.FunctionUtils;
 import org.springframework.cloud.stream.annotation.Input;
 import org.springframework.cloud.stream.annotation.Output;
 import org.springframework.cloud.stream.annotation.StreamListener;
 import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.util.Assert;
 
 import reactor.core.publisher.Flux;
 
 /**
  * @author Mark Fisher
  */
-public class StreamListeningFunctionInvoker extends AbstractFunctionInvoker<Flux<Object>, Flux<Object>> {
+public class StreamListeningFunctionInvoker extends AbstractFunctionInvoker<Flux<?>, Flux<?>> {
 
-	public StreamListeningFunctionInvoker(Function<Flux<Object>, Flux<Object>> function) {
-		super(function);
+	public StreamListeningFunctionInvoker(Function<?, ?> function) {
+		super(wrapIfNecessary(function));
 	}
 
 	@StreamListener
 	@Output(Processor.OUTPUT)
-	public Flux<Object> handle(@Input(Processor.INPUT) Flux<Object> input) {
+	public Flux<?> handle(@Input(Processor.INPUT) Flux<?> input) {
 		return this.doInvoke(input);
 	}
+
+	private static Function<Flux<?>, Flux<?>> wrapIfNecessary(Function function) {
+		Assert.notNull(function, "Function must not be null");
+		if (!FunctionUtils.isFluxFunction(function)) {
+			function = new FluxFunction(function);
+		}
+		return function;
+	}
 }

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

@@ -18,9 +18,12 @@ package org.springframework.cloud.function.stream;
 
 import java.util.function.Supplier;
 
+import org.springframework.cloud.function.support.FluxSupplier;
+import org.springframework.cloud.function.support.FunctionUtils;
 import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
 
 import reactor.core.publisher.Flux;
 
@@ -31,8 +34,12 @@ public class SupplierInvokingMessageProducer<T> extends MessageProducerSupport {
 
 	private final Supplier<Flux<T>> supplier;
 
-	public SupplierInvokingMessageProducer(Supplier<Flux<T>> supplier) {
-		this.supplier = supplier;
+	public SupplierInvokingMessageProducer(Supplier<?> supplier) {
+		Assert.notNull(supplier, "Supplier must not be null");
+		if (!FunctionUtils.isFluxSupplier(supplier)) {
+			supplier = new FluxSupplier<>(supplier);
+		}
+		this.supplier = (Supplier<Flux<T>>) supplier;
 		this.setOutputChannelName(Source.OUTPUT);
 	}