diff --git a/function/http-request-function/README.adoc b/function/http-request-function/README.adoc new file mode 100644 index 00000000..efe44b2e --- /dev/null +++ b/function/http-request-function/README.adoc @@ -0,0 +1,32 @@ +# HTTP Request Function + +This module provides an HTTP request function that can be reused and composed in other applications. +The `Function` uses the reactive `WebClient` from `Spring WebFlux` and is implemented as a `java.util.function.Function`. +This function gives you a reactive stream of `ResponseEntity` given a stream of request messages as the function a signature of `Function,Flux>`. +Users have to subscribe to the returned `Flux` to receive the data. + +## Beans for injection + +You can import the `HttpRequestFunction` configuration in a Spring Boot application and then inject the following bean. + +`httpRequestFunction` + +You may inject this as `HttpRequestFunction`. + +You can use `httpRequestFunction` as a qualifier when injecting. + +Once injected, you can use the `apply` method of the `Function` to invoke it and then subscribe to the returned `Flux`. + +## Configuration Options + +All configuration properties are prefixed with `http.request`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionProperties.java[HttpRequestFunctionProperties.java] + +## Tests + +See this link:src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionApplicationTests.java[test suite] for examples of how this function is used. + +## Other usage + +See this link:../../../applications/processor/http-request-processor/README.adoc[README] where this function is used to create a Spring Cloud Stream application to process HTTP requests. \ No newline at end of file diff --git a/function/http-request-function/pom.xml b/function/http-request-function/pom.xml new file mode 100644 index 00000000..4b12e820 --- /dev/null +++ b/function/http-request-function/pom.xml @@ -0,0 +1,61 @@ + + + 4.0.0 + http-request-function + 1.0.0-SNAPSHOT + http-request-function + Spring Native Function for Submitting an HTTP request + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + 4.6.0 + + + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework + spring-messaging + + + org.springframework.boot + spring-boot-starter-webflux + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + com.squareup.okhttp3 + mockwebserver + test + + + io.projectreactor + reactor-test + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + + diff --git 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 new file mode 100644 index 00000000..ab7e1ab9 --- /dev/null +++ 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>} containing body and headers and + * returns a {@code Flux>}. + */ + public static class HttpRequestFunction implements Function>, Flux>> { + 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> apply(Flux> 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()))); + } + } +} diff --git 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 new file mode 100644 index 00000000..248d6860 --- /dev/null +++ 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; + } + +} diff --git 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 new file mode 100644 index 00000000..a3d57557 --- /dev/null +++ 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(); +} diff --git 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 new file mode 100644 index 00000000..8f13e6fa --- /dev/null +++ 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 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 { + } +} diff --git a/pom.xml b/pom.xml index c0ec7657..0381842c 100644 --- a/pom.xml +++ b/pom.xml @@ -61,6 +61,7 @@ function/filter-function function/header-enricher-function + function/http-request-function function/spel-function function/payload-converter-function function/splitter-function