Support HTTP GET of single value value Function

E.g. a Function<Long, Foo> can be used to fetch a single entity of
type Foo with a long ID., via /{function}/{id}

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

@@ -27,6 +27,8 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.BeanDefinition;
@@ -47,8 +49,6 @@ import org.springframework.core.type.StandardMethodMetadata;
 import org.springframework.stereotype.Component;
 import org.springframework.util.ClassUtils;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-
 import reactor.core.publisher.Flux;
 
 /**
@@ -70,9 +70,9 @@ public class ContextFunctionCatalogAutoConfiguration {
 	private Map<String, Consumer<?>> consumers = Collections.emptyMap();
 
 	@Bean
-	public FunctionCatalog functionCatalog(ContextFunctionPostProcessor processor, ObjectMapper mapper) {
+	public FunctionCatalog functionCatalog(ContextFunctionPostProcessor processor,
-		return new InMemoryFunctionCatalog(
+			ObjectMapper mapper) {
-				processor.wrapSuppliers(mapper, suppliers),
+		return new InMemoryFunctionCatalog(processor.wrapSuppliers(mapper, suppliers),
 				processor.wrapFunctions(mapper, functions),
 				processor.wrapConsumers(mapper, consumers));
 	}
@@ -99,7 +99,7 @@ public class ContextFunctionCatalogAutoConfiguration {
 				if (this.suppliers.contains(key)) {
 					@SuppressWarnings("unchecked")
 					Supplier<Flux<?>> supplier = (Supplier<Flux<?>>) suppliers.get(key);
-					result.put(key, wrapSupplier(supplier, mapper));
+					result.put(key, wrapSupplier(supplier, mapper, key));
 				}
 				else {
 					result.put(key, suppliers.get(key));
@@ -119,7 +119,9 @@ public class ContextFunctionCatalogAutoConfiguration {
 					result.put(key, wrapFunction(function, mapper, key));
 				}
 				else if (!isFluxFunction(key, functions.get(key))) {
-					result.put(key, new FluxFunction(functions.get(key)));
+					@SuppressWarnings({ "unchecked", "rawtypes" })
+					FluxFunction value = new FluxFunction(functions.get(key));
+					result.put(key, value);
 				}
 				else {
 					result.put(key, functions.get(key));
@@ -166,12 +168,16 @@ public class ContextFunctionCatalogAutoConfiguration {
 				Object source = beanDefinition.getSource();
 				if (source instanceof StandardMethodMetadata) {
 					StandardMethodMetadata metadata = (StandardMethodMetadata) source;
-					Type returnType = metadata.getIntrospectedMethod().getGenericReturnType();
+					Type returnType = metadata.getIntrospectedMethod()
+							.getGenericReturnType();
 					if (returnType instanceof ParameterizedType) {
-						Type[] types = ((ParameterizedType) returnType).getActualTypeArguments();
+						Type[] types = ((ParameterizedType) returnType)
+								.getActualTypeArguments();
 						if (types != null && types.length == 2) {
-							return (types[0].getTypeName().startsWith(Flux.class.getName())
+							return (types[0].getTypeName()
-									&& types[1].getTypeName().startsWith(Flux.class.getName()));
+									.startsWith(Flux.class.getName())
+									&& types[1].getTypeName()
+											.startsWith(Flux.class.getName()));
 						}
 					}
 				}
@@ -213,9 +219,10 @@ public class ContextFunctionCatalogAutoConfiguration {
 		}
 
 		private ProxySupplier wrapSupplier(Supplier<Flux<?>> supplier,
-				ObjectMapper mapper) {
+				ObjectMapper mapper, String name) {
 			ProxySupplier wrapped = new ProxySupplier(mapper);
 			wrapped.setDelegate(supplier);
+			wrapped.setOutputType(findType(name));
 			return wrapped;
 		}
 
@@ -223,7 +230,8 @@ public class ContextFunctionCatalogAutoConfiguration {
 				ObjectMapper mapper, String name) {
 			ProxyFunction wrapped = new ProxyFunction(mapper);
 			wrapped.setDelegate(function);
-			wrapped.setType(findType(name));
+			wrapped.setInputType(findType(name));
+			wrapped.setOutputType(findOutputType(name));
 			return wrapped;
 		}
 
@@ -231,16 +239,16 @@ public class ContextFunctionCatalogAutoConfiguration {
 				ObjectMapper mapper, String name) {
 			ProxyConsumer wrapped = new ProxyConsumer(mapper);
 			wrapped.setDelegate(consumer);
-			wrapped.setType(findType(name));
+			wrapped.setInputType(findType(name));
 			return wrapped;
 		}
 
-		private Class<?> findType(RootBeanDefinition definition) {
+		private Class<?> findType(RootBeanDefinition definition, int index) {
 			StandardMethodMetadata source = (StandardMethodMetadata) definition
 					.getSource();
 			ParameterizedType type = (ParameterizedType) (source.getIntrospectedMethod()
 					.getGenericReturnType());
-			type = (ParameterizedType) type.getActualTypeArguments()[0];
+			type = (ParameterizedType) type.getActualTypeArguments()[index];
 			Type param = type.getActualTypeArguments()[0];
 			if (param instanceof ParameterizedType) {
 				ParameterizedType concrete = (ParameterizedType) param;
@@ -251,7 +259,11 @@ public class ContextFunctionCatalogAutoConfiguration {
 		}
 
 		private Class<?> findType(String name) {
-			return findType((RootBeanDefinition) registry.getBeanDefinition(name));
+			return findType((RootBeanDefinition) registry.getBeanDefinition(name), 0);
+		}
+
+		private Class<?> findOutputType(String name) {
+			return findType((RootBeanDefinition) registry.getBeanDefinition(name), 1);
 		}
 
 	}
@@ -263,7 +275,9 @@ abstract class ProxyWrapper<T> {
 
 	private T delegate;
 
-	private Class<?> type;
+	private Class<?> inputType;
+
+	private Class<?> outputType;
 
 	public ProxyWrapper(ObjectMapper mapper) {
 		this.mapper = mapper;
@@ -277,20 +291,28 @@ abstract class ProxyWrapper<T> {
 		return delegate;
 	}
 
-	public void setType(Class<?> type) {
+	public void setInputType(Class<?> inputType) {
-		this.type = type;
+		this.inputType = inputType;
 	}
 
-	public Class<?> getType() {
+	public void setOutputType(Class<?> outputType) {
-		return this.type;
+		this.outputType = outputType;
+	}
+
+	public Class<?> getInputType() {
+		return this.inputType;
+	}
+
+	public Class<?> getOutputType() {
+		return outputType;
 	}
 
 	public Object fromJson(String value) {
-		if (getType().equals(String.class)) {
+		if (getInputType().equals(String.class)) {
 			return value;
 		}
 		try {
-			return mapper.readValue(value, getType());
+			return mapper.readValue(value, getInputType());
 		}
 		catch (Exception e) {
 			throw new IllegalStateException("Cannot convert from JSON: " + value);
@@ -298,6 +320,9 @@ abstract class ProxyWrapper<T> {
 	}
 
 	public String toJson(Object value) {
+		if (String.class.equals(getOutputType()) && value instanceof String) {
+			return (String) value;
+		}
 		try {
 			return mapper.writeValueAsString(value);
 		}
@@ -308,7 +333,7 @@ abstract class ProxyWrapper<T> {
 
 	@Override
 	public String toString() {
-		return "ProxyWrapper [delegate=" + delegate + ", type=" + type + "]";
+		return "ProxyWrapper [delegate=" + delegate + ", inputType=" + inputType + "]";
 	}
 
 }

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

@@ -79,4 +79,15 @@ public class FunctionController {
 		Flux<String> result = (Flux<String>) supplier.get();
 		return debug ? result.log() : result;
 	}
+
+	@GetMapping(path = "/{name}/{value}")
+	public Flux<String> value(@PathVariable String name, @PathVariable String value) {
+		Function<Flux<?>, Flux<?>> function = functions.lookupFunction(name);
+		if (function != null) {
+			@SuppressWarnings({ "unchecked" })
+			Flux<String> result = (Flux<String>) function.apply(Flux.just(value));
+			return debug ? result.log() : result;
+		}
+		throw new IllegalArgumentException("no such function: " + name);
+	}
 }

spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/RestApplicationTests.java

@@ -150,6 +150,16 @@ public class RestApplicationTests {
 				.isEqualTo("[FOO][BAR]");
 	}
 
+	@Test
+	public void uppercaseGet() {
+		assertThat(rest.getForObject("/uppercase/foo", String.class)).isEqualTo("[FOO]");
+	}
+
+	@Test
+	public void convertGet() {
+		assertThat(rest.getForObject("/wrap/123", String.class)).isEqualTo("..123..");
+	}
+
 	@Test
 	public void uppercaseJsonArray() throws Exception {
 		assertThat(rest.exchange(
@@ -193,6 +203,11 @@ public class RestApplicationTests {
 					.map(value -> "[" + value.trim().toUpperCase() + "]");
 		}
 
+		@Bean
+		public Function<Flux<Integer>, Flux<String>> wrap() {
+			return flux -> flux.log().map(value -> ".." + value + "..");
+		}
+
 		@Bean
 		public Function<Flux<HashMap<String, String>>, Flux<Map<String, String>>> maps() {
 			return flux -> flux.map(value -> {