Only expose Publisher via FunctionCatalog

Flux.from() is cheap and can be used to marshal the inputs everywhere
internally. With this change users ought to be able to register any
function of any Publisher type.

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/SpringBootRequestHandler.java

@@ -23,6 +23,8 @@ import java.util.List;
 import com.amazonaws.services.lambda.runtime.Context;
 import com.amazonaws.services.lambda.runtime.RequestHandler;
 
+import org.reactivestreams.Publisher;
+
 import reactor.core.publisher.Flux;
 
 /**
@@ -43,13 +45,13 @@ public class SpringBootRequestHandler<E, O> extends SpringFunctionInitializer
 	public Object handleRequest(E event, Context context) {
 		initialize();
 		Object input = convertEvent(event);
-		Flux<?> output = apply(extract(input));
+		Publisher<?> output = apply(extract(input));
 		return result(input, output);
 	}
 
-	private Object result(Object input, Flux<?> output) {
+	private Object result(Object input, Publisher<?> output) {
 		List<O> result = new ArrayList<>();
-		for (Object value : output.toIterable()) {
+		for (Object value : Flux.from(output).toIterable()) {
 			result.add(convertOutput(value));
 		}
 		if (isSingleValue(input) && result.size() == 1) {

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/SpringBootStreamHandler.java

@@ -27,6 +27,8 @@ import com.amazonaws.services.lambda.runtime.Context;
 import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import org.reactivestreams.Publisher;
+
 import org.springframework.beans.factory.annotation.Autowired;
 
 import reactor.core.publisher.Flux;
@@ -53,13 +55,13 @@ public class SpringBootStreamHandler extends SpringFunctionInitializer
 			throws IOException {
 		initialize();
 		Object value = convertStream(input);
-		Flux<?> flux = apply(extract(value));
+		Publisher<?> flux = apply(extract(value));
 		mapper.writeValue(output, result(value, flux));
 	}
 
-	private Object result(Object input, Flux<?> flux) {
+	private Object result(Object input, Publisher<?> flux) {
 		List<Object> result = new ArrayList<>();
-		for (Object value : flux.toIterable()) {
+		for (Object value : Flux.from(flux).toIterable()) {
 			result.add(value);
 		}
 		if (isSingleValue(input) && result.size()==1) {

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/SpringFunctionInitializer.java

@@ -50,11 +50,11 @@ public class SpringFunctionInitializer implements Closeable {
 
 	private final Class<?> configurationClass;
 
-	private Function<Flux<?>, Publisher<?>> function;
+	private Function<Publisher<?>, Publisher<?>> function;
 
-	private Consumer<Flux<?>> consumer;
+	private Consumer<Publisher<?>> consumer;
 
-	private Supplier<Flux<?>> supplier;
+	private Supplier<Publisher<?>> supplier;
 
 	private AtomicBoolean initialized = new AtomicBoolean();
 
@@ -146,7 +146,7 @@ public class SpringFunctionInitializer implements Closeable {
 				: (this.consumer != null ? this.consumer : this.supplier);
 	}
 
-	protected Flux<?> apply(Flux<?> input) {
+	protected Publisher<?> apply(Publisher<?> input) {
 		if (this.function != null) {
 			return Flux.from(function.apply(input));
 		}

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/test/java/org/springframework/cloud/function/adapter/aws/SpringFunctionInitializerTests.java

@@ -53,7 +53,7 @@ public class SpringFunctionInitializerTests {
 	public void functionBean() {
 		initializer = new SpringFunctionInitializer(FluxFunctionConfig.class);
 		initializer.initialize();
-		Flux<?> result = initializer.apply(Flux.just(new Foo()));
+		Flux<?> result = Flux.from(initializer.apply(Flux.just(new Foo())));
 		assertThat(result.blockFirst()).isInstanceOf(Bar.class);
 	}
 
@@ -61,7 +61,7 @@ public class SpringFunctionInitializerTests {
 	public void functionCatalog() {
 		initializer = new SpringFunctionInitializer(FunctionConfig.class);
 		initializer.initialize();
-		Flux<?> result = initializer.apply(Flux.just(new Foo()));
+		Flux<?> result = Flux.from(initializer.apply(Flux.just(new Foo())));
 		assertThat(result.blockFirst()).isInstanceOf(Bar.class);
 	}
 
@@ -70,7 +70,7 @@ public class SpringFunctionInitializerTests {
 		initializer = new SpringFunctionInitializer(NamedFunctionConfig.class);
 		System.setProperty("function.name", "other");
 		initializer.initialize();
-		Flux<?> result = initializer.apply(Flux.just(new Foo()));
+		Flux<?> result = Flux.from(initializer.apply(Flux.just(new Foo())));
 		assertThat(result.blockFirst()).isInstanceOf(Bar.class);
 	}
 
@@ -78,7 +78,7 @@ public class SpringFunctionInitializerTests {
 	public void consumerCatalog() {
 		initializer = new SpringFunctionInitializer(ConsumerConfig.class);
 		initializer.initialize();
-		Flux<?> result = initializer.apply(Flux.just(new Foo()));
+		Flux<?> result = Flux.from(initializer.apply(Flux.just(new Foo())));
 		assertThat(result.toStream().collect(Collectors.toList())).isEmpty();
 	}
 
@@ -86,7 +86,7 @@ public class SpringFunctionInitializerTests {
 	public void supplierCatalog() {
 		initializer = new SpringFunctionInitializer(SupplierConfig.class);
 		initializer.initialize();
-		Flux<?> result = initializer.apply(Flux.empty());
+		Flux<?> result = Flux.from(initializer.apply(Flux.empty()));
 		assertThat(result.blockFirst()).isInstanceOf(Bar.class);
 	}
 

spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/AzureSpringBootRequestHandler.java

@@ -22,6 +22,8 @@ import java.util.List;
 
 import com.microsoft.azure.serverless.functions.ExecutionContext;
 
+import org.reactivestreams.Publisher;
+
 import reactor.core.publisher.Flux;
 
 /**
@@ -44,7 +46,7 @@ public class AzureSpringBootRequestHandler<I, O> extends AzureSpringFunctionInit
 		initialize(context);
 
 		Object convertedEvent = convertEvent(foo);
-		Flux<?> output = apply(extract(convertedEvent));
+		Publisher<?> output = apply(extract(convertedEvent));
 		return result(convertedEvent, output);
 	}
 
@@ -59,9 +61,9 @@ public class AzureSpringBootRequestHandler<I, O> extends AzureSpringFunctionInit
 		return Flux.just(input);
 	}
 
-	private O result(Object input, Flux<?> output) {
+	private O result(Object input, Publisher<?> output) {
 		List<Object> result = new ArrayList<>();
-		for (Object value : output.toIterable()) {
+		for (Object value : Flux.from(output).toIterable()) {
 			result.add(convertOutput(value));
 		}
 		if (isSingleValue(input) && result.size() == 1) {

spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/java/org/springframework/cloud/function/adapter/azure/AzureSpringFunctionInitializer.java

@@ -28,7 +28,8 @@ import java.util.function.Function;
 import java.util.jar.Manifest;
 
 import com.microsoft.azure.serverless.functions.ExecutionContext;
-import reactor.core.publisher.Flux;
+
+import org.reactivestreams.Publisher;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -37,12 +38,14 @@ import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.util.ClassUtils;
 
+import reactor.core.publisher.Flux;
+
 /**
  * @author Soby Chacko
  */
 public class AzureSpringFunctionInitializer implements Closeable {
 
-	private Function<Flux<?>, Flux<?>> function;
+	private Function<Publisher<?>, Publisher<?>> function;
 
 	private AtomicBoolean initialized = new AtomicBoolean();
 
@@ -110,7 +113,7 @@ public class AzureSpringFunctionInitializer implements Closeable {
 		}
 	}
 
