Add http-request-function and processor app

* Update README * Clean up per review * Added function README * Remove `HttpRequestProcessorProperties$Retry` from the whitelist
2020-05-28 11:01:36 -04:00
parent bdbe8bc231
commit 27e53d1fba
7 changed files with 537 additions and 0 deletions
--- a/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java
+++ b/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2018-2020 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
+ *
+ *      https://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.fn.http.request;
+
+import java.time.Duration;
+import java.util.function.Function;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.ResponseEntity;
+import org.springframework.messaging.Message;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.util.DefaultUriBuilderFactory;
+import org.springframework.web.util.UriBuilderFactory;
+
+/**
+ * Configuration for a {@link Function} that makes HTTP requests to a resource and for
+ * each request, returns a {@link ResponseEntity}.
+ *
+ * @author David Turanski
+ *
+ **/
+@Configuration
+@EnableConfigurationProperties(HttpRequestFunctionProperties.class)
+public class HttpRequestFunctionConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean(WebClient.class)
+	public WebClient webClient() {
+		return WebClient.builder()
+				.build();
+	}
+
+	@Bean
+	public HttpRequestFunction httpRequestFunction(WebClient webClient, HttpRequestProperties properties) {
+		return new HttpRequestFunction(webClient, properties);
+	}
+
+	/**
+	 * Function that accepts a {@code Flux<Message<?>>} containing body and headers and
+	 * returns a {@code Flux<ResponseEntity<?>>}.
+	 */
+	public static class HttpRequestFunction implements Function<Flux<Message<?>>, Flux<ResponseEntity<?>>> {
+		private final WebClient webClient;
+
+		private final UriBuilderFactory uriBuilderFactory = new DefaultUriBuilderFactory();
+
+		private final HttpRequestProperties properties;
+
+		public HttpRequestFunction(WebClient webClient, HttpRequestProperties properties) {
+			this.webClient = webClient;
+			this.properties = properties;
+		}
+
+		@Override
+		public Flux<ResponseEntity<?>> apply(Flux<Message<?>> messageFlux) {
+			return messageFlux.flatMap(message -> this.webClient
+					.method(properties.getHttpMethod())
+					.uri(uriBuilderFactory.uriString(properties.getUrl()).build())
+					.bodyValue(properties.getBody() == null ? message.getPayload() : properties.getBody())
+					.headers(httpHeaders -> httpHeaders.addAll(properties.getHeaders()))
+					.retrieve()
+					.toEntity(properties.getExpectedResponseType())
+					.timeout(Duration.ofMillis(properties.getTimeout())));
+		}
+	}
+}
--- a/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionProperties.java
+++ b/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionProperties.java
@@ -0,0 +1,131 @@
+/*
+ * Copyright 2015-2020 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
+ *
+ *      https://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.fn.http.request;
+
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Configuration properties for the Http Client Processor module.
+ *
+ * @author Waldemar Hummer
+ * @author Mark Fisher
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ * @author David Turanski
+ */
+@Validated
+@ConfigurationProperties("http.request")
+public class HttpRequestFunctionProperties implements HttpRequestProperties {
+	private static final HttpMethod DEFAULT_HTTP_METHOD = HttpMethod.GET;
+
+	private static final Class<?> DEFAULT_RESPONSE_TYPE = String.class;
+	/**
+	 * The URL to issue an http request to, as a static value.
+	 */
+	private String url;
+
+	/**
+	 * The (static) request body; if neither this nor bodyExpression is provided, the payload
+	 * will be used.
+	 */
+	private Object body;
+
+	/**
+	 * The kind of http method to use.
+	 */
+	private HttpMethod httpMethod = DEFAULT_HTTP_METHOD;
+
+	/**
+	 * The type used to interpret the response.
+	 */
+	private Class<?> expectedResponseType = DEFAULT_RESPONSE_TYPE;
+
+	/**
+	 * Request timeout in milliseconds.
+	 */
+	private long timeout = 30_000;
+
+	/**
+	 * A Map of HTTP request headers.
+	 */
+	private HttpHeaders headers = new HttpHeaders();
+
+	@NotEmpty
+	@Override
+	public String getUrl() {
+		return this.url;
+	}
+
+	public void setUrl(String url) {
+		this.url = url;
+	}
+
+	@NotNull
+	@Override
+	public HttpMethod getHttpMethod() {
+		return httpMethod;
+	}
+
+	public void setHttpMethod(HttpMethod httpMethod) {
+		this.httpMethod = httpMethod;
+	}
+
+
+	@NotNull
+	@Override
+	public Class<?> getExpectedResponseType() {
+		return expectedResponseType;
+	}
+
+	public void setExpectedResponseType(Class<?> expectedResponseType) {
+		this.expectedResponseType = expectedResponseType;
+	}
+
+	@Override
+	public Object getBody() {
+		return body;
+	}
+
+	public void setBody(Object body) {
+		this.body = body;
+	}
+
+	@Override
+	public long getTimeout() {
+		return this.timeout;
+	}
+
+	public void setTimeout(long timeout) {
+		this.timeout = timeout;
+	}
+
+	@Override
+	public HttpHeaders getHeaders() {
+		return headers;
+	}
+
+	public void setHeaders(HttpHeaders headers) {
+		this.headers = headers;
+	}
+
+}
--- a/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestProperties.java
+++ b/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestProperties.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2020-2020 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
+ *
+ *      https://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.fn.http.request;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+
+public interface HttpRequestProperties {
+	String getUrl();
+	Object getBody();
+	long getTimeout();
+	HttpHeaders getHeaders();
+	Class<?> getExpectedResponseType();
+	HttpMethod getHttpMethod();
+}
--- a/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionApplicationTests.java
+++ b/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionApplicationTests.java
@@ -0,0 +1,198 @@
+/*
+ * Copyright 2018-2020 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
+ *
+ *      https://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.fn.http.request;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import okhttp3.mockwebserver.Dispatcher;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.fn.http.request.HttpRequestFunctionConfiguration.HttpRequestFunction;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class HttpRequestFunctionApplicationTests {
+	private MockWebServer server;
+
+	private ApplicationContextRunner runner;
+
+	@BeforeEach
+	void setup() {
+		this.server = new MockWebServer();
+		this.runner = new ApplicationContextRunner()
+				.withUserConfiguration(HttpRequestFunctionApplication.class)
+				.withPropertyValues(
+						"http.request.url=" + url());
+	}
+
+	@AfterEach
+	void shutdown() throws IOException {
+		this.server.shutdown();
+	}
+
+	@Test
+	void shouldReturnString() {
+
+		server.enqueue(new MockResponse()
+				.setResponseCode(HttpStatus.OK.value())
+				.setBody("hello"));
+
+		runner.run(context -> {
+			HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class);
+			Message<?> message = MessageBuilder.withPayload("").build();
+			StepVerifier.create(httpRequestFunction.apply(Flux.just(message)))
+					.assertNext((ResponseEntity r) -> {
+						assertThat(r.getBody()).isEqualTo("hello");
+						assertThat(r.getStatusCode().is2xxSuccessful()).isTrue();
+					})
+					.expectComplete()
+					.verify();
+		});
+	}
+
+	@Test
+	void shouldPostJson() {
+
+		server.setDispatcher(new Dispatcher() {
+			@Override
+			public MockResponse dispatch(RecordedRequest recordedRequest) {
+				return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE,
+						recordedRequest.getHeader(HttpHeaders.CONTENT_TYPE))
+						.setBody(recordedRequest.getBody())
+						.setResponseCode(HttpStatus.CREATED.value());
+			}
+		});
+
+		runner.withPropertyValues("http.request.http-method=POST", "http.request.headers[Content-Type]=application/json").run(context -> {
+			HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class);
+			String json = "{\"hello\":\"world\"}";
+			Message<?> message = MessageBuilder.withPayload(json)
+					.build();
+			StepVerifier.create(httpRequestFunction.apply(Flux.just(message)))
+					.assertNext((ResponseEntity r) -> {
+						assertThat(r.getBody()).isEqualTo(json);
+						assertThat(r.getStatusCode().is2xxSuccessful()).isTrue();
+						assertThat(r.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_JSON);
+					})
+					.expectComplete()
+					.verify();
+			RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS);
+			assertThat(request.getMethod()).isEqualTo("POST");
+		});
+	}
+
+	@Test
+	void shouldPostPojoAsJson() {
+
+		server.setDispatcher(new Dispatcher() {
+			@Override
+			public MockResponse dispatch(RecordedRequest recordedRequest) {
+				return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE,
+						recordedRequest.getHeader(HttpHeaders.CONTENT_TYPE))
+						.setBody(recordedRequest.getBody())
+						.setResponseCode(HttpStatus.CREATED.value());
+			}
+		});
+
+		runner.withPropertyValues("http.request.http-method=POST",
+				"http.request.headers[Content-Type]=application/json",
+				"http.request.expected-response-type=" + Map.class.getName()).run(context -> {
+					HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class);
+					Map<String, String> json = Collections.singletonMap("hello", "world");
+					Message<?> message = MessageBuilder.withPayload(json)
+							.build();
+					StepVerifier.create(httpRequestFunction.apply(Flux.just(message)))
+							.assertNext((ResponseEntity r) -> {
+								assertThat(r.getBody()).isEqualTo(json);
+								assertThat(r.getStatusCode().is2xxSuccessful()).isTrue();
+								assertThat(r.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_JSON);
+							})
+							.expectComplete()
+							.verify();
+					RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS);
+					assertThat(request.getMethod()).isEqualTo("POST");
+				});
+	}
+
+	@Test
+	void shouldDelete() {
+
+		server.setDispatcher(new Dispatcher() {
+			@Override
+			public MockResponse dispatch(RecordedRequest recordedRequest) {
+				return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE,
+						recordedRequest.getHeader(HttpHeaders.CONTENT_TYPE))
+						.setBody(recordedRequest.getBody())
+						.setResponseCode(HttpStatus.ACCEPTED.value());
+			}
+		});
+
+		runner.withPropertyValues("http.request.http-method=DELETE",
+				"http.request.expected-response-type=" + Void.class.getName()).run(context -> {
+					HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class);
+					Message<?> message = MessageBuilder.withPayload("")
+							.build();
+					StepVerifier.create(httpRequestFunction.apply(Flux.just(message)))
+							.assertNext((ResponseEntity r) -> {
+								assertThat(r.getBody()).isNull();
+								assertThat(r.getStatusCode().is2xxSuccessful()).isTrue();
+							})
+							.expectComplete()
+							.verify();
+					RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS);
+					assertThat(request.getMethod()).isEqualTo("DELETE");
+				});
+	}
+
+	@Test
+	void shouldThrowErrorIfCannotConnect() throws IOException {
+		server.shutdown();
+		runner.run(context -> {
+			HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class);
+			StepVerifier.create(httpRequestFunction.apply(Flux.just(new GenericMessage(""))))
+					.expectErrorMatches(throwable -> throwable.getMessage().startsWith("Connection refused"))
+					.verify();
+		});
+	}
+
+	private String url() {
+		return String.format("http://localhost:%d", server.getPort());
+	}
+
+	@SpringBootApplication
+	static class HttpRequestFunctionApplication {
+	}
+}