Add SupplierExporter (HTTP POST of Suppliers on startup)

Add throwable handling to azure base class

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

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2018 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;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public interface DestinationResolver {
+
+	String destination(Supplier<?> supplier, String name, Object value);
+}

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2018 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.net.URI;
+
+import org.springframework.http.HttpHeaders;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public interface RequestBuilder {
+	
+	URI uri(String destination);
+	
+	HttpHeaders headers(String destination, Object value);
+
+}

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2018 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;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class SimpleDestinationResolver implements DestinationResolver {
+
+	@Override
+	public String destination(Supplier<?> supplier, String name, Object value) {
+		return name.contains("|") ? name.substring(0, name.indexOf("|")).trim() : name;
+	}
+
+}

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

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2018 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.net.URI;
+import java.net.URISyntaxException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.core.env.Environment;
+import org.springframework.http.HttpHeaders;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class SimpleRequestBuilder implements RequestBuilder {
+
+	private String baseUrl = "http://${destination}";
+	private Map<String, String> headers = new LinkedHashMap<>();
+	private Environment environment;
+
+	public SimpleRequestBuilder(Environment environment) {
+		this.environment = environment;
+	}
+
+	@Override
+	public HttpHeaders headers(String destination, Object value) {
+		// TODO: add message headers if any
+		HttpHeaders result = new HttpHeaders();
+		for (String key : headers.keySet()) {
+			String header = headers.get(key);
+			header = header.replace("${destination}", destination);
+			header = environment.resolvePlaceholders(header);
+			result.add(key, header);
+		}
+		return result;
+	}
+
+	@Override
+	public URI uri(String destination) {
+		try {
+			return new URI(environment
+					.resolvePlaceholders(baseUrl.replace("${destination}", destination)));
+		}
+		catch (URISyntaxException e) {
+			throw new IllegalStateException("Cannot create URI", e);
+		}
+	}
+
+	public void setTemplateUrl(String baseUrl) {
+		this.baseUrl = baseUrl;
+	}
+
+	public void setHeaders(Map<String, String> headers) {
+		this.headers.putAll(headers);
+	}
+
+}

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2018 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.autoconfigure.condition.AnyNestedCondition;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnNotWebApplication;
+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.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+import org.springframework.web.reactive.function.client.WebClient;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@Configuration
+@ConditionalOnClass(WebClient.class)
+@Conditional(SourceActiveCondition.class)
+@EnableConfigurationProperties(SupplierProperties.class)
+@ConditionalOnProperty(prefix = "spring.cloud.function.web.supplier", name = "enabled", matchIfMissing = true)
+public class SupplierAutoConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean
+	public SupplierExporter sourceForwarder(RequestBuilder requestBuilder,
+			DestinationResolver destinationResolver, FunctionCatalog catalog,
+			WebClient.Builder builder, SupplierProperties props) {
+		return new SupplierExporter(requestBuilder, destinationResolver, catalog,
+				builder.build(), props);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public RequestBuilder simpleRequestBuilder(SupplierProperties props,
+			Environment environment) {
+		SimpleRequestBuilder builder = new SimpleRequestBuilder(environment);
+		if (props.getTemplateUrl() != null) {
+			builder.setTemplateUrl(props.getTemplateUrl());
+		}
+		builder.setHeaders(props.getHeaders());
+		return builder;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public DestinationResolver simpleDestinationResolver() {
+		return new SimpleDestinationResolver();
+	}
+
+	static class SourceActiveCondition extends AnyNestedCondition {
+
+		public SourceActiveCondition() {
+			super(ConfigurationPhase.PARSE_CONFIGURATION);
+		}
+
+		@ConditionalOnNotWebApplication
+		static class OnNotWebapp {
+		}
+
+		@ConditionalOnProperty(prefix = "spring.cloud.function.web.supplier", name = "enabled")
+		static class Enabled {
+		}
+	}
+}

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

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2018 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.net.URI;
+import java.util.Collections;
+import java.util.Set;
+import java.util.function.Supplier;
+
+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.web.reactive.function.client.ClientResponse;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+/**
+ * Forwards items obtained from a {@link Supplier} or set of suppliers to an external HTTP
+ * endpoint.
+ * 
+ * @author Dave Syer
+ *
+ */
+public class SupplierExporter implements SmartLifecycle {
+
+	private FunctionCatalog catalog;
+	private WebClient client;
+	private volatile boolean running;
+	private volatile boolean ok = true;
+	private boolean autoStartup = true;
+	private boolean debug = true;
+	private String supplier;
+	private volatile Disposable subscription;
+
+	private DestinationResolver destinationResolver;
+	private RequestBuilder requestBuilder;
+
+	public SupplierExporter(RequestBuilder requestBuilder,
+			DestinationResolver destinationResolver, FunctionCatalog catalog,
+			WebClient client, SupplierProperties 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();
+	}
+
+	@Override
+	public void start() {
+		
+		if (this.running) {
+			return;
+		}
+
+		this.running = true;
+		this.ok = true;
+
+		Flux<Object> streams = Flux.empty();
+		Set<String> names = this.supplier == null ? catalog.getNames(Supplier.class)
+				: Collections.singleton(this.supplier);
+		for (String name : names) {
+			Supplier<Flux<Object>> supplier = catalog.lookup(Supplier.class, name);
+			streams = streams.mergeWith(forward(supplier, name));
+		}
+
+		this.subscription = streams.doOnError(error -> {
+			this.ok = false;
+			if (!this.debug) {
+				error.printStackTrace();
+			}
+		}).doOnTerminate(() -> this.running = false).doOnNext(value -> {
+			if (this.subscription != null && !this.running) {
+				this.subscription.dispose();
+			}
+		}).subscribe();
+	}
+
+	private Flux<ClientResponse> forward(Supplier<Flux<Object>> supplier, String name) {
+		return supplier.get().publishOn(Schedulers.parallel()).flatMap(value -> {
+			String destination = destinationResolver.destination(supplier, name, value);
+			return post(uri(destination), destination, value);
+		});
+	}
+
+	private Mono<ClientResponse> post(URI uri, String destination, Object value) {
+		Mono<ClientResponse> result = client.post().uri(uri)
+				.headers(headers -> headers(headers, destination, value))
+				.body(BodyInserters.fromObject(value)).exchange();
+		if (this.debug) {
+			result = result.log();
+		}
+		return result;
+	}
+
+	private void headers(HttpHeaders headers, String destination, Object value) {
+		headers.putAll(requestBuilder.headers(destination, value));
+	}
+
+	private URI uri(String destination) {
+		return requestBuilder.uri(destination);
+	}
+
+	public boolean isOk() {
+		return this.ok;
+	}
+
+	@Override
+	public void stop() {
+		this.running = false;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	@Override
+	public int getPhase() {
+		return 0;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		stop();
+		callback.run();
+	}
+
+}

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

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2018 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.supplier")
+public class SupplierProperties {
+
+	private boolean autoStartup = true;
+	private boolean debug = true;
+	private String name;
+	private String templateUrl;
+	private boolean enabled;
+	public boolean isEnabled() {
+		return this.enabled;
+	}
+
+	public void setEnabled(boolean enabled) {
+		this.enabled = enabled;
+	}
+
+	private Map<String, String> headers = new LinkedHashMap<>();
+
+	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 void setName(String name) {
+		this.name = name;
+	}
+
+	public String getName() {
+		return this.name;
+	}
+
+	public void setTemplateUrl(String templateUrl) {
+		this.templateUrl = templateUrl;
+	}
+
+	public String getTemplateUrl() {
+		return this.templateUrl;
+	}
+	
+	public Map<String, String> getHeaders() {
+		return this.headers;
+	}
+}

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

@@ -1,6 +1,7 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
 org.springframework.cloud.function.web.flux.ReactorAutoConfiguration,\
-org.springframework.cloud.function.web.mvc.ReactorAutoConfiguration
+org.springframework.cloud.function.web.mvc.ReactorAutoConfiguration,\
+org.springframework.cloud.function.web.source.SupplierAutoConfiguration
 
 org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc=\
 org.springframework.cloud.function.web.flux.ReactorAutoConfiguration,\

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

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2018 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.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.RestApplication;
+import org.springframework.cloud.function.web.source.SourceAutoConfigurationIntegrationTests.ApplicationConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = WebEnvironment.NONE)
+@ContextConfiguration(classes = { RestApplication.class, ApplicationConfiguration.class })
+public class SourceAutoConfigurationIntegrationTests {
+
+	@Autowired
+	private SupplierExporter forwarder;
+
+	@Test
+	public void fails() throws Exception {
+		int count = 0;
+		while(forwarder.isRunning() && count++<100) {
+			Thread.sleep(20);
+		}
+		// It completed
+		assertThat(forwarder.isRunning()).isFalse();
+		// But failed
+		assertThat(forwarder.isOk()).isFalse();
+	}
+
+	@EnableAutoConfiguration
+	@TestConfiguration
+	public static class ApplicationConfiguration {
+		@Bean
+		public Supplier<String> word() {
+			return () -> "foo";
+		}
+	}
+}

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

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2018 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.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+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.RestApplication;
+import org.springframework.cloud.function.web.source.WebAppIntegrationTests.ApplicationConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+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.RANDOM_PORT, properties = {
+		"spring.main.web-application-type=reactive",
+		"spring.cloud.function.web.supplier.templateUrl=http://localhost:${local.server.port}/values",
+		// in a webapp we have to explicitly enable the export
+		"spring.cloud.function.web.supplier.enabled=true",
+		// manually so we know the webapp is listening when we start
+		"spring.cloud.function.web.supplier.autoStartup=false"})
+@ContextConfiguration(classes = { RestApplication.class, ApplicationConfiguration.class })
+public class WebAppIntegrationTests {
+	
+	private static Log logger = LogFactory.getLog(WebAppIntegrationTests.class);
+	
+	@Autowired
+	private SupplierExporter forwarder;
+
+	@Autowired
+	private ApplicationConfiguration app;
+
+	@Test
+	public void posts() throws Exception {
+		forwarder.start();
+		app.latch.await(10, TimeUnit.SECONDS);
+		assertThat(app.values).hasSize(1);
+	}
+
+	@EnableAutoConfiguration
+	@TestConfiguration
+	@RestController
+	public static class ApplicationConfiguration {
+		private List<String> values = new ArrayList<>();
+		private CountDownLatch latch = new CountDownLatch(1);
+
+		@Bean
+		public Supplier<String> word() {
+			return () -> "foo";
+		}
+
+		// An endpoint to catch the values being exported
+		@PostMapping("/values")
+		public String value(@RequestBody String body) {
+			logger.info("Body: " + body);
+			values.add(body);
+			latch.countDown();
+			return "ok";
+		}
+	}
+}