Add POJO sample and switch to using that for testing

spring-cloud-function-context/pom.xml

@@ -26,6 +26,10 @@
 			<artifactId>spring-cloud-function-core</artifactId>
 			<version>${spring-cloud-function.version}</version>
 		</dependency>
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-configuration-processor</artifactId>

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

@@ -15,13 +15,26 @@
  */
 package org.springframework.cloud.function.context;
 
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
 import java.util.Collections;
 import java.util.Map;
 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;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.beans.factory.support.SimpleBeanDefinitionRegistry;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -29,6 +42,12 @@ import org.springframework.cloud.function.registry.DefaultFunctionRegistryAutoCo
 import org.springframework.cloud.function.registry.FunctionCatalog;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.ResolvableType;
+import org.springframework.core.type.StandardMethodMetadata;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
+
+import reactor.core.publisher.Flux;
 
 @Configuration
 @ConditionalOnClass(ApplicationContextFunctionCatalog.class)
@@ -48,4 +67,195 @@ public class ContextFunctionCatalogAutoConfiguration {
 		return new ApplicationContextFunctionCatalog(functions, consumers, suppliers);
 	}
 
+	@Component
+	public static class ContextFunctionPostProcessor
+			implements BeanFactoryPostProcessor, BeanDefinitionRegistryPostProcessor {
+
+		private BeanDefinitionRegistry registry;
+
+		private BeanDefinitionRegistry targets = new SimpleBeanDefinitionRegistry();
+
+		@Override
+		public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry)
+				throws BeansException {
+			this.registry = registry;
+		}
+
+		@Override
+		public void postProcessBeanFactory(ConfigurableListableBeanFactory factory)
+				throws BeansException {
+			for (String name : factory.getBeanDefinitionNames()) {
+				if (isGenericFunction(factory, name)) {
+					wrapFunction(factory, name);
+				}
+				else if (isGenericSupplier(factory, name)) {
+					wrapSupplier(factory, name);
+				}
+			}
+		}
+
+		private boolean isGenericFunction(ConfigurableListableBeanFactory factory,
+				String name) {
+			return factory.isTypeMatch(name,
+					ResolvableType.forClassWithGenerics(Function.class, Flux.class,
+							Flux.class))
+					&& !factory.isTypeMatch(name,
+							ResolvableType.forClassWithGenerics(Function.class,
+									ResolvableType.forClassWithGenerics(Flux.class,
+											String.class),
+									ResolvableType.forClassWithGenerics(Flux.class,
+											String.class)));
+		}
+
+		private boolean isGenericSupplier(ConfigurableListableBeanFactory factory,
+				String name) {
+			return factory.isTypeMatch(name,
+					ResolvableType.forClassWithGenerics(Supplier.class, Flux.class))
+					&& !factory.isTypeMatch(name,
+							ResolvableType.forClassWithGenerics(Supplier.class,
+									ResolvableType.forClassWithGenerics(Flux.class,
+											String.class)));
+		}
+
+		private void wrapFunction(ConfigurableListableBeanFactory factory, String name) {
+			BeanDefinition definition = registry.getBeanDefinition(name);
+			BeanDefinitionBuilder builder = BeanDefinitionBuilder
+					.genericBeanDefinition(ProxyFunction.class);
+			builder.addPropertyValue("delegate", definition);
+			builder.addPropertyValue("name", name);
+			targets.registerBeanDefinition(name, definition);
+			builder.addPropertyValue("registry", targets);
+			registry.registerBeanDefinition(name, builder.getRawBeanDefinition());
+		}
+
+		private void wrapSupplier(ConfigurableListableBeanFactory factory, String name) {
+			BeanDefinition definition = registry.getBeanDefinition(name);
+			BeanDefinitionBuilder builder = BeanDefinitionBuilder
+					.genericBeanDefinition(ProxySupplier.class);
+			builder.addPropertyValue("delegate", definition);
+			builder.addPropertyValue("name", name);
+			targets.registerBeanDefinition(name, definition);
+			builder.addPropertyValue("registry", targets);
+			registry.registerBeanDefinition(name, builder.getRawBeanDefinition());
+		}
+	}
+}
+
+class ProxyFunction implements Function<Flux<String>, Flux<String>> {
+
+	private ObjectMapper mapper;
+
+	private Function<Flux<Object>, Flux<Object>> delegate;
+
+	private String name;
+
+	private Class<?> type;
+
+	private BeanDefinitionRegistry registry;
+
+	@Autowired
+	public ProxyFunction(ObjectMapper mapper) {
+		this.mapper = mapper;
+	}
+
+	public void setDelegate(Function<Flux<Object>, Flux<Object>> delegate) {
+		this.delegate = delegate;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public void setRegistry(BeanDefinitionRegistry registry) {
+		this.registry = registry;
+	}
+
+	private Class<?> findType(RootBeanDefinition definition) {
+		StandardMethodMetadata source = (StandardMethodMetadata) definition.getSource();
+		ParameterizedType type = (ParameterizedType) (source.getIntrospectedMethod()
+				.getGenericReturnType());
+		type = (ParameterizedType) type.getActualTypeArguments()[0];
+		Type param = type.getActualTypeArguments()[0];
+		return ClassUtils.resolveClassName(param.getTypeName(),
+				registry.getClass().getClassLoader());
+	}
+
+	@Override
+	public Flux<String> apply(Flux<String> input) {
+		if (type == null) {
+			type = findType((RootBeanDefinition) registry.getBeanDefinition(name));
+		}
+		return delegate.apply(input.map(value -> {
+			try {
+				return mapper.readValue(value, type);
+			}
+			catch (Exception e) {
+				throw new IllegalStateException("Cannot convert from JSON: " + input);
+			}
+		})).map(value -> {
+			try {
+				return mapper.writeValueAsString(value);
+			}
+			catch (Exception e) {
+				throw new IllegalStateException("Cannot convert to JSON: " + input);
+			}
+		});
+	}
+
+}
+
+class ProxySupplier implements Supplier<Flux<String>> {
+
+	private ObjectMapper mapper;
+
+	private Supplier<Flux<Object>> delegate;
+
+	private String name;
+
+	private Class<?> type;
+
+	private BeanDefinitionRegistry registry;
+
+	@Autowired
+	public ProxySupplier(ObjectMapper mapper) {
+		this.mapper = mapper;
+	}
+
+	public void setDelegate(Supplier<Flux<Object>> delegate) {
+		this.delegate = delegate;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public void setRegistry(BeanDefinitionRegistry registry) {
+		this.registry = registry;
+	}
+
+	private Class<?> findType(RootBeanDefinition definition) {
+		StandardMethodMetadata source = (StandardMethodMetadata) definition.getSource();
+		ParameterizedType type = (ParameterizedType) (source.getIntrospectedMethod()
+				.getGenericReturnType());
+		type = (ParameterizedType) type.getActualTypeArguments()[0];
+		Type param = type.getActualTypeArguments()[0];
+		return ClassUtils.resolveClassName(param.getTypeName(),
+				registry.getClass().getClassLoader());
+	}
+
+	@Override
+	public Flux<String> get() {
+		if (type == null) {
+			type = findType((RootBeanDefinition) registry.getBeanDefinition(name));
+		}
+		return delegate.get().map(value -> {
+			try {
+				return mapper.writeValueAsString(value);
+			}
+			catch (Exception e) {
+				throw new IllegalStateException("Cannot convert to JSON: " + value);
+			}
+		});
+	}
+
 }

spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionAdminController.java

@@ -86,7 +86,7 @@ public class FunctionAdminController implements CommandLineRunner {
 
 	@Override
 	public void run(String... args) throws Exception {
-		deploy("sample", "maven://com.example:function-sample:1.0.0.BUILD-SNAPSHOT");
+		deploy("sample", "maven://com.example:function-sample-pojo:1.0.0.BUILD-SNAPSHOT");
 	}
 
 	private String deploy(String name, String path, String... args) throws Exception {

spring-cloud-function-deployer/src/test/java/org/springframework/cloud/function/deployer/FunctionExtractingAppDeployerIntegrationTests.java

@@ -37,6 +37,7 @@ public class FunctionExtractingAppDeployerIntegrationTests {
 	@BeforeClass
 	public static void open() {
 		port = SocketUtils.findAvailableTcpPort();
+		System.setProperty("debug", "true");
 		context = new ApplicationRunner().start("--server.port=" + port);
 	}
 
@@ -48,10 +49,17 @@ public class FunctionExtractingAppDeployerIntegrationTests {
 	}
 
 	@Test
-	public void test() {
+	public void words() {
 		assertThat(new TestRestTemplate()
 				.getForObject("http://localhost:" + port + "/words", String.class))
-						.isEqualTo("foobar");
+						.isEqualTo("{\"value\":\"foo\"}{\"value\":\"bar\"}");
+	}
+
+	@Test
+	public void uppercase() {
+		assertThat(new TestRestTemplate().postForObject(
+				"http://localhost:" + port + "/uppercase", "{\"value\":\"foo\"}",
+				String.class)).isEqualTo("{\"value\":\"FOO\"}");
 	}
 
 }

spring-cloud-function-deployer/thin-slim.properties

@@ -1,6 +1,5 @@
 exclusions.spring-web-reactive: org.springframework:spring-web-reactive
 exclusions.spring-cloud-function-web: org.springframework.cloud:spring-cloud-function-web
-exclusions.spring-web: org.springframework:spring-web
 exclusions.reator-netty: io.projectreactor.ipc:reactor-netty
 exclusions.spring-cloud-stream: org.springframework.cloud:spring-cloud-stream
 exclusions.spring-cloud-stream-reactive: org.springframework.cloud:spring-cloud-stream-reactive
@@ -9,6 +8,8 @@ exclusions.spring-cloud-stream-binder-kafka: org.springframework.cloud:spring-cl
 exclusions.spring-boot-starter-web: org.springframework.boot:spring-boot-starter-web
 exclusions.spring-boot-starter-stream: org.springframework.boot:spring-boot-starter-stream
 exclusions.spring-boot-starter-actuator: org.springframework.boot:spring-boot-starter-actuator
+dependencies.spring-web: org.springframework:spring-web
+dependencies.jackson-databind: com.fasterxml.jackson.core:jackson-databind
 dependencies.spring-boot-starter: org.springframework.boot:spring-boot-starter
 dependencies.spring-cloud-function-context: org.springframework.cloud:spring-cloud-function-context:1.0.0.BUILD-SNAPSHOT
 

spring-cloud-function-samples/pom.xml

@@ -14,6 +14,7 @@
 
 	<modules>
 		<module>spring-cloud-function-sample</module>
+		<module>spring-cloud-function-sample-pojo</module>
 	</modules>
 
 </project>

spring-cloud-function-samples/spring-cloud-function-sample-pojo/pom.xml

@@ -0,0 +1,77 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.example</groupId>
+	<artifactId>function-sample-pojo</artifactId>
+	<version>1.0.0.BUILD-SNAPSHOT</version>
+	<packaging>jar</packaging>
+	<name>spring-cloud-function-sample</name>
+	<description>Spring Cloud Function Web Support</description>
+
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>2.0.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<properties>
+		<java.version>1.8</java.version>
+		<spring-cloud-function.version>1.0.0.BUILD-SNAPSHOT</spring-cloud-function.version>
+		<wrapper.version>0.0.1.BUILD-SNAPSHOT</wrapper.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-function-web</artifactId>
+			<version>${spring-cloud-function.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-function-context</artifactId>
+			<version>${spring-cloud-function.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<optional>true</optional>
+		</dependency>
+	</dependencies>
+
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-dependencies</artifactId>
+				<version>2.0.0.BUILD-SNAPSHOT</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-jar-plugin</artifactId>
+				<version>3.0.0</version>
+			</plugin>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+				<version>1.5.0.BUILD-SNAPSHOT</version>
+				<dependencies>
+					<dependency>
+						<groupId>org.springframework.boot.experimental</groupId>
+						<artifactId>spring-boot-thin-launcher</artifactId>
+						<version>${wrapper.version}</version>
+					</dependency>
+				</dependencies>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>

spring-cloud-function-samples/spring-cloud-function-sample-pojo/src/main/java/com/example/SampleApplication.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2013-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 com.example;
+
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+
+import reactor.core.publisher.Flux;
+
+@SpringBootApplication
+public class SampleApplication {
+
+	@Bean
+	public Function<Flux<Foo>, Flux<Bar>> uppercase() {
+		return flux -> flux.map(value -> new Bar(value.uppercase()));
+	}
+
+	@Bean
+	public Supplier<Flux<Bar>> words() {
+		return () -> Flux.fromArray(new Bar[] { new Bar("foo"), new Bar("bar") });
+	}
+
+	@Bean
+	public Function<Flux<Foo>, Flux<Bar>> lowercase() {
+		return flux -> flux.map(value -> new Bar(value.lowercase()));
+	}
+
+	public static void main(String[] args) throws Exception {
+		SpringApplication.run(SampleApplication.class, args);
+	}
+
+}
+
+class Foo {
+
+	private String value;
+
+	Foo() {
+	}
+
+	public String lowercase() {
+		return value.toLowerCase();
+	}
+
+	public Foo(String value) {
+		this.value = value;
+	}
+
+	public String uppercase() {
+		return value.toUpperCase();
+	}
+
+	public String getValue() {
+		return value;
+	}
+
+	public void setValue(String value) {
+		this.value = value;
+	}
+}
+
+class Bar {
+
+	private String value;
+
+	Bar() {
+	}
+
+	public Bar(String value) {
+		this.value = value;
+	}
+
+	public String getValue() {
+		return value;
+	}
+
+	public void setValue(String value) {
+		this.value = value;
+	}
+
+}