Add new feature to initialize a Supplier from a remote HTTP endpoint

Kind of like the SupplierExporter but to create the Supplier itself.
With this in place you can define the templateUrl (destination) and
the originaUrl (source) and use the app as a pipeline for events
from/to HTTP.

Provide functional bean support for HTTP export

Add autoconfig to AWS adapter for custom runtime

Fix HttpSupplier to always supply Message if headers are included

Fix registration of origin supplier in functional beans

Add docs on new AWS features

Add custom runtime sample

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

@@ -0,0 +1,173 @@
+/*
+ * Copyright 2012-2019 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.web.source;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@ConfigurationProperties("spring.cloud.function.web.export")
+public class ExporterProperties {
+
+	/**
+	 * Flag to indicate that the supplier emits HTTP requests automatically on startup.
+	 */
+	private boolean autoStartup = true;
+
+	/**
+	 * Flag to indicate that extra logging is required for the supplier.
+	 */
+	private boolean debug = true;
+
+	/**
+	 * Properties related to a source of items (via an HTTP GET on startup).
+	 */
+	private Source source = new Source();
+
+	/**
+	 * Properties related to a sink of items (via an HTTP POST on startup).
+	 */
+	private Sink sink = new Sink();
+
+	/**
+	 * Flag to enable the export of a supplier.
+	 */
+	private boolean enabled;
+
+	public boolean isEnabled() {
+		return this.enabled;
+	}
+
+	public void setEnabled(boolean enabled) {
+		this.enabled = enabled;
+	}
+
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public boolean isDebug() {
+		return this.debug;
+	}
+
+	public void setDebug(boolean debug) {
+		this.debug = debug;
+	}
+
+	public Source getSource() {
+		return this.source;
+	}
+
+	public Sink getSink() {
+		return this.sink;
+	}
+
+	public static class Source {
+
+		/**
+		 * URL template for creating a virtual Supplier from HTTP GET.
+		 */
+		private String url;
+
+		/**
+		 * If the origin url is set, the type of content expected (e.g. a POJO class).
+		 * Defaults to String.
+		 */
+		private Class<?> type;
+
+		/**
+		 * Include the incoming headers in the outgoing Supplier. If true the supplier
+		 * will be of generic type Message of T equal to the source type.
+		 */
+		private boolean includeHeaders = true;
+
+		public String getUrl() {
+			return this.url;
+		}
+
+		public void setUrl(String url) {
+			this.url = url;
+		}
+
+		public Class<?> getType() {
+			return this.type == null ? String.class : this.type;
+		}
+
+		public void setType(Class<?> type) {
+			this.type = type;
+		}
+
+		public void setIncludeHeaders(boolean includeHeaders) {
+			this.includeHeaders = includeHeaders;
+		}
+
+		public boolean isIncludeHeaders() {
+			return this.includeHeaders;
+		}
+
+	}
+
+	public static class Sink {
+
+		/**
+		 * URL template for outgoing HTTP requests. Each item from the supplier is POSTed
+		 * to this target.
+		 */
+		private String url;
+
+		/**
+		 * Additional headers to append to the outgoing HTTP requests.
+		 */
+		private Map<String, String> headers = new LinkedHashMap<>();
+
+		/**
+		 * The name of a specific existing Supplier to export from the function catalog.
+		 */
+		private String name;
+
+		public String getName() {
+			return this.name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public String getUrl() {
+			return this.url;
+		}
+
+		public void setUrl(String url) {
+			this.url = url;
+		}
+
+		public Map<String, String> getHeaders() {
+			return this.headers;
+		}
+
+	}
+
+}

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

@@ -16,6 +16,11 @@
 
 package org.springframework.cloud.function.web.source;
 
+import java.util.function.Supplier;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -23,7 +28,9 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnNotWebAppli
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.function.context.FunctionCatalog;
-import org.springframework.cloud.function.web.source.SupplierAutoConfiguration.SourceActiveCondition;
+import org.springframework.cloud.function.context.FunctionRegistration;
+import org.springframework.cloud.function.context.FunctionType;
+import org.springframework.cloud.function.web.source.FunctionExporterAutoConfiguration.SourceActiveCondition;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
@@ -37,26 +44,48 @@ import org.springframework.web.reactive.function.client.WebClient;
 @Configuration
 @ConditionalOnClass(WebClient.class)
 @Conditional(SourceActiveCondition.class)