-	protected Flux<?> apply(Flux<?> input) {
+	protected Publisher<?> apply(Publisher<?> input) {
 		if (this.function != null) {
 			return function.apply(input);
 		}

spring-cloud-function-adapters/spring-cloud-function-adapter-openwhisk/src/main/java/org/springframework/cloud/function/adapter/openwhisk/OpenWhiskActionHandler.java

@@ -16,17 +16,24 @@
 
 package org.springframework.cloud.function.adapter.openwhisk;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
-import reactor.core.publisher.Flux;
 
-import java.util.*;
+import reactor.core.publisher.Flux;
 
 /**
  * @author Mark Fisher
@@ -55,15 +62,15 @@ public class OpenWhiskActionHandler extends OpenWhiskFunctionInitializer {
 		Object input = convertEvent(request.getValue());
 		Object result = NO_INPUT_PROVIDED;
 		if(input !=null ) {
-			Flux<?> output = apply(extract(input));
+			Publisher<?> output = apply(extract(input));
 			result = result(input, output);
 		}
 		return serializeBody(result);
 	}
 
-	private Object result(Object input, Flux<?> output) {
+	private Object result(Object input, Publisher<?> output) {
 		List<Object> result = new ArrayList<>();
-		for (Object value : output.toIterable()) {
+		for (Object value : Flux.from(output).toIterable()) {
 			result.add(value);
 		}
 		if (isSingleValue(input) && result.size() == 1) {

spring-cloud-function-adapters/spring-cloud-function-adapter-openwhisk/src/main/java/org/springframework/cloud/function/adapter/openwhisk/OpenWhiskFunctionInitializer.java

@@ -23,6 +23,7 @@ import java.util.function.Supplier;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.function.context.FunctionCatalog;
@@ -39,11 +40,11 @@ public class OpenWhiskFunctionInitializer {
 
 	private static Log logger = LogFactory.getLog(OpenWhiskFunctionInitializer.class);
 
-	private Function<Flux<?>, Flux<?>> function;
+	private Function<Publisher<?>, Publisher<?>> function;
 
-	private Consumer<Flux<?>> consumer;
+	private Consumer<Publisher<?>> consumer;
 
-	private Supplier<Flux<?>> supplier;
+	private Supplier<Publisher<?>> supplier;
 
 	private AtomicBoolean initialized = new AtomicBoolean();
 
@@ -86,7 +87,7 @@ public class OpenWhiskFunctionInitializer {
 				: (this.consumer != null ? this.consumer : this.supplier);
 	}
 
-	protected Flux<?> apply(Flux<?> input) {
+	protected Publisher<?> apply(Publisher<?> input) {
 		if (this.function != null) {
 			return function.apply(input);
 		}

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/FluxWrapperDetector.java

@@ -22,9 +22,6 @@ import org.reactivestreams.Publisher;
 
 import org.springframework.cloud.function.context.WrapperDetector;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
 /**
  * @author Dave Syer
  *
@@ -33,8 +30,11 @@ public class FluxWrapperDetector implements WrapperDetector {
 
 	@Override
 	public boolean isWrapper(Type type) {
-		return Publisher.class.equals(type) || Flux.class.equals(type)
+		if (type instanceof Class<?>) {
-				|| Mono.class.equals(type);
+			Class<?> cls = (Class<?>) type;
+			return Publisher.class.isAssignableFrom(cls);
+		}
+		return false;
 	}
 
 }

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

@@ -80,7 +80,7 @@ public abstract class AbstractStreamListeningInvoker
 	}
 
 	protected Mono<Void> consumer(String name, Flux<Message<?>> flux) {
-		Consumer<Object> consumer = functionCatalog.lookup(Consumer.class, name);
+		Consumer<Publisher<?>> consumer = functionCatalog.lookup(Consumer.class, name);
 		flux = flux.publish().refCount(2);
 		// The consumer will subscribe to the input flux, so we need to listen separately
 		consumer.accept(flux.map(message -> convertInput(consumer).apply(message))
@@ -89,7 +89,7 @@ public abstract class AbstractStreamListeningInvoker
 	}
 
 	protected Flux<Message<?>> function(String name, Flux<Message<?>> flux) {
-		Function<Object, Flux<?>> function = functionCatalog.lookup(Function.class, name);
+		Function<Publisher<?>, Publisher<?>> function = functionCatalog.lookup(Function.class, name);
 		return flux.publish(values -> {
 			Publisher<?> result = function
 					.apply(values.map(message -> convertInput(function).apply(message)));

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

@@ -22,6 +22,8 @@ import java.util.Map;
 import java.util.Set;
 import java.util.function.Supplier;
 
+import org.reactivestreams.Publisher;
+
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.message.MessageUtils;
 import org.springframework.cloud.stream.messaging.Source;
@@ -93,12 +95,12 @@ public class SupplierInvokingMessageProducer<T> extends MessageProducerSupport {
 		if (!disposables.containsKey(name)) {
 			synchronized (disposables) {
 				if (!disposables.containsKey(name)) {
-					Supplier<Flux<?>> supplier = functionCatalog.lookup(Supplier.class,
+					Supplier<Publisher<?>> supplier = functionCatalog.lookup(Supplier.class,
 							name);
 					if (supplier != null) {
 						suppliers.add(name);
 						disposables.put(name,
-								supplier.get().subscribeOn(Schedulers.elastic())
+								Flux.from(supplier.get()).subscribeOn(Schedulers.elastic())
 										.subscribe(m -> send(name, m)));
 					}
 				}
@@ -107,7 +109,7 @@ public class SupplierInvokingMessageProducer<T> extends MessageProducerSupport {
 	}
 
 	private void send(String name, Object payload) {
-		Supplier<Flux<?>> supplier = functionCatalog.lookup(Supplier.class, name);
+		Supplier<Publisher<?>> supplier = functionCatalog.lookup(Supplier.class, name);
 		Message<?> message = MessageUtils.unpack(supplier, payload);
 		message = MessageBuilder.fromMessage(message)
 				.setHeaderIfAbsent(StreamConfigurationProperties.ROUTE_KEY, name).build();

spring-cloud-function-task/src/main/java/org/springframework/cloud/function/task/TaskConfiguration.java

@@ -20,6 +20,8 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+import org.reactivestreams.Publisher;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -29,7 +31,6 @@ import org.springframework.cloud.task.configuration.EnableTask;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /**
@@ -46,11 +47,11 @@ public class TaskConfiguration {
 
 	@Bean
 	public CommandLineRunner commandLineRunner(FunctionCatalog registry) {
-		final Supplier<Flux<Object>> supplier = registry.lookup(Supplier.class,
+		final Supplier<Publisher<Object>> supplier = registry.lookup(Supplier.class,
 				properties.getSupplier());
-		final Function<Flux<Object>, Flux<Object>> function = registry
+		final Function<Publisher<Object>, Publisher<Object>> function = registry
 				.lookup(Function.class, properties.getFunction());
-		final Consumer<Flux<Object>> consumer = consumer(registry);
+		final Consumer<Publisher<Object>> consumer = consumer(registry);
 		CommandLineRunner runner = new CommandLineRunner() {
 
 			@Override
@@ -61,14 +62,14 @@ public class TaskConfiguration {
 		return runner;
 	}
 
-	private Consumer<Flux<Object>> consumer(FunctionCatalog registry) {
+	private Consumer<Publisher<Object>> consumer(FunctionCatalog registry) {
-		Consumer<Flux<Object>> consumer = registry.lookup(Consumer.class,
+		Consumer<Publisher<Object>> consumer = registry.lookup(Consumer.class,
 				properties.getConsumer());
 		if (consumer != null) {
 			return consumer;
 		}
-		Function<Flux<Object>, Mono<Void>> function = registry.lookup(Function.class,
+		Function<Publisher<Object>, Publisher<Void>> function = registry.lookup(Function.class,
 				properties.getConsumer());
-		return flux -> function.apply(flux).subscribe();
+		return flux -> Mono.from(function.apply(flux)).subscribe();
 	}
 }

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionController.java

@@ -73,10 +73,10 @@ public class FunctionController {
 			@RequestBody FluxRequest<?> body) {
 
 		@SuppressWarnings("unchecked")
-		Function<Flux<?>, Flux<?>> function = (Function<Flux<?>, Flux<?>>) request
+		Function<Publisher<?>, Publisher<?>> function = (Function<Publisher<?>, Publisher<?>>) request
 				.getAttribute(WebRequestConstants.FUNCTION, WebRequest.SCOPE_REQUEST);
 		@SuppressWarnings("unchecked")
-		Consumer<Flux<?>> consumer = (Consumer<Flux<?>>) request
+		Consumer<Publisher<?>> consumer = (Consumer<Publisher<?>>) request
 				.getAttribute(WebRequestConstants.CONSUMER, WebRequest.SCOPE_REQUEST);
 		Boolean single = (Boolean) request.getAttribute(WebRequestConstants.INPUT_SINGLE,
 				WebRequest.SCOPE_REQUEST);
@@ -116,7 +116,7 @@ public class FunctionController {
 	}
 
 	private Publisher<?> response(WebRequest request, Object handler, Boolean single,
-			Flux<?> result) {
+			Publisher<?> result) {
 
 		if (single != null && single && isOutputSingle(handler)) {
 			request.setAttribute(WebRequestConstants.OUTPUT_SINGLE, true,
@@ -146,10 +146,10 @@ public class FunctionController {
 	@ResponseBody
 	public Publisher<?> get(WebRequest request) {
 		@SuppressWarnings("unchecked")
-		Function<Flux<?>, Flux<?>> function = (Function<Flux<?>, Flux<?>>) request
+		Function<Publisher<?>, Publisher<?>> function = (Function<Publisher<?>, Publisher<?>>) request
 				.getAttribute(WebRequestConstants.FUNCTION, WebRequest.SCOPE_REQUEST);
 		@SuppressWarnings("unchecked")
-		Supplier<Flux<?>> supplier = (Supplier<Flux<?>>) request
+		Supplier<Publisher<?>> supplier = (Supplier<Publisher<?>>) request
 				.getAttribute(WebRequestConstants.SUPPLIER, WebRequest.SCOPE_REQUEST);
 		String argument = (String) request.getAttribute(WebRequestConstants.ARGUMENT,
 				WebRequest.SCOPE_REQUEST);
@@ -160,15 +160,15 @@ public class FunctionController {
 		return response(request, supplier, true, supplier(supplier));
 	}
 
-	private Flux<?> supplier(Supplier<Flux<?>> supplier) {
+	private Publisher<?> supplier(Supplier<Publisher<?>> supplier) {
-		Flux<?> result = supplier.get();
+		Publisher<?> result = supplier.get();
 		if (logger.isDebugEnabled()) {
 			logger.debug("Handled GET with supplier");
 		}
-		return debug ? result.log() : result;
+		return debug ? Flux.from(result).log() : result;
 	}
 
-	private Mono<?> value(Function<Flux<?>, Flux<?>> function, String value) {
+	private Mono<?> value(Function<Publisher<?>, Publisher<?>> function, String value) {
 		Object input = converter.convert(function, value);
 		Mono<?> result = Mono.from(function.apply(Flux.just(input)));
 		if (logger.isDebugEnabled()) {

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionHandlerMapping.java

@@ -22,6 +22,8 @@ import java.util.function.Supplier;
 
 import javax.servlet.http.HttpServletRequest;
 
+import org.reactivestreams.Publisher;
+
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -118,7 +120,7 @@ public class FunctionHandlerMapping extends RequestMappingHandlerMapping
 			return null;
 		}
 		path = path.startsWith("/") ? path.substring(1) : path;
-		Consumer<Object> consumer = functions.lookup(Consumer.class, path);
+		Consumer<Publisher<?>> consumer = functions.lookup(Consumer.class, path);
 		if (consumer != null) {
 			request.setAttribute(WebRequestConstants.CONSUMER, consumer);
 			return consumer;
@@ -136,7 +138,7 @@ public class FunctionHandlerMapping extends RequestMappingHandlerMapping
 			return null;
 		}
 		path = path.startsWith("/") ? path.substring(1) : path;
-		Supplier<Object> supplier = functions.lookup(Supplier.class, path);
+		Supplier<Publisher<?>> supplier = functions.lookup(Supplier.class, path);
 		if (supplier != null) {
 			request.setAttribute(WebRequestConstants.SUPPLIER, supplier);
 			return supplier;