Convert functions etc. after context starts

The app deployer now has to reach into the function contexts and extract a catalog and call its methods reflectively.
2017-01-04 10:43:18 +00:00
parent 13774abe39
commit 91717ec9a6
9 changed files with 327 additions and 258 deletions
--- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java
+++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java
@@ -18,7 +18,10 @@ package org.springframework.cloud.function.context;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -27,15 +30,11 @@ 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.AbstractBeanDefinition;
-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;
@@ -64,17 +63,23 @@ public class ContextFunctionCatalogAutoConfiguration {
 	private Map<String, Supplier<?>> suppliers = Collections.emptyMap();

 	@Bean
-	public FunctionCatalog functionCatalog() {
-		return new ApplicationContextFunctionCatalog(functions, consumers, suppliers);
+	public FunctionCatalog functionCatalog(ContextFunctionPostProcessor processor,
+			ObjectMapper mapper) {
+		return new ApplicationContextFunctionCatalog(
+				processor.wrapFunctions(mapper, functions),
+				processor.wrapConsumers(mapper, consumers),
+				processor.wrapSuppliers(mapper, suppliers));
 	}

 	@Component
 	public static class ContextFunctionPostProcessor
 			implements BeanFactoryPostProcessor, BeanDefinitionRegistryPostProcessor {

-		private BeanDefinitionRegistry registry;
+		private Set<String> functions = new HashSet<>();
+		private Set<String> consumers = new HashSet<>();
+		private Set<String> suppliers = new HashSet<>();

-		private BeanDefinitionRegistry targets = new SimpleBeanDefinitionRegistry();
+		private BeanDefinitionRegistry registry;

 		@Override
 		public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry)
@@ -82,18 +87,67 @@ public class ContextFunctionCatalogAutoConfiguration {
 			this.registry = registry;
 		}

+		public Map<String, Supplier<?>> wrapSuppliers(ObjectMapper mapper,
+				Map<String, Supplier<?>> suppliers) {
+			Map<String, Supplier<?>> result = new HashMap<>();
+			for (String key : suppliers.keySet()) {
+				if (this.suppliers.contains(key)) {
+					@SuppressWarnings("unchecked")
+					Supplier<Flux<?>> supplier = (Supplier<Flux<?>>) suppliers.get(key);
+					result.put(key, wrapSupplier(supplier, mapper));
+				}
+				else {
+					result.put(key, suppliers.get(key));
+				}
+			}
+			return result;
+		}
+
+		public Map<String, Consumer<?>> wrapConsumers(ObjectMapper mapper,
+				Map<String, Consumer<?>> consumers) {
+			Map<String, Consumer<?>> result = new HashMap<>();
+			for (String key : consumers.keySet()) {
+				if (this.consumers.contains(key)) {
+					@SuppressWarnings("unchecked")
+					Consumer<Flux<?>> consumer = (Consumer<Flux<?>>) consumers.get(key);
+					result.put(key, wrapConsumer(consumer, mapper, key));
+				}
+				else {
+					result.put(key, consumers.get(key));
+				}
+			}
+			return result;
+		}
+
+		public Map<String, Function<?, ?>> wrapFunctions(ObjectMapper mapper,
+				Map<String, Function<?, ?>> functions) {
+			Map<String, Function<?, ?>> result = new HashMap<>();
+			for (String key : functions.keySet()) {
+				if (this.functions.contains(key)) {
+					@SuppressWarnings("unchecked")
+					Function<Flux<?>, Flux<?>> function = (Function<Flux<?>, Flux<?>>) functions
+							.get(key);
+					result.put(key, wrapFunction(function, mapper, key));
+				}
+				else {
+					result.put(key, functions.get(key));
+				}
+			}
+			return result;
+		}
+
 		@Override
 		public void postProcessBeanFactory(ConfigurableListableBeanFactory factory)
 				throws BeansException {
 			for (String name : factory.getBeanDefinitionNames()) {
 				if (isGenericFunction(factory, name)) {
-					wrapFunction(factory, name);
+					this.functions.add(name);
 				}
 				else if (isGenericSupplier(factory, name)) {
-					wrapSupplier(factory, name);
+					this.suppliers.add(name);
 				}
 				else if (isGenericConsumer(factory, name)) {
-					wrapConsumer(factory, name);
+					this.consumers.add(name);
 				}
 			}
 		}
@@ -131,40 +185,44 @@ public class ContextFunctionCatalogAutoConfiguration {
 											String.class)));
 		}

-		private void wrapFunction(ConfigurableListableBeanFactory factory, String name) {
-			AbstractBeanDefinition wrapped = getInputAwareWrappedBean(name,
-					ProxyFunction.class).getBeanDefinition();
-			registry.registerBeanDefinition(name, wrapped);
+		private ProxyFunction wrapFunction(Function<Flux<?>, Flux<?>> function,
+				ObjectMapper mapper, String name) {
+			ProxyFunction wrapped = new ProxyFunction(mapper);
+			wrapped.setDelegate(function);
+			wrapped.setType(findType(name));
+			return wrapped;
 		}

-		private void wrapSupplier(ConfigurableListableBeanFactory factory, String name) {
-			AbstractBeanDefinition wrapped = getWrappedBean(name, ProxySupplier.class)
-					.getBeanDefinition();
-			registry.registerBeanDefinition(name, wrapped);
+		private ProxySupplier wrapSupplier(Supplier<Flux<?>> supplier,
+				ObjectMapper mapper) {
+			ProxySupplier wrapped = new ProxySupplier(mapper);
+			wrapped.setDelegate(supplier);
+			return wrapped;
 		}

-		private void wrapConsumer(ConfigurableListableBeanFactory factory, String name) {
-			AbstractBeanDefinition wrapped = getInputAwareWrappedBean(name,
-					ProxyConsumer.class).getBeanDefinition();
-			registry.registerBeanDefinition(name, wrapped);
+		private ProxyConsumer wrapConsumer(Consumer<Flux<?>> consumer,
+				ObjectMapper mapper, String name) {
+			ProxyConsumer wrapped = new ProxyConsumer(mapper);
+			wrapped.setDelegate(consumer);
+			wrapped.setType(findType(name));
+			return wrapped;
 		}

-		private BeanDefinitionBuilder getInputAwareWrappedBean(String name,
-				Class<?> type) {
-			BeanDefinitionBuilder builder = getWrappedBean(name, type);
-			builder.addPropertyValue("name", name);
-			targets.registerBeanDefinition(name, registry.getBeanDefinition(name));
-			builder.addPropertyValue("registry", targets);
-			return builder;
+		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());
 		}

-		private BeanDefinitionBuilder getWrappedBean(String name, Class<?> type) {
-			BeanDefinition definition = registry.getBeanDefinition(name);
-			BeanDefinitionBuilder builder = BeanDefinitionBuilder
-					.genericBeanDefinition(type);
-			builder.addPropertyValue("delegate", definition);
-			return builder;
+		private Class<?> findType(String name) {
+			return findType((RootBeanDefinition) registry.getBeanDefinition(name));
 		}
+
 	}
 }

@@ -174,12 +232,8 @@ abstract class ProxyWrapper<T> {

 	private T delegate;

-	private String name;
-
 	private Class<?> type;

-	private BeanDefinitionRegistry registry;
-
 	public ProxyWrapper(ObjectMapper mapper) {
 		this.mapper = mapper;
 	}
@@ -192,29 +246,12 @@ abstract class ProxyWrapper<T> {
 		return 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());
+	public void setType(Class<?> type) {
+		this.type = type;
 	}

 	public Class<?> getType() {
-		if (type == null) {
-			type = findType((RootBeanDefinition) registry.getBeanDefinition(name));
-		}
-		return type;
+		return this.type;
 	}

 	public Object fromJson(String value) {
@@ -235,6 +272,11 @@ abstract class ProxyWrapper<T> {
 		}
 	}

+	@Override
+	public String toString() {
+		return "ProxyWrapper [delegate=" + delegate + ", type=" + type + "]";
+	}
+
 }

 class ProxyFunction extends ProxyWrapper<Function<Flux<?>, Flux<?>>>
--- a/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionAdminController.java
+++ b/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionAdminController.java
@@ -43,14 +43,14 @@ import org.springframework.web.bind.annotation.RestController;
@RequestMapping("/admin")
 public class FunctionAdminController implements CommandLineRunner {

-	private final FunctionExtractingAppDeployer deployer;
+	private final FunctionExtractingFunctionCatalog deployer;

 	private Map<String, String> deployed = new LinkedHashMap<>();

 	private Map<String, String> names = new LinkedHashMap<>();

 	@Autowired
-	public FunctionAdminController(FunctionExtractingAppDeployer deployer) {
+	public FunctionAdminController(FunctionExtractingFunctionCatalog deployer) {
 		this.deployer = deployer;
 	}

--- a/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionExtractingAppDeployer.java
+++ b/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionExtractingAppDeployer.java
@@ -1,183 +0,0 @@
-/*
- * Copyright 2016-2017 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 org.springframework.cloud.function.deployer;
-
-import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
-import org.springframework.cloud.deployer.thin.ThinJarAppDeployer;
-import org.springframework.cloud.deployer.thin.ThinJarAppWrapper;
-import org.springframework.cloud.function.registry.FunctionCatalog;
-import org.springframework.util.MethodInvoker;
-import org.springframework.util.ReflectionUtils;
-
-public class FunctionExtractingAppDeployer extends ThinJarAppDeployer
-		implements FunctionCatalog {
-
-	private static final Log logger = LogFactory
-			.getLog(FunctionExtractingAppDeployer.class);
-
-	private final Map<String, Function<?, ?>> functions = new HashMap<>();
-	private final Map<String, Consumer<?>> consumers = new HashMap<>();
-	private final Map<String, Supplier<?>> suppliers = new HashMap<>();
-
-	public FunctionExtractingAppDeployer() {
-		this("thin", "slim");
-	}
-
-	public FunctionExtractingAppDeployer(String name, String... profiles) {
-		super(name, profiles);
-	}
-
-	@SuppressWarnings("unchecked")
-	@Override
-	public <T> Consumer<T> lookupConsumer(String name) {
-		return (Consumer<T>) consumers.get(name);
-	}
-
-	@SuppressWarnings("unchecked")
-	@Override
-	public <T, R> Function<T, R> lookupFunction(String name) {
-		return (Function<T, R>) functions.get(name);
-	}
-
-	@Override
-	public <T, R> Function<T, R> composeFunction(String... functionNames) {
-		Function<T, R> function = this.lookupFunction(functionNames[0]);
-		for (int i = 1; i < functionNames.length; i++) {
-			function = function.andThen(this.lookupFunction(functionNames[i]));
-		}
-		return function;
-	}
-
-	@SuppressWarnings("unchecked")
-	@Override
-	public <T> Supplier<T> lookupSupplier(String name) {
-		return (Supplier<T>) suppliers.get(name);
-	}
-
-	@Override
-	public String deploy(AppDeploymentRequest request) {
-		String id = super.deploy(request);
-		functions.putAll(functions(id));
-		suppliers.putAll(suppliers(id));
-		consumers.putAll(consumers(id));
-		return id;
-	}
-
-	@Override
-	public void undeploy(String id) {
-		super.undeploy(id);
-		for (String name : functions(id).keySet()) {
-			functions.remove(name);
-		}
-		for (String name : suppliers(id).keySet()) {
-			suppliers.remove(name);
-		}
-		for (String name : consumers(id).keySet()) {
-			consumers.remove(name);
-		}
-	}
-
-	private Map<String, Function<?, ?>> functions(String id) {
-		Map<String, Function<?, ?>> map = new HashMap<>();
-		ThinJarAppWrapper wrapper = getWrapper(id);
-		if (wrapper == null) {
-			return map;
-		}
-		try {
-			@SuppressWarnings("unchecked")
-			Map<String, ? extends Function<?, ?>> result = (Map<String, ? extends Function<?, ?>>) getBeans(
-					wrapper, Function.class);
-			map.putAll(result);
-		}
-		catch (Exception e) {
-			throw new IllegalStateException("Cannot extract functions", e);
-		}
-		logger.info("Loaded functions: " + map.keySet());
-		return map;
-	}
-
-	private Map<String, Consumer<?>> consumers(String id) {
-		Map<String, Consumer<?>> map = new HashMap<>();
-		ThinJarAppWrapper wrapper = getWrapper(id);
-		if (wrapper == null) {
-			return map;
-		}
-		try {
-			@SuppressWarnings("unchecked")
-			Map<String, ? extends Consumer<?>> result = (Map<String, ? extends Consumer<?>>) getBeans(
-					wrapper, Consumer.class);
-			map.putAll(result);
-		}
-		catch (Exception e) {
-			throw new IllegalStateException("Cannot extract consumers", e);
-		}
-		logger.info("Loaded consumers: " + map.keySet());
-		return map;
-	}
-
-	private Map<String, Supplier<?>> suppliers(String id) {
-		Map<String, Supplier<?>> map = new HashMap<>();
-		ThinJarAppWrapper wrapper = getWrapper(id);
-		if (wrapper == null) {
-			return map;
-		}
-		try {
-			@SuppressWarnings("unchecked")
-			Map<String, ? extends Supplier<?>> result = (Map<String, ? extends Supplier<?>>) getBeans(
-					wrapper, Supplier.class);
-			map.putAll(result);
-		}
-		catch (Exception e) {
-			throw new IllegalStateException("Cannot extract suppliers", e);
-		}
-		logger.info("Loaded suppliers: " + map.keySet());
-		return map;
-	}
-
-	private <T> Map<String, ? extends T> getBeans(ThinJarAppWrapper wrapper,
-			Class<T> type) throws IllegalAccessException, ClassNotFoundException,
-			NoSuchMethodException, InvocationTargetException {
-		Object app = findContext(wrapper);
-		MethodInvoker invoker = new MethodInvoker();
-		invoker.setTargetObject(app);
-		invoker.setTargetMethod("getBeansOfType");
-		invoker.setArguments(new Object[] { type });
-		invoker.prepare();
-		@SuppressWarnings("unchecked")
-		Map<String, T> result = (Map<String, T>) invoker.invoke();
-		return result;
-	}
-
-	private Object findContext(ThinJarAppWrapper wrapper) {
-		Object app = wrapper.getApp();
-		Field field = ReflectionUtils.findField(app.getClass(), "context");
-		ReflectionUtils.makeAccessible(field);
-		app = ReflectionUtils.getField(field, app);
-		return app;
-	}
-
-}
--- a/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionExtractingAutoConfiguration.java
+++ b/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionExtractingAutoConfiguration.java
@@ -26,13 +26,13 @@ import org.springframework.context.annotation.Configuration;
 *
 */
@Configuration
-@ConditionalOnClass(FunctionExtractingAppDeployer.class)
+@ConditionalOnClass(FunctionExtractingFunctionCatalog.class)
@AutoConfigureBefore(DefaultFunctionRegistryAutoConfiguration.class)
 public class FunctionExtractingAutoConfiguration {

 	@Bean
-	public FunctionExtractingAppDeployer functionCatalog() {
-		return new FunctionExtractingAppDeployer();
+	public FunctionExtractingFunctionCatalog functionCatalog() {
+		return new FunctionExtractingFunctionCatalog();
 	}

 }
--- a/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionExtractingFunctionCatalog.java
+++ b/spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionExtractingFunctionCatalog.java
@@ -0,0 +1,105 @@
+/*
+ * Copyright 2016-2017 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 org.springframework.cloud.function.deployer;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
+import org.springframework.cloud.deployer.thin.ThinJarAppDeployer;
+import org.springframework.cloud.function.registry.FunctionCatalog;
+import org.springframework.util.MethodInvoker;
+
+public class FunctionExtractingFunctionCatalog implements FunctionCatalog {
+
+	private final Set<String> deployed = new HashSet<>();
+
+	private ThinJarAppDeployer deployer;
+
+	public FunctionExtractingFunctionCatalog() {
+		this("thin", "slim");
+	}
+
+	public FunctionExtractingFunctionCatalog(String name, String... profiles) {
+		deployer = new ThinJarAppDeployer(name, profiles);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T> Consumer<T> lookupConsumer(String name) {
+		return (Consumer<T>) find(name, "lookupConsumer");
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T, R> Function<T, R> lookupFunction(String name) {
+		return (Function<T, R>) find(name, "lookupFunction");
+	}
+
+	@Override
+	public <T, R> Function<T, R> composeFunction(String... functionNames) {
+		Function<T, R> function = this.lookupFunction(functionNames[0]);
+		for (int i = 1; i < functionNames.length; i++) {
+			function = function.andThen(this.lookupFunction(functionNames[i]));
+		}
+		return function;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T> Supplier<T> lookupSupplier(String name) {
+		return (Supplier<T>) find(name, "lookupSupplier");
+	}
+
+	public String deploy(AppDeploymentRequest request) {
+		String id = deployer.deploy(request);
+		deployed.add(id);
+		return id;
+	}
+
+	public void undeploy(String id) {
+		deployer.undeploy(id);
+		deployed.remove(id);
+	}
+
+	private Object find(String name, String method) {
+		for (String id : deployed) {
+			Object catalog = deployer.getBean(id, FunctionCatalog.class);
+			if (catalog == null) {
+				continue;
+			}
+			try {
+				MethodInvoker invoker = new MethodInvoker();
+				invoker.setTargetObject(catalog);
+				invoker.setTargetMethod(method);
+				invoker.setArguments(new Object[] { name });
+				invoker.prepare();
+				Object result = invoker.invoke();
+				if (result != null) {
+					return result;
+				}
+			}
+			catch (Exception e) {
+				throw new IllegalStateException("Cannot extract catalog", e);
+			}
+		}
+		return null;
+	}
+
+}
--- a/spring-cloud-function-deployer/src/test/java/org/springframework/cloud/function/deployer/FunctionAppDeployerTests.java
+++ b/spring-cloud-function-deployer/src/test/java/org/springframework/cloud/function/deployer/FunctionAppDeployerTests.java
@@ -0,0 +1,105 @@
+/*
+ * Copyright 2012-2015 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 org.springframework.cloud.function.deployer;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.springframework.boot.loader.thin.ArchiveUtils;
+import org.springframework.boot.loader.tools.LogbackInitializer;
+import org.springframework.cloud.deployer.spi.app.DeploymentState;
+import org.springframework.cloud.deployer.spi.core.AppDefinition;
+import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
+import org.springframework.cloud.deployer.thin.ThinJarAppDeployer;
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.core.io.Resource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@RunWith(Parameterized.class)
+public class FunctionAppDeployerTests {
+
+	static {
+		LogbackInitializer.initialize();
+	}
+
+	private static ThinJarAppDeployer deployer = new ThinJarAppDeployer();
+
+	@BeforeClass
+	public static void skip() {
+		try {
+			ArchiveUtils.getArchiveRoot(ArchiveUtils.getArchive(
+					"maven://org.springframework.cloud:spring-cloud-function-web:1.0.0.BUILD-SNAPSHOT"));
+		}
+		catch (Exception e) {
+			Assume.assumeNoException(
+					"Could not locate jar for tests. Please build spring-cloud-function locally first.",
+					e);
+		}
+	}
+
+	@Parameterized.Parameters
+	public static List<Object[]> data() {
+		// Repeat a couple of times to ensure it's consistent
+		return Arrays.asList(new Object[2][0]);
+	}
+
+	@Test
+	public void web() throws Exception {
+		String first = deploy(
+				"maven://org.springframework.cloud:spring-cloud-function-web:1.0.0.BUILD-SNAPSHOT",
+				"--web.path=/words", "--function.name=uppercase");
+		// Deployment is blocking so it either failed or succeeded.
+		assertThat(deployer.status(first).getState()).isEqualTo(DeploymentState.deployed);
+		deployer.undeploy(first);
+	}
+
+	@Test
+	public void stream() throws Exception {
+		String first = deploy(
+				"maven://org.springframework.cloud:spring-cloud-function-stream:1.0.0.BUILD-SNAPSHOT",
+				"--spring.cloud.stream.bindings.input.destination=words",
+				"--spring.cloud.stream.bindings.output.destination=uppercaseWords",
+				"--function.name=uppercase");
+		// Deployment is blocking so it either failed or succeeded.
+		assertThat(deployer.status(first).getState()).isEqualTo(DeploymentState.deployed);
+		deployer.undeploy(first);
+	}
+
+	private String deploy(String jarName, String... args) throws Exception {
+		Resource resource = new FileSystemResource(
+				ArchiveUtils.getArchiveRoot(ArchiveUtils.getArchive(jarName)));
+		AppDefinition definition = new AppDefinition(resource.getFilename(),
+				Collections.emptyMap());
+		AppDeploymentRequest request = new AppDeploymentRequest(definition, resource,
+				Collections.emptyMap(), Arrays.asList(args));
+		String deployed = deployer.deploy(request);
+		return deployed;
+	}
+
+}
--- a/spring-cloud-function-deployer/src/test/java/org/springframework/cloud/function/deployer/FunctionExtractingFunctionCatalogIntegrationTests.java
+++ b/spring-cloud-function-deployer/src/test/java/org/springframework/cloud/function/deployer/FunctionExtractingFunctionCatalogIntegrationTests.java
@@ -29,7 +29,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 * @author Dave Syer
 *
 */
-public class FunctionExtractingAppDeployerIntegrationTests {
+public class FunctionExtractingFunctionCatalogIntegrationTests {

 	private static ConfigurableApplicationContext context;
 	private static int port;
@@ -37,7 +37,7 @@ public class FunctionExtractingAppDeployerIntegrationTests {
 	@BeforeClass
 	public static void open() {
 		port = SocketUtils.findAvailableTcpPort();
-		System.setProperty("debug", "true");
+		// System.setProperty("debug", "true");
 		context = new ApplicationRunner().start("--server.port=" + port);
 	}

--- a/spring-cloud-function-deployer/src/test/java/org/springframework/cloud/function/deployer/FunctionExtractingFunctionCatalogTests.java
+++ b/spring-cloud-function-deployer/src/test/java/org/springframework/cloud/function/deployer/FunctionExtractingFunctionCatalogTests.java
@@ -25,7 +25,6 @@ import org.junit.rules.ExpectedException;

 import org.springframework.boot.loader.thin.ArchiveUtils;
 import org.springframework.boot.loader.tools.LogbackInitializer;
-import org.springframework.cloud.deployer.spi.app.DeploymentState;
 import org.springframework.cloud.deployer.spi.core.AppDefinition;
 import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
 import org.springframework.core.io.FileSystemResource;
@@ -39,7 +38,7 @@ import reactor.core.publisher.Flux;
 * @author Dave Syer
 *
 */
-public class FunctionExtractingAppDeployerTests {
+public class FunctionExtractingFunctionCatalogTests {

 	private static String id;

@@ -47,7 +46,7 @@ public class FunctionExtractingAppDeployerTests {
 		LogbackInitializer.initialize();
 	}

-	private static FunctionExtractingAppDeployer deployer = new FunctionExtractingAppDeployer();
+	private static FunctionExtractingFunctionCatalog deployer = new FunctionExtractingFunctionCatalog();

 	@Rule
 	public ExpectedException expected = ExpectedException.none();
@@ -58,7 +57,6 @@ public class FunctionExtractingAppDeployerTests {
 			id = deploy("maven://com.example:function-sample:1.0.0.BUILD-SNAPSHOT");
 			// "--debug");
 		}
-		assertThat(deployer.status(id).getState()).isEqualTo(DeploymentState.deployed);
 	}

 	@Test
--- a/spring-cloud-function-stream/pom.xml
+++ b/spring-cloud-function-stream/pom.xml
@@ -15,9 +15,15 @@

 	<properties>
 		<java.version>1.8</java.version>
+		<reactor.version>3.0.2.RELEASE</reactor.version>
 	</properties>

 	<dependencies>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+			<version>${reactor.version}</version>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream</artifactId>
@@ -32,10 +38,6 @@
 			<artifactId>spring-cloud-function-core</artifactId>
 			<version>${project.version}</version>
 		</dependency>
-		<dependency>
-			<groupId>io.projectreactor</groupId>
-			<artifactId>reactor-core</artifactId>
-		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream-binder-rabbit</artifactId>