-@EnableConfigurationProperties(SupplierProperties.class)
-@ConditionalOnProperty(prefix = "spring.cloud.function.web.supplier", name = "enabled", matchIfMissing = true)
-class SupplierAutoConfiguration {
+@EnableConfigurationProperties(ExporterProperties.class)
+@ConditionalOnProperty(prefix = "spring.cloud.function.web.export", name = "enabled", matchIfMissing = true)
+public class FunctionExporterAutoConfiguration {
 
-	@Bean
-	public SupplierExporter sourceForwarder(RequestBuilder requestBuilder,
-			DestinationResolver destinationResolver, FunctionCatalog catalog,
-			WebClient.Builder builder, SupplierProperties props) {
-		return new SupplierExporter(requestBuilder, destinationResolver, catalog,
-				builder.build(), props);
+	private ExporterProperties props;
+
+	@Autowired
+	FunctionExporterAutoConfiguration(ExporterProperties props) {
+		this.props = props;
 	}
 
 	@Bean
-	public RequestBuilder simpleRequestBuilder(SupplierProperties props,
-			Environment environment) {
-		SimpleRequestBuilder builder = new SimpleRequestBuilder(environment);
-		if (props.getTemplateUrl() != null) {
-			builder.setTemplateUrl(props.getTemplateUrl());
+	@ConditionalOnProperty(prefix = "spring.cloud.function.web.export.sink", name = "url")
+	public SupplierExporter sourceForwarder(RequestBuilder requestBuilder,
+			DestinationResolver destinationResolver, FunctionCatalog catalog,
+			WebClient.Builder builder) {
+		return new SupplierExporter(requestBuilder, destinationResolver, catalog,
+				builder.build(), this.props);
+	}
+
+	@Bean
+	@ConditionalOnProperty(prefix = "spring.cloud.function.web.export.source", name = "url")
+	public FunctionRegistration<Supplier<Flux<?>>> origin(WebClient.Builder builder) {
+		HttpSupplier supplier = new HttpSupplier(builder.build(), this.props);
+		FunctionRegistration<Supplier<Flux<?>>> registration = new FunctionRegistration<>(
+				supplier);
+		FunctionType type = FunctionType.supplier(this.props.getSource().getType())
+				.wrap(Flux.class);
+		if (this.props.getSource().isIncludeHeaders()) {
+			type = type.message();
 		}
-		builder.setHeaders(props.getHeaders());
+		registration = registration.type(type);
+		return registration;
+	}
+
+	@Bean
+	public RequestBuilder simpleRequestBuilder(Environment environment) {
+		SimpleRequestBuilder builder = new SimpleRequestBuilder(environment);
+		if (this.props.getSink().getUrl() != null) {
+			builder.setTemplateUrl(this.props.getSink().getUrl());
+		}
+		builder.setHeaders(this.props.getSink().getHeaders());
 		return builder;
 	}
 
@@ -77,7 +106,7 @@ class SupplierAutoConfiguration {
 
 		}
 
-		@ConditionalOnProperty(prefix = "spring.cloud.function.web.supplier", name = "enabled")
+		@ConditionalOnProperty(prefix = "spring.cloud.function.web.export", name = "enabled")
 		static class Enabled {
 
 		}

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

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2012-2019 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.web.source;
+
+import org.springframework.boot.web.reactive.context.ReactiveWebApplicationContext;
+import org.springframework.cloud.function.context.FunctionCatalog;
+import org.springframework.cloud.function.context.FunctionRegistration;
+import org.springframework.cloud.function.context.config.ContextFunctionCatalogInitializer;
+import org.springframework.context.ApplicationContextInitializer;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.util.ClassUtils;
+import org.springframework.web.context.WebApplicationContext;
+import org.springframework.web.reactive.function.client.WebClient;
+
+/**
+ * @author Dave Syer
+ * @since 2.0
+ *
+ */
+class FunctionExporterInitializer
+		implements ApplicationContextInitializer<GenericApplicationContext> {
+
+	@Override
+	public void initialize(GenericApplicationContext context) {
+		if (ContextFunctionCatalogInitializer.enabled && context.getEnvironment()
+				.getProperty("spring.functional.enabled", Boolean.class, false)
+				&& isExporting(context)) {
+			registerWebClient(context);
+			registerExport(context);
+		}
+	}
+
+	private void registerWebClient(GenericApplicationContext context) {
+		if (context.getBeanFactory().getBeanNamesForType(WebClient.Builder.class, false,
+				false).length == 0) {
+			context.registerBean(WebClient.Builder.class, () -> WebClient.builder());
+		}
+	}
+
+	private boolean isExporting(GenericApplicationContext context) {
+		if (context.getEnvironment().getProperty("spring.cloud.function.web.export",
+				Boolean.class, false)) {
+			return true;
+		}
+		if (ClassUtils.isPresent("org.springframework.web.context.WebApplicationContext",
+				getClass().getClassLoader())) {
+			if (context instanceof WebApplicationContext
+					|| context instanceof ReactiveWebApplicationContext) {
+				return false;
+			}
+		}
+		return true;
+	}
+
+	private void registerExport(GenericApplicationContext context) {
+		context.registerBean(ExporterProperties.class, () -> new ExporterProperties());
+		context.registerBean(FunctionExporterAutoConfiguration.class,
+				() -> new FunctionExporterAutoConfiguration(
+						context.getBean(ExporterProperties.class)));
+		if (context.getBeanFactory().getBeanNamesForType(DestinationResolver.class, false,
+				false).length == 0) {
+			context.registerBean(DestinationResolver.class,
+					() -> context.getBean(FunctionExporterAutoConfiguration.class)
+							.simpleDestinationResolver());
+		}
+		if (context.getBeanFactory().getBeanNamesForType(RequestBuilder.class, false,
+				false).length == 0) {
+			context.registerBean(RequestBuilder.class,
+					() -> context.getBean(FunctionExporterAutoConfiguration.class)
+							.simpleRequestBuilder(context.getEnvironment()));
+		}
+		if (context.getEnvironment()
+				.getProperty("spring.cloud.function.web.export.source.url") != null) {
+			context.registerBean("origin", FunctionRegistration.class,
+					() -> context.getBean(FunctionExporterAutoConfiguration.class)
+							.origin(context.getBean(WebClient.Builder.class)));
+		}
+		if (context.getEnvironment()
+				.getProperty("spring.cloud.function.web.export.sink.url") != null) {
+			context.registerBean(SupplierExporter.class,
+					() -> context.getBean(FunctionExporterAutoConfiguration.class)
+							.sourceForwarder(context.getBean(RequestBuilder.class),
+									context.getBean(DestinationResolver.class),
+									context.getBean(FunctionCatalog.class),
+									context.getBean(WebClient.Builder.class)));
+		}
+	}
+
+}

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

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2018-2019 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.web.source;
+
+import java.util.function.Supplier;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.function.web.util.HeaderUtils;
+import org.springframework.http.HttpStatus;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.web.reactive.function.client.ClientResponse;
+import org.springframework.web.reactive.function.client.WebClient;
+
+/**
+ * A {@link Supplier} that pulls data from an HTTP endpoint. Repeatedly polls the endpoint
+ * until a non-2xx response is received.
+ *
+ * @author Dave Syer
+ */
+public class HttpSupplier implements Supplier<Flux<?>> {
+
+	private static Log logger = LogFactory.getLog(HttpSupplier.class);
+
+	private WebClient client;
+
+	private ExporterProperties props;
+
+	/**
+	 * @param client the WebClient to use. The baseUrl should be set.
+	 * @param props the ExporterProperties to use to parameterize the requests.
+	 */
+	public HttpSupplier(WebClient client, ExporterProperties props) {
+		this.client = client;
+		this.props = props;
+	}
+
+	@Override
+	public Flux<?> get() {
+		return get(this.client);
+	}
+
+	private Flux<?> get(WebClient client) {
+		Flux<?> result = client.get().uri(this.props.getSource().getUrl()).exchange()
+				.flatMap(this::transform).repeat();
+		if (this.props.isDebug()) {
+			result = result.log();
+		}
+		return result.onErrorResume(TerminateException.class, error -> Mono.empty());
+	}
+
+	private Mono<?> transform(ClientResponse response) {
+		HttpStatus status = response.statusCode();
+		if (!status.is2xxSuccessful()) {
+			if (this.props.isDebug()) {
+				logger.info("Terminated supplier with status=" + response.statusCode());
+			}
+			return Mono.error(TerminateException.INSTANCE);
+		}
+		return response.bodyToMono(this.props.getSource().getType())
+				.map(value -> message(response, value));
+	}
+
+	private Object message(ClientResponse response, Object payload) {
+		if (!this.props.getSource().isIncludeHeaders()) {
+			return payload;
+		}
+		return MessageBuilder.withPayload(payload)
+				.copyHeaders(HeaderUtils.fromHttp(
+						HeaderUtils.sanitize(response.headers().asHttpHeaders())))
+				.build();
+	}
+
+	private static class TerminateException extends RuntimeException {
+
+		static final TerminateException INSTANCE = new TerminateException();
+
+		TerminateException() {
+			super("Planned termination");
+		}
+
+		@Override
+		public synchronized Throwable fillInStackTrace() {
+			return this;
+		}
+
+	}
+
+}

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

@@ -18,11 +18,15 @@ package org.springframework.cloud.function.web.source;
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
+import org.springframework.cloud.function.web.util.HeaderUtils;
 import org.springframework.core.env.Environment;
 import org.springframework.http.HttpHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 
 /**
  * @author Dave Syer
@@ -42,13 +46,17 @@ class SimpleRequestBuilder implements RequestBuilder {
 
 	@Override
 	public HttpHeaders headers(String destination, Object value) {
-		// TODO: add message headers if any
-		HttpHeaders result = new HttpHeaders();
+		MessageHeaders incoming = new MessageHeaders(Collections.emptyMap());
+		if (value instanceof Message) {
+			Message<?> message = (Message<?>) value;
+			incoming = message.getHeaders();
+		}
+		HttpHeaders result = HeaderUtils.fromMessage(incoming);
 		for (String key : this.headers.keySet()) {
 			String header = this.headers.get(key);
 			header = header.replace("${destination}", destination);
 			header = this.environment.resolvePlaceholders(header);
-			result.add(key, header);
+			result.set(key, header);
 		}
 		return result;
 	}
@@ -56,8 +64,8 @@ class SimpleRequestBuilder implements RequestBuilder {
 	@Override
 	public URI uri(String destination) {
 		try {
-			return new URI(this.environment.resolvePlaceholders(
-					this.baseUrl.replace("${destination}", destination)));
+			return new URI(this.baseUrl.replace("${destination}", destination)
+					.replace("{{destination}}", destination));
 		}
 		catch (URISyntaxException e) {
 			throw new IllegalStateException("Cannot create URI", e);

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

@@ -21,15 +21,16 @@ import java.util.Collections;
 import java.util.Set;
 import java.util.function.Supplier;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.http.HttpHeaders;
-import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.messaging.Message;
 import org.springframework.web.reactive.function.client.ClientResponse;
 import org.springframework.web.reactive.function.client.WebClient;
 
@@ -40,7 +41,9 @@ import org.springframework.web.reactive.function.client.WebClient;
  * @author Dave Syer
  *
  */
-class SupplierExporter implements SmartLifecycle {
+public class SupplierExporter implements SmartLifecycle {
+
+	private static Log logger = LogFactory.getLog(SupplierExporter.class);
 
 	private final FunctionCatalog catalog;
 
@@ -64,14 +67,14 @@ class SupplierExporter implements SmartLifecycle {
 
 	SupplierExporter(RequestBuilder requestBuilder,
 			DestinationResolver destinationResolver, FunctionCatalog catalog,
-			WebClient client, SupplierProperties props) {
+			WebClient client, ExporterProperties props) {
 		this.requestBuilder = requestBuilder;
 		this.destinationResolver = destinationResolver;
 		this.catalog = catalog;
 		this.client = client;
 		this.debug = props.isDebug();
 		this.autoStartup = props.isAutoStartup();
-		this.supplier = props.getName();
+		this.supplier = props.getSink().getName();
 	}
 
 	@Override
@@ -82,19 +85,24 @@ class SupplierExporter implements SmartLifecycle {
 
 		this.running = true;
 		this.ok = true;
+		logger.info("Starting");
 
 		Flux<Object> streams = Flux.empty();
 		Set<String> names = this.supplier == null ? this.catalog.getNames(Supplier.class)
 				: Collections.singleton(this.supplier);
 		for (String name : names) {
 			Supplier<Flux<Object>> supplier = this.catalog.lookup(Supplier.class, name);
+			if (supplier == null) {
+				logger.warn("No such Supplier: " + name);
+				continue;
+			}
 			streams = streams.mergeWith(forward(supplier, name));
 		}
 
 		this.subscription = streams.doOnError(error -> {
 			this.ok = false;
 			if (!this.debug) {
-				error.printStackTrace();
+				logger.info(error);
 			}
 		}).doOnTerminate(() -> this.running = false).doOnNext(value -> {
 			if (this.subscription != null && !this.running) {
@@ -104,17 +112,25 @@ class SupplierExporter implements SmartLifecycle {
 	}
 
 	private Flux<ClientResponse> forward(Supplier<Flux<Object>> supplier, String name) {
-		return supplier.get().publishOn(Schedulers.parallel()).flatMap(value -> {
+		return supplier.get().flatMap(value -> {
 			String destination = this.destinationResolver.destination(supplier, name,
 					value);
+			if (this.debug) {
+				logger.info("Posting to: " + destination);
+			}
 			return post(uri(destination), destination, value);
 		});
 	}
 
 	private Mono<ClientResponse> post(URI uri, String destination, Object value) {
+		Object body = value;
+		if (value instanceof Message) {
+			Message<?> message = (Message<?>) value;
+			body = message.getPayload();
+		}
 		Mono<ClientResponse> result = this.client.post().uri(uri)
-				.headers(headers -> headers(headers, destination, value))
-				.body(BodyInserters.fromObject(value)).exchange();
+				.headers(headers -> headers(headers, destination, value)).syncBody(body)
+				.exchange();
 		if (this.debug) {
 			result = result.log();
 		}
@@ -135,6 +151,7 @@ class SupplierExporter implements SmartLifecycle {
 
 	@Override
 	public void stop() {
+		logger.info("Stopping");
 		this.running = false;
 	}
 

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

@@ -23,9 +23,10 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 
 /**
  * @author Dave Syer
- *
+ * @deprecated in favour of {@link ExporterProperties}
  */
 @ConfigurationProperties("spring.cloud.function.web.supplier")
+@Deprecated
 public class SupplierProperties {
 
 	private boolean autoStartup = true;

spring-cloud-function-web/src/main/resources/META-INF/spring.factories

@@ -1,9 +1,10 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
 org.springframework.cloud.function.web.flux.ReactorAutoConfiguration,\
 org.springframework.cloud.function.web.mvc.ReactorAutoConfiguration,\
-org.springframework.cloud.function.web.source.SupplierAutoConfiguration
+org.springframework.cloud.function.web.source.FunctionExporterAutoConfiguration
 org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc=\
 org.springframework.cloud.function.web.flux.ReactorAutoConfiguration,\
 org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration
 org.springframework.context.ApplicationContextInitializer=\
-org.springframework.cloud.function.web.function.FunctionEndpointInitializer
+org.springframework.cloud.function.web.function.FunctionEndpointInitializer,\
+org.springframework.cloud.function.web.source.FunctionExporterInitializer

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

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2012-2019 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.test;
+
+import java.util.function.Function;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
+import org.springframework.cloud.function.context.FunctionRegistration;
+import org.springframework.cloud.function.context.FunctionType;
+import org.springframework.cloud.function.context.test.FunctionalSpringBootTest;
+import org.springframework.cloud.function.test.FunctionalExporterTests.ApplicationConfiguration;
+import org.springframework.cloud.function.web.source.SupplierExporter;
+import org.springframework.context.ApplicationContextInitializer;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.SocketUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@RunWith(SpringRunner.class)
+@FunctionalSpringBootTest(classes = ApplicationConfiguration.class, webEnvironment = WebEnvironment.NONE, properties = {
+		"spring.main.web-application-type=none",
+		"spring.cloud.function.web.export.sink.url=http://localhost:${my.port}",
+		"spring.cloud.function.web.export.source.url=http://localhost:${my.port}",
+		"spring.cloud.function.web.export.sink.name=origin|uppercase",
+		"spring.cloud.function.web.export.debug=true",
+		"spring.cloud.function.web.export.enabled=true" })
+public class FunctionalExporterTests {
+
+	@Autowired
+	private SupplierExporter forwarder;
+
+	private static RestConfiguration app;
+
+	private static ConfigurableApplicationContext context;
+
+	@BeforeClass
+	public static void init() throws Exception {
+		String port = "" + SocketUtils.findAvailableTcpPort();
+		System.setProperty("server.port", port);
+		System.setProperty("my.port", port);
+		context = SpringApplication.run(RestConfiguration.class,
+				"--spring.main.web-application-type=reactive");
+		app = context.getBean(RestConfiguration.class);
+		// Sometimes the server doesn't start quick enough
+		Thread.sleep(500L);
+	}
+
+	@AfterClass
+	public static void close() {
+		System.clearProperty("server.port");
+		if (context != null) {
+			context.close();
+		}
+	}
+
+	@Test
+	public void words() throws Exception {
+		int count = 0;
+		while (this.forwarder.isRunning() && count++ < 1000) {
+			Thread.sleep(20);
+		}
+		// It completed
+		assertThat(this.forwarder.isRunning()).isFalse();
+		assertThat(FunctionalExporterTests.app.inputs).contains("HELLO");
+		assertThat(this.forwarder.isOk()).isTrue();
+	}
+
+	@SpringBootConfiguration
+	protected static class ApplicationConfiguration
+			implements ApplicationContextInitializer<GenericApplicationContext> {
+
+		Function<Message<String>, Message<String>> uppercase() {
+			return value -> MessageBuilder.withPayload(value.getPayload().toUpperCase())
+					.copyHeaders(value.getHeaders()).build();
+		}
+
+		@Override
+		public void initialize(GenericApplicationContext context) {
+			context.registerBean("uppercase", FunctionRegistration.class,
+					() -> new FunctionRegistration<>(uppercase()).type(
+							FunctionType.from(String.class).to(String.class).message()));
+		}
+
+	}
+
+}

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2018-2019 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.test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+@SpringBootConfiguration
+@EnableAutoConfiguration
+@RestController
+public class RestConfiguration {
+
+	private static Log logger = LogFactory.getLog(RestConfiguration.class);
+
+	List<String> inputs = new ArrayList<>();
+
+	private Iterator<String> outputs = Arrays.asList("hello", "world").iterator();
+
+	@GetMapping("/")
+	ResponseEntity<String> home() {
+		logger.info("HOME");
+		if (this.outputs.hasNext()) {
+			return ResponseEntity.ok(this.outputs.next());
+		}
+		return ResponseEntity.notFound().build();
+	}
+
+	@PostMapping("/")
+	ResponseEntity<String> accept(@RequestBody String body) {
+		logger.info("ACCEPT");
+		this.inputs.add(body);
+		return ResponseEntity.accepted().body(body);
+	}
+
+	public static void main(String[] args) throws Exception {
+		SpringApplication.run(RestConfiguration.class, args);
+	}
+
+}

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

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2018-2019 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.web.source;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Function;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.cloud.function.web.source.FunctionAutoConfigurationIntegrationTests.ApplicationConfiguration;
+import org.springframework.cloud.function.web.source.FunctionAutoConfigurationIntegrationTests.RestConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.ResponseEntity;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.SocketUtils;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = WebEnvironment.DEFINED_PORT, properties = {
+		"spring.cloud.function.web.export.sink.url=http://localhost:${server.port}",
+		"spring.cloud.function.web.export.source.url=http://localhost:${server.port}",
+		"spring.cloud.function.web.export.sink.name=origin|uppercase",
+		"spring.cloud.function.web.export.debug=true",
+		"spring.cloud.function.web.export.enabled=true" })
+@ContextConfiguration(classes = { RestConfiguration.class,
+		ApplicationConfiguration.class })
+public class FunctionAutoConfigurationIntegrationTests {
+
+	@Autowired
+	private SupplierExporter forwarder;
+
+	@Autowired
+	private RestConfiguration app;
+
+	@BeforeClass
+	public static void init() {
+		System.setProperty("server.port", "" + SocketUtils.findAvailableTcpPort());
+	}
+
+	@AfterClass
+	public static void close() {
+		System.clearProperty("server.port");
+	}
+
+	@Test
+	public void copiesMessages() throws Exception {
+		int count = 0;
+		while (this.forwarder.isRunning() && count++ < 100) {
+			Thread.sleep(20);
+		}
+		// It completed
+		assertThat(this.forwarder.isRunning()).isFalse();
+		assertThat(this.forwarder.isOk()).isTrue();
+		assertThat(this.app.inputs).contains("HELLO");
+		assertThat(this.app.inputs).contains("WORLD");
+	}
+
+	@EnableAutoConfiguration
+	@TestConfiguration
+	public static class ApplicationConfiguration {
+
+		@Bean
+		public Function<String, String> uppercase() {
+			return value -> value.toUpperCase();
+		}
+
+	}
+
+	@TestConfiguration
+	@RestController
+	public static class RestConfiguration {
+
+		private static Log logger = LogFactory.getLog(RestConfiguration.class);
+
+		private List<String> inputs = new ArrayList<>();
+
+		private Iterator<String> outputs = Arrays.asList("hello", "world").iterator();
+
+		@GetMapping("/")
+		ResponseEntity<String> home() {
+			logger.info("HOME");
+			if (this.outputs.hasNext()) {
+				return ResponseEntity.ok(this.outputs.next());
+			}
+			return ResponseEntity.notFound().build();
+		}
+
+		@PostMapping("/")
+		void accept(@RequestBody String body) {
+			logger.info("ACCEPT");
+			this.inputs.add(body);
+		}
+
+	}
+
+}

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

@@ -26,7 +26,6 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
 import org.springframework.boot.test.context.TestConfiguration;
-import org.springframework.cloud.function.web.RestApplication;
 import org.springframework.cloud.function.web.source.SourceAutoConfigurationIntegrationTests.ApplicationConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.test.context.ContextConfiguration;
@@ -39,9 +38,11 @@ import static org.assertj.core.api.Assertions.assertThat;
  *
  */
 @RunWith(SpringRunner.class)
-@SpringBootTest(webEnvironment = WebEnvironment.NONE, //
-		properties = "spring.cloud.function.web.supplier.templateUrl=http://localhost:9999/notthere")
-@ContextConfiguration(classes = { RestApplication.class, ApplicationConfiguration.class })
+// @formatter:off
+@SpringBootTest(webEnvironment = WebEnvironment.NONE,
+	properties = "spring.cloud.function.web.export.sink.url=http://nosuchhost")
+// @formatter:on
+@ContextConfiguration(classes = { ApplicationConfiguration.class })
 public class SourceAutoConfigurationIntegrationTests {
 
 	@Autowired

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

@@ -24,6 +24,8 @@ import java.util.function.Supplier;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -37,6 +39,7 @@ import org.springframework.cloud.function.web.source.WebAppIntegrationTests.Appl
 import org.springframework.context.annotation.Bean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.SocketUtils;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
@@ -48,13 +51,13 @@ import static org.assertj.core.api.Assertions.assertThat;
  *
  */
 @RunWith(SpringRunner.class)
-@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT, properties = {
+@SpringBootTest(webEnvironment = WebEnvironment.DEFINED_PORT, properties = {
 		"spring.main.web-application-type=reactive",
-		"spring.cloud.function.web.supplier.templateUrl=http://localhost:${local.server.port}/values",
+		"spring.cloud.function.web.export.sink.url=http://localhost:${server.port}/values",
 		// in a webapp we have to explicitly enable the export
-		"spring.cloud.function.web.supplier.enabled=true",
+		"spring.cloud.function.web.export.enabled=true",
 		// manually so we know the webapp is listening when we start
-		"spring.cloud.function.web.supplier.autoStartup=false" })
+		"spring.cloud.function.web.export.autoStartup=false" })
 @ContextConfiguration(classes = { RestApplication.class, ApplicationConfiguration.class })
 public class WebAppIntegrationTests {
 
@@ -66,6 +69,16 @@ public class WebAppIntegrationTests {
 	@Autowired
 	private ApplicationConfiguration app;
 
+	@BeforeClass
+	public static void init() {
+		System.setProperty("server.port", "" + SocketUtils.findAvailableTcpPort());
+	}
+
+	@AfterClass
+	public static void close() {
+		System.clearProperty("server.port");
+	}
+
 	@Test
 	public void posts() throws Exception {
 		this.forwarder.start();