From 8e5d631db956b4df738af6e8b46e908605419fc6 Mon Sep 17 00:00:00 2001
From: markfisher <mfisher@pivotal.io>
Date: Sat, 1 Oct 2016 16:52:59 -0400
Subject: [PATCH] support function composition for web and stream

---
 README.md                                           | 10 ++++++++++
 .../cloud/function/stream/StreamConfiguration.java  | 13 +++++++++++--
 .../cloud/function/web/RestConfiguration.java       | 10 +++++++---
 3 files changed, 28 insertions(+), 5 deletions(-)

diff --git a/README.md b/README.md
index 6f7d7f43d..7c628ef55 100644
--- a/README.md
+++ b/README.md
@@ -16,4 +16,14 @@ Run a REST Microservice using that Function:
 ./web.sh -p /words -f uppercase
 ```
 
+To compose Functions:
+
+(assuming the `uppercase` function was already registered as above)
+
+```
+./register.sh -n pluralize -f "f->f.map(s->s+\"S\")"
+
+./web.sh -p /words -f uppercase,pluralize
+```
+
 (more docs soon)
diff --git a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java
index f0da678a7..56a1e30dd 100644
--- a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java
+++ b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.function.stream;
 
+import java.util.function.Function;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.function.invoker.AbstractFunctionInvoker;
@@ -24,6 +26,9 @@ import org.springframework.cloud.function.registry.FunctionRegistry;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.context.annotation.Bean;
+import org.springframework.util.StringUtils;
+
+import reactor.core.publisher.Flux;
 
 /**
  * @author Mark Fisher
@@ -41,7 +46,11 @@ public class StreamConfiguration {
 	}
 
 	@Bean
-	public AbstractFunctionInvoker<?,?> invoker() {
-		return new StreamListeningFunctionInvoker(registry().lookup(properties.getName()));
+	public AbstractFunctionInvoker<?,?> invoker(FunctionRegistry registry) {
+		String name = properties.getName();
+		Function<Flux<Object>, Flux<Object>> function = (name.indexOf(',') == -1)
+				? registry.lookup(name)
+				: registry.compose(StringUtils.commaDelimitedListToStringArray(name));
+		return new StreamListeningFunctionInvoker(function);
 	}
 }
diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestConfiguration.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestConfiguration.java
index 3d2217b51..e4ec41693 100644
--- a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestConfiguration.java
+++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestConfiguration.java
@@ -19,24 +19,25 @@ package org.springframework.cloud.function.web;
 import static org.springframework.http.MediaType.TEXT_PLAIN;
 import static org.springframework.web.reactive.function.BodyExtractors.toFlux;
 import static org.springframework.web.reactive.function.BodyInserters.fromPublisher;
-import static org.springframework.web.reactive.function.RequestPredicates.contentType;
 import static org.springframework.web.reactive.function.RequestPredicates.POST;
+import static org.springframework.web.reactive.function.RequestPredicates.contentType;
 
 import java.util.concurrent.Executors;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.function.registry.FileSystemFunctionRegistry;
 import org.springframework.cloud.function.registry.FunctionRegistry;
-import org.springframework.cloud.function.registry.InMemoryFunctionRegistry;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
+import org.springframework.util.StringUtils;
 import org.springframework.web.reactive.function.Request;
 import org.springframework.web.reactive.function.Response;
 import org.springframework.web.reactive.function.RouterFunction;
@@ -65,7 +66,10 @@ public class RestConfiguration {
 
 	@Bean
 	public HttpHandler httpHandler(FunctionRegistry registry) {
-		Function<Flux<String>, Flux<String>> function = registry.lookup(functionProperties.getName());
+		String name = functionProperties.getName();
+		Function<Flux<String>, Flux<String>> function = (name.indexOf(',') == -1)
+				? registry.lookup(name)
+				: registry.compose(StringUtils.commaDelimitedListToStringArray(name));
 		FunctionInvokingHandler handler = new FunctionInvokingHandler(function);
 		RouterFunction<Publisher<String>> route = RouterFunctions.route(
 				POST(webProperties.getPath()).and(contentType(TEXT_PLAIN)), handler::handleText);