diff --git a/README.adoc b/README.adoc index 89303004..a50f9839 100644 --- a/README.adoc +++ b/README.adoc @@ -129,7 +129,7 @@ For e.g if you are only interested in jdbc-supplier and log-consumer, do the fol ==== Building core for Stream Applications -`./mvnw clean install -f applicaitons/stream-applications-core` +`./mvnw clean install -f applications/stream-applications-core` === Building the applications diff --git a/applications/processor/http-request-processor/README.adoc b/applications/processor/http-request-processor/README.adoc new file mode 100644 index 00000000..8973a3b8 --- /dev/null +++ b/applications/processor/http-request-processor/README.adoc @@ -0,0 +1,59 @@ +//tag::ref-doc[] += Http Request Processor + +A processor app that makes requests to an HTTP resource and emits the response body as a message payload. + +== Input +=== Headers +Any Required HTTP headers must be explicitly set via the `headers` or `headers-expression` property. See examples below. +Header values may also be used to construct: + + * the request body when referenced in the `body-expression` property. + * the HTTP method when referenced in the `http-method-expression` property. + * the URL when referenced in the `url-expression` property. + +=== Payload + +The payload is used as the request body for a POST request by default, and can be any Java type. +It should be an empty String for a GET request. +The payload may also be used to construct: + +* the request body when referenced in the `body-expression` property. +* the HTTP method when referenced in the `http-method-expression` property. +* the URL when referenced in the `url-expression` property. + +The underlying https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/web/reactive/function/client/WebClient.html[WebClient] supports Jackson JSON serialization to support any request and response types if necessary. +The `expected-response-type` property, `String.class` by default, may be set to any class in your application class path. +Note that user defined payload types will require adding required dependencies to your pom file. + +== Output + +=== Headers +No HTTP message headers are mapped to the outbound Message. + +=== Payload +The raw output object is https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/http/ResponseEntity.html[ResponseEntity] any of its fields (e.g., `body`, `headers`) or accessor methods (`statusCode`) may be referenced as part of the `reply-expression`. +By default the outbound Message payload is the response body. +Note that ResponseEntity (referenced by the expression `#root`) cannot be deserialized by Jackson by default, but may be rendered as a `HashMap`. + +== Options + +The **$$http-request$$** $$processor$$ has the following options: + +== Options + +//tag::configuration-properties[] +$$http.request.processor.body$$:: $$The (static) request body; if neither this nor bodyExpression is provided, the payload will be used.$$ *($$Object$$, default: `$$$$`)* +$$http.request.processor.body-expression$$:: $$A SpEL expression to derive the request body from the incoming message.$$ *($$Expression$$, default: `$$$$`)* +$$http.request.processor.expected-response-type$$:: $$The type used to interpret the response.$$ *($$Class$$, default: `$$$$`)* +$$http.request.processor.headers$$:: $$A Map of HTTP request headers.$$ *($$HttpHeaders$$, default: `$$$$`)* +$$http.request.processor.headers-expression$$:: $$A SpEL expression used to derive the http headers map to use.$$ *($$Expression$$, default: `$$$$`)* +$$http.request.processor.http-method$$:: $$The kind of http method to use.$$ *($$HttpMethod$$, default: `$$$$`, possible values: `GET`,`HEAD`,`POST`,`PUT`,`PATCH`,`DELETE`,`OPTIONS`,`TRACE`)* +$$http.request.processor.http-method-expression$$:: $$A SpEL expression to derive the request method from the incoming message.$$ *($$Expression$$, default: `$$$$`)* +$$http.request.processor.reply-expression$$:: $$A SpEL expression used to compute the final result, applied against the whole http {@link org.springframework.http.ResponseEntity}.$$ *($$Expression$$, default: `$$body$$`)* +$$http.request.processor.timeout$$:: $$Request timeout in milliseconds.$$ *($$Long$$, default: `$$30000$$`)* +$$http.request.processor.url$$:: $$The URL to issue an http request to, as a static value.$$ *($$String$$, default: `$$$$`)* +$$http.request.processor.url-expression$$:: $$A SpEL expression against incoming message to determine the URL to use.$$ *($$Expression$$, default: `$$$$`)* +//end::configuration-properties[] + +//end::ref-doc[] diff --git a/applications/processor/http-request-processor/pom.xml b/applications/processor/http-request-processor/pom.xml new file mode 100644 index 00000000..ea1751a3 --- /dev/null +++ b/applications/processor/http-request-processor/pom.xml @@ -0,0 +1,104 @@ + + + 4.0.0 + http-request-processor + 3.0.0-SNAPSHOT + http-request-processor + HTTP request processor apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.boot + spring-boot-configuration-processor + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + org.springframework.cloud.fn + http-request-function + ${java-functions.version} + + + com.squareup.okhttp3 + mockwebserver + test + + + io.projectreactor + reactor-test + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + http-request + processor + ${project.version} + org.springframework.cloud.stream.app.processor.http.request.HttpRequestProcessorConfiguration.class + + httpRequestProcessor + + + + + org.springframework.cloud.stream.app + http-request-processor + ${project.version} + + + + + true + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorConfiguration.java b/applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorConfiguration.java new file mode 100644 index 00000000..d6705194 --- /dev/null +++ b/applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorConfiguration.java @@ -0,0 +1,86 @@ +/* + * 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.stream.app.processor.http.request; + +import java.util.function.Function; + +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.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.messaging.Message; +import org.springframework.web.reactive.function.client.WebClient; + +import static org.springframework.cloud.fn.http.request.HttpRequestFunctionConfiguration.HttpRequestFunction; + +@Configuration +@EnableConfigurationProperties({ HttpRequestProcessorProperties.class }) +public class HttpRequestProcessorConfiguration { + + private static Log log = LogFactory.getLog(HttpRequestProcessorConfiguration.class); + + @Bean + @ConditionalOnMissingBean(WebClient.class) + public WebClient webClient() { + return WebClient.builder() + .build(); + } + + @Bean + HttpRequestFunctionFactory httpRequestFunctionFactory(WebClient webClient, + HttpRequestProcessorProperties properties) { + return new HttpRequestFunctionFactory(webClient, properties); + } + + @Bean + Function, ?> httpRequestProcessor(HttpRequestFunctionFactory httpRequestFunctionFactory, + HttpRequestProcessorProperties properties) { + + return message -> Mono.from(httpRequestFunctionFactory.getHttpRequestFunction(message).apply(Flux.just(message)) + .map(responseEntity -> properties.getReplyExpression().getValue(responseEntity))) + .doOnError(e -> log.error(e.getMessage(), e)) + .block(); + } + + static class HttpRequestFunctionFactory { + + private final WebClient webClient; + + private final HttpRequestProcessorProperties properties; + + private final HttpRequestFunction instance; + + HttpRequestFunctionFactory(WebClient webClient, HttpRequestProcessorProperties properties) { + this.properties = properties; + this.webClient = webClient; + this.instance = properties.usesRequestExpressions() ? null + : new HttpRequestFunction(webClient, properties); + } + + HttpRequestFunction getHttpRequestFunction(Message message) { + if (instance != null) { + return instance; + } + return new HttpRequestFunction(webClient, properties.evaluateFunctionProperties(message)); + } + } +} diff --git a/applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorProperties.java b/applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorProperties.java new file mode 100644 index 00000000..d736a6b9 --- /dev/null +++ b/applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorProperties.java @@ -0,0 +1,254 @@ +/* + * 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.stream.app.processor.http.request; + +import java.util.Map; + +import javax.validation.constraints.AssertTrue; +import javax.validation.constraints.NotNull; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.fn.http.request.HttpRequestFunctionProperties; +import org.springframework.cloud.fn.http.request.HttpRequestProperties; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.messaging.Message; +import org.springframework.validation.annotation.Validated; + +/** + * Configuration properties for the Http Request Processor module. + * + * @author Waldemar Hummer + * @author Mark Fisher + * @author Christian Tzolov + * @author Artem Bilan + * @author David Turanski + */ + +@Validated +@ConfigurationProperties("http.request.processor") +public class HttpRequestProcessorProperties 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(); + + /** + * A SpEL expression against incoming message to determine the URL to use. + */ + private Expression urlExpression; + + /** + * A SpEL expression to derive the request method from the incoming message. + */ + private Expression httpMethodExpression; + + /** + * A SpEL expression to derive the request body from the incoming message. + */ + private Expression bodyExpression; + + /** + * A SpEL expression used to derive the http headers map to use. + */ + private Expression headersExpression; + + /** + * A SpEL expression used to compute the final result, applied against the whole http + * {@link org.springframework.http.ResponseEntity}. + */ + private Expression replyExpression = new SpelExpressionParser().parseExpression("body"); + + public Expression getUrlExpression() { + return urlExpression; + } + + public void setUrlExpression(Expression urlExpression) { + this.urlExpression = urlExpression; + } + + public Expression getHttpMethodExpression() { + return httpMethodExpression; + } + + public void setHttpMethodExpression(Expression httpMethodExpression) { + this.setHttpMethod(null); + this.httpMethodExpression = httpMethodExpression; + } + + @Override + public String getUrl() { + return this.url; + } + + public void setUrl(String url) { + this.url = url; + } + + @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; + } + + public Expression getBodyExpression() { + return bodyExpression; + } + + public void setBodyExpression(Expression bodyExpression) { + this.bodyExpression = bodyExpression; + } + + public Expression getHeadersExpression() { + return headersExpression; + } + + public void setHeadersExpression(Expression headersExpression) { + this.headersExpression = headersExpression; + } + + @NotNull + public Expression getReplyExpression() { + return replyExpression; + } + + public void setReplyExpression(Expression replyExpression) { + this.replyExpression = replyExpression; + } + + @AssertTrue(message = "Exactly one of 'url' or 'urlExpression' is required") + public boolean isExactlyOneUrl() { + return getUrl() == null ^ urlExpression == null; + } + + @AssertTrue(message = "At most one of 'body' or 'bodyExpression' is allowed") + public boolean isAtMostOneBody() { + return getBody() == null || bodyExpression == null; + } + + @AssertTrue(message = "At most one of 'httpMethod' or 'httpMethodExpression' is allowed") + public boolean isAtMostOneHttpMethod() { + return getHttpMethod() == null || httpMethodExpression == null; + } + + public boolean usesRequestExpressions() { + return headersExpression != null || + bodyExpression != null || + httpMethodExpression != null || + urlExpression != null; + } + + HttpRequestFunctionProperties evaluateFunctionProperties(Message message) { + HttpRequestFunctionProperties properties = new HttpRequestFunctionProperties(); + properties.setUrl(urlExpression != null ? urlExpression.getValue(message, String.class) : getUrl()); + properties.setBody(bodyExpression != null ? bodyExpression.getValue(message) : getBody()); + properties.setHttpMethod(httpMethodExpression != null ? httpMethodExpression.getValue(message, HttpMethod.class) + : getHttpMethod()); + + HttpHeaders headers = new HttpHeaders(); + headers.addAll(getHeaders()); + if (headersExpression != null) { + Map headersMap = headersExpression.getValue(message, Map.class); + for (Map.Entry header : headersMap.entrySet()) { + if (header.getKey() != null && header.getValue() != null) { + headers.add(header.getKey().toString(), + header.getValue().toString()); + } + } + } + properties.setHeaders(headers); + + properties.setTimeout(getTimeout()); + + properties.setExpectedResponseType(getExpectedResponseType()); + + return properties; + } +} diff --git a/applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..7c7fef7b --- /dev/null +++ b/applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1 @@ +configuration-properties.classes=org.springframework.cloud.stream.app.processor.http.request.HttpRequestProcessorProperties diff --git a/applications/processor/http-request-processor/src/test/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorTests.java b/applications/processor/http-request-processor/src/test/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorTests.java new file mode 100644 index 00000000..655ccdc0 --- /dev/null +++ b/applications/processor/http-request-processor/src/test/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorTests.java @@ -0,0 +1,196 @@ +/* + * 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.stream.app.processor.http.request; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.mockwebserver.Dispatcher; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; + +public class HttpRequestProcessorTests { + + private static MockWebServer server; + + private ApplicationContextRunner applicationContextRunner; + + @BeforeEach + void setup() { + applicationContextRunner = new ApplicationContextRunner().withUserConfiguration( + TestChannelBinderConfiguration.getCompleteConfiguration(HttpRequestProcessorApp.class)); + } + + @BeforeAll + static void startServer() { + server = new MockWebServer(); + server.setDispatcher(new Dispatcher() { + @Override + public MockResponse dispatch(RecordedRequest recordedRequest) { + return new MockResponse() + .setBody(recordedRequest.getBody()) + .setResponseCode(HttpStatus.OK.value()); + } + }); + } + + @AfterAll + static void shutdownServer() throws IOException { + server.shutdown(); + } + + private String url() { + return String.format("http://localhost:%d", server.getPort()); + } + + @Test + void requestUsingExpressions() throws IOException { + applicationContextRunner + .withPropertyValues( + "http.request.processor.url-expression=headers['url']", + "http.request.processor.http-method-expression=headers['method']", + "http.request.processor.body-expression=headers['body']", + "http.request.processor.headers-expression={Accept:'application/json'}", + "http.request.processor.reply-expression=#root") + .run(context -> { + Message message = MessageBuilder.withPayload("") + .setHeader("url", url()) + .setHeader("method", "POST") + .setHeader("body", "{\"hello\":\"world\"}") + .build(); + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + ObjectMapper objectMapper = context.getBean(ObjectMapper.class); + + inputDestination.send(message); + Message reply = outputDestination.receive(100); + + // Cannot deserialize ResponseEntity directly. + Map responseEntityAsMap = objectMapper.readValue(reply.getPayload(), HashMap.class); + + System.out.println(responseEntityAsMap); + + assertThat(responseEntityAsMap.get("statusCode")).isEqualTo("OK"); + assertThat(responseEntityAsMap.get("body")).isEqualTo(message.getHeaders().get("body")); + assertThat(reply.getHeaders().get(MessageHeaders.CONTENT_TYPE)) + .isEqualTo(MediaType.APPLICATION_JSON); + }); + } + + @Test + void requestUsingReturnType() throws IOException { + applicationContextRunner + .withPropertyValues( + "http.request.processor.url=" + url(), + "http.request.processor.httpMethod=POST", + "http.request.processor.headers[Accept]=application/octet-stream", + "http.request.processor.expectedResponseType=byte[]", + "spring.cloud.stream.bindings.httpRequestProcessor-out-0.contentType=application/octet-stream") + .run(context -> { + Message message = MessageBuilder.withPayload("hello") + .build(); + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + inputDestination.send(message); + Message reply = outputDestination.receive(100); + assertThat(new String(reply.getPayload())).isEqualTo(message.getPayload()); + assertThat(reply.getHeaders().get(MessageHeaders.CONTENT_TYPE)) + .isEqualTo(MediaType.APPLICATION_OCTET_STREAM); + }); + } + + @Test + void requestUsingJsonPathMethodExpression() throws IOException { + applicationContextRunner + .withPropertyValues( + "http.request.processor.url=" + url(), + "http.request.processor.httpMethodExpression=#jsonPath(payload,'$.myMethod')") + .run(context -> { + Message message = MessageBuilder + .withPayload("{\"name\":\"Fred\",\"age\":41, \"myMethod\":\"POST\"}") + .build(); + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + inputDestination.send(message); + Message reply = outputDestination.receive(100); + assertThat(new String(reply.getPayload())).isEqualTo(message.getPayload()); + }); + } + + @Test + void cannotSpecifyBothUrlandUrlExpression() { + applicationContextRunner + .withPropertyValues("http.request.processor.url=http://example.com", + "http.request.processor.url-expression=headers['url']") + .run(context -> { + assertThatIllegalStateException().isThrownBy(() -> { + context.start(); + }); + }); + } + + @Test + void cannotSpecifyBothHttpMethosdandHttpMethodExpression() { + applicationContextRunner + .withPropertyValues("http.request.processor.http-method=POST", + "http.request.processor.http-method-expression=headers['method']") + .run(context -> { + assertThatIllegalStateException().isThrownBy(() -> { + context.start(); + }); + }); + } + + @Test + void cannotSpecifyBothHeadersAndHeadersExpression() { + applicationContextRunner + .withPropertyValues("http.request.processor.headers[Content-Type]=application/json", + "http.request.processor.headers-expression={'Content-Type': headers['content']}") + .run(context -> { + assertThatIllegalStateException().isThrownBy(() -> { + context.start(); + }); + }); + } + + @SpringBootApplication + static class HttpRequestProcessorApp { + } +} diff --git a/applications/processor/http-request-processor/src/test/resources/META-INF/spring.binders b/applications/processor/http-request-processor/src/test/resources/META-INF/spring.binders new file mode 100644 index 00000000..9fd72cf8 --- /dev/null +++ b/applications/processor/http-request-processor/src/test/resources/META-INF/spring.binders @@ -0,0 +1,2 @@ +integration:\ +org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration diff --git a/applications/processor/pom.xml b/applications/processor/pom.xml index c42c4dda..007cdfdb 100644 --- a/applications/processor/pom.xml +++ b/applications/processor/pom.xml @@ -11,10 +11,11 @@ bridge-processor + filter-processor groovy-processor header-enricher-processor + http-request-processor splitter-processor - filter-processor transform-processor script-processor diff --git a/functions/function/http-request-function/README.adoc b/functions/function/http-request-function/README.adoc new file mode 100644 index 00000000..efe44b2e --- /dev/null +++ b/functions/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/functions/function/http-request-function/pom.xml b/functions/function/http-request-function/pom.xml new file mode 100644 index 00000000..4b12e820 --- /dev/null +++ b/functions/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/functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java b/functions/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/functions/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/functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionProperties.java b/functions/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/functions/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/functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestProperties.java b/functions/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/functions/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/functions/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionApplicationTests.java b/functions/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/functions/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/functions/pom.xml b/functions/pom.xml index c0ec7657..0381842c 100644 --- a/functions/pom.xml +++ b/functions/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