diff --git a/applications/processor/http-request-processor/README.adoc b/applications/processor/http-request-processor/README.adoc index 8973a3b8..6687e745 100644 --- a/applications/processor/http-request-processor/README.adoc +++ b/applications/processor/http-request-processor/README.adoc @@ -43,17 +43,13 @@ 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: `$$$$`)* +$$http.request.body-expression$$:: $$A SpEL expression to derive the request body from the incoming message.$$ *($$Expression$$, default: `$$$$`)* +$$http.request.expected-response-type$$:: $$The type used to interpret the response.$$ *($$Class$$, default: `$$$$`)* +$$http.request.headers-expression$$:: $$A SpEL expression used to derive the http headers map to use.$$ *($$Expression$$, default: `$$$$`)* +$$http.request.http-method-expression$$:: $$A SpEL expression to derive the request method from the incoming message.$$ *($$Expression$$, default: `$$'GET'$$`)* +$$http.request.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.timeout$$:: $$Request timeout in milliseconds.$$ *($$Long$$, default: `$$30000$$`)* +$$http.request.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 index ffe29619..177c5483 100644 --- a/applications/processor/http-request-processor/pom.xml +++ b/applications/processor/http-request-processor/pom.xml @@ -60,15 +60,15 @@ http-request processor ${project.version} - org.springframework.cloud.stream.app.processor.http.request.HttpRequestProcessorConfiguration.class + org.springframework.cloud.fn.http.request.HttpRequestFunctionConfiguration.class - httpRequestProcessor + httpRequestFunction - org.springframework.cloud.stream.app - http-request-processor + org.springframework.cloud.fn + http-request-function ${project.version} 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 deleted file mode 100644 index d6705194..00000000 --- a/applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorConfiguration.java +++ /dev/null @@ -1,86 +0,0 @@ -/* - * 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 deleted file mode 100644 index d736a6b9..00000000 --- a/applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorProperties.java +++ /dev/null @@ -1,254 +0,0 @@ -/* - * 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 index 7c7fef7b..6e966678 100644 --- 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 @@ -1 +1,2 @@ -configuration-properties.classes=org.springframework.cloud.stream.app.processor.http.request.HttpRequestProcessorProperties +configuration-properties.classes=org.springframework.cloud.fn.http.request.HttpRequestFunctionProperties + diff --git a/applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties index 7c7fef7b..6e966678 100644 --- a/applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties +++ b/applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -1 +1,2 @@ -configuration-properties.classes=org.springframework.cloud.stream.app.processor.http.request.HttpRequestProcessorProperties +configuration-properties.classes=org.springframework.cloud.fn.http.request.HttpRequestFunctionProperties + 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 index 655ccdc0..08863392 100644 --- 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 @@ -32,9 +32,11 @@ import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.fn.http.request.HttpRequestFunctionConfiguration; 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.context.annotation.Import; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.messaging.Message; @@ -42,7 +44,6 @@ 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 { @@ -79,14 +80,14 @@ public class HttpRequestProcessorTests { } @Test - void requestUsingExpressions() throws IOException { + void requestUsingExpressions() { 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") + "http.request.reply-expression=#root", + "http.request.url-expression=headers['url']", + "http.request.http-method-expression=headers['method']", + "http.request.body-expression=headers['body']" + ) .run(context -> { Message message = MessageBuilder.withPayload("") .setHeader("url", url()) @@ -98,13 +99,11 @@ public class HttpRequestProcessorTests { ObjectMapper objectMapper = context.getBean(ObjectMapper.class); inputDestination.send(message); - Message reply = outputDestination.receive(100); + Message reply = outputDestination.receive(10000); // 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)) @@ -116,11 +115,11 @@ public class HttpRequestProcessorTests { 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") + "http.request.url-expression='" + url() + "'", + "http.request.http-method-expression='POST'", + "http.request.headers-expression={Accept:'application/octet-stream'}", + "http.request.expected-response-type=byte[]", + "spring.cloud.stream.bindings.httpRequestFunction-out-0.contentType=application/octet-stream") .run(context -> { Message message = MessageBuilder.withPayload("hello") .build(); @@ -128,7 +127,7 @@ public class HttpRequestProcessorTests { OutputDestination outputDestination = context.getBean(OutputDestination.class); inputDestination.send(message); - Message reply = outputDestination.receive(100); + Message reply = outputDestination.receive(10000); assertThat(new String(reply.getPayload())).isEqualTo(message.getPayload()); assertThat(reply.getHeaders().get(MessageHeaders.CONTENT_TYPE)) .isEqualTo(MediaType.APPLICATION_OCTET_STREAM); @@ -139,8 +138,8 @@ public class HttpRequestProcessorTests { void requestUsingJsonPathMethodExpression() throws IOException { applicationContextRunner .withPropertyValues( - "http.request.processor.url=" + url(), - "http.request.processor.httpMethodExpression=#jsonPath(payload,'$.myMethod')") + "http.request.url-expression='" + url() + "'", + "http.request.http-method-expression=#jsonPath(payload,'$.myMethod')") .run(context -> { Message message = MessageBuilder .withPayload("{\"name\":\"Fred\",\"age\":41, \"myMethod\":\"POST\"}") @@ -149,48 +148,13 @@ public class HttpRequestProcessorTests { OutputDestination outputDestination = context.getBean(OutputDestination.class); inputDestination.send(message); - Message reply = outputDestination.receive(100); + Message reply = outputDestination.receive(10000); 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 + @Import(HttpRequestFunctionConfiguration.class) static class HttpRequestProcessorApp { } } diff --git a/applications/sink/cassandra-sink/README.adoc b/applications/sink/cassandra-sink/README.adoc index 0906689b..0de9b5fb 100644 --- a/applications/sink/cassandra-sink/README.adoc +++ b/applications/sink/cassandra-sink/README.adoc @@ -16,17 +16,17 @@ The **$$cassandra$$** $$sink$$ has the following options: //tag::configuration-properties[] $$spring.data.cassandra.cluster-name$$:: $$$$ *($$String$$, default: `$$$$`)* $$spring.data.cassandra.compression$$:: $$Compression supported by the Cassandra binary protocol.$$ *($$Compression$$, default: `$$none$$`, possible values: `LZ4`,`SNAPPY`,`NONE`)* -$$spring.data.cassandra.connect-timeout$$:: $$$$ *($$Duration$$, default: `$$$$`)* -$$spring.data.cassandra.consistency-level$$:: $$$$ *($$DefaultConsistencyLevel$$, default: `$$$$`, possible values: `ANY`,`ONE`,`TWO`,`THREE`,`QUORUM`,`ALL`,`LOCAL_ONE`,`LOCAL_QUORUM`,`EACH_QUORUM`,`SERIAL`,`LOCAL_SERIAL`)* +$$spring.data.cassandra.connect-timeout$$:: $$Socket option: connection time out.$$ *($$Duration$$, default: `$$$$`)* +$$spring.data.cassandra.consistency-level$$:: $$Queries consistency level.$$ *($$DefaultConsistencyLevel$$, default: `$$$$`, possible values: `ANY`,`ONE`,`TWO`,`THREE`,`QUORUM`,`ALL`,`LOCAL_ONE`,`LOCAL_QUORUM`,`EACH_QUORUM`,`SERIAL`,`LOCAL_SERIAL`)* $$spring.data.cassandra.contact-points$$:: $$Cluster node addresses in the form 'host:port', or a simple 'host' to use the configured port.$$ *($$List$$, default: `$$[127.0.0.1:9042]$$`)* $$spring.data.cassandra.fetch-size$$:: $$$$ *($$Integer$$, default: `$$$$`)* $$spring.data.cassandra.keyspace-name$$:: $$Keyspace name to use.$$ *($$String$$, default: `$$$$`)* $$spring.data.cassandra.local-datacenter$$:: $$Datacenter that is considered "local". Contact points should be from this datacenter.$$ *($$String$$, default: `$$$$`)* +$$spring.data.cassandra.page-size$$:: $$Queries default page size.$$ *($$Integer$$, default: `$$5000$$`)* $$spring.data.cassandra.password$$:: $$Login password of the server.$$ *($$String$$, default: `$$$$`)* -$$spring.data.cassandra.port$$:: $$Port to use if a contact point does not specify one.$$ *($$Integer$$, default: `$$9042$$`)* -$$spring.data.cassandra.read-timeout$$:: $$$$ *($$Duration$$, default: `$$$$`)* +$$spring.data.cassandra.read-timeout$$:: $$Socket option: read time out.$$ *($$Duration$$, default: `$$$$`)* $$spring.data.cassandra.schema-action$$:: $$Schema action to take at startup.$$ *($$String$$, default: `$$none$$`)* -$$spring.data.cassandra.serial-consistency-level$$:: $$$$ *($$DefaultConsistencyLevel$$, default: `$$$$`, possible values: `ANY`,`ONE`,`TWO`,`THREE`,`QUORUM`,`ALL`,`LOCAL_ONE`,`LOCAL_QUORUM`,`EACH_QUORUM`,`SERIAL`,`LOCAL_SERIAL`)* +$$spring.data.cassandra.serial-consistency-level$$:: $$Queries serial consistency level.$$ *($$DefaultConsistencyLevel$$, default: `$$$$`, possible values: `ANY`,`ONE`,`TWO`,`THREE`,`QUORUM`,`ALL`,`LOCAL_ONE`,`LOCAL_QUORUM`,`EACH_QUORUM`,`SERIAL`,`LOCAL_SERIAL`)* $$spring.data.cassandra.session-name$$:: $$Name of the Cassandra session.$$ *($$String$$, default: `$$$$`)* $$spring.data.cassandra.ssl$$:: $$Enable SSL support.$$ *($$Boolean$$, default: `$$false$$`)* $$spring.data.cassandra.username$$:: $$Login user of the server.$$ *($$String$$, default: `$$$$`)* diff --git a/functions/function/http-request-function/README.adoc b/functions/function/http-request-function/README.adoc index efe44b2e..d0a0ad41 100644 --- a/functions/function/http-request-function/README.adoc +++ b/functions/function/http-request-function/README.adoc @@ -23,7 +23,7 @@ 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 +## Examples See this link:src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionApplicationTests.java[test suite] for examples of how this function is used. diff --git a/functions/function/http-request-function/pom.xml b/functions/function/http-request-function/pom.xml index 4b12e820..6da28f67 100644 --- a/functions/function/http-request-function/pom.xml +++ b/functions/function/http-request-function/pom.xml @@ -19,6 +19,11 @@ + + org.springframework.cloud.fn + config-common + ${project.version} + org.springframework.boot spring-boot-starter-validation 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 index ab7e1ab9..61f50a3d 100644 --- 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 @@ -17,6 +17,7 @@ package org.springframework.cloud.fn.http.request; import java.time.Duration; +import java.util.Map; import java.util.function.Function; import reactor.core.publisher.Flux; @@ -25,6 +26,8 @@ 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.HttpHeaders; +import org.springframework.http.HttpMethod; import org.springframework.http.ResponseEntity; import org.springframework.messaging.Message; import org.springframework.web.reactive.function.client.WebClient; @@ -50,7 +53,7 @@ public class HttpRequestFunctionConfiguration { } @Bean - public HttpRequestFunction httpRequestFunction(WebClient webClient, HttpRequestProperties properties) { + public HttpRequestFunction httpRequestFunction(WebClient webClient, HttpRequestFunctionProperties properties) { return new HttpRequestFunction(webClient, properties); } @@ -58,28 +61,57 @@ public class HttpRequestFunctionConfiguration { * Function that accepts a {@code Flux>} containing body and headers and * returns a {@code Flux>}. */ - public static class HttpRequestFunction implements Function>, Flux>> { + public static class HttpRequestFunction implements Function>, Flux> { private final WebClient webClient; private final UriBuilderFactory uriBuilderFactory = new DefaultUriBuilderFactory(); - private final HttpRequestProperties properties; + private final HttpRequestFunctionProperties properties; - public HttpRequestFunction(WebClient webClient, HttpRequestProperties properties) { + public HttpRequestFunction(WebClient webClient, HttpRequestFunctionProperties properties) { this.webClient = webClient; this.properties = properties; } @Override - public Flux> apply(Flux> messageFlux) { + 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())) + .method(resolveHttpMethod(message)) + .uri(uriBuilderFactory.uriString(resolveUrl(message)).build()) + .bodyValue(resolveBody(message)) + .headers(httpHeaders -> httpHeaders.addAll(resolveHeaders(message))) .retrieve() .toEntity(properties.getExpectedResponseType()) + .map(responseEntity -> properties.getReplyExpression().getValue(responseEntity)) .timeout(Duration.ofMillis(properties.getTimeout()))); } + + private String resolveUrl(Message message) { + return properties.getUrlExpression().getValue(message, String.class); + } + + private HttpMethod resolveHttpMethod(Message message) { + return properties.getHttpMethodExpression().getValue(message, HttpMethod.class); + } + + private Object resolveBody(Message message) { + return properties.getBodyExpression() != null ? properties.getBodyExpression().getValue(message) + : message.getPayload(); + } + + private HttpHeaders resolveHeaders(Message message) { + HttpHeaders headers = new HttpHeaders(); + if (properties.getHeadersExpression() != null) { + Map headersMap = properties.getHeadersExpression().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()); + } + } + } + return headers; + } + } } 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 index 248d6860..f180279f 100644 --- 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 @@ -16,12 +16,14 @@ 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.expression.Expression; import org.springframework.http.HttpMethod; +import org.springframework.http.ResponseEntity; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.expression.ValueExpression; import org.springframework.validation.annotation.Validated; /** @@ -35,25 +37,9 @@ import org.springframework.validation.annotation.Validated; */ @Validated @ConfigurationProperties("http.request") -public class HttpRequestFunctionProperties implements HttpRequestProperties { - private static final HttpMethod DEFAULT_HTTP_METHOD = HttpMethod.GET; +public class HttpRequestFunctionProperties { 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. @@ -66,33 +52,49 @@ public class HttpRequestFunctionProperties implements HttpRequestProperties { private long timeout = 30_000; /** - * A Map of HTTP request headers. + * A SpEL expression against incoming message to determine the URL to use. */ - private HttpHeaders headers = new HttpHeaders(); + private Expression urlExpression; - @NotEmpty - @Override - public String getUrl() { - return this.url; + /** + * A SpEL expression to derive the request method from the incoming message. + */ + private Expression httpMethodExpression = new ValueExpression(HttpMethod.GET); + + /** + * 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 FunctionExpression(ResponseEntity::getBody); + + @NotNull + public Expression getUrlExpression() { + return urlExpression; } - public void setUrl(String url) { - this.url = url; + public void setUrlExpression(Expression urlExpression) { + this.urlExpression = urlExpression; + } + + public Expression getHttpMethodExpression() { + return httpMethodExpression; + } + + public void setHttpMethodExpression(Expression httpMethodExpression) { + this.httpMethodExpression = httpMethodExpression; } @NotNull - @Override - public HttpMethod getHttpMethod() { - return httpMethod; - } - - public void setHttpMethod(HttpMethod httpMethod) { - this.httpMethod = httpMethod; - } - - - @NotNull - @Override public Class getExpectedResponseType() { return expectedResponseType; } @@ -101,16 +103,6 @@ public class HttpRequestFunctionProperties implements HttpRequestProperties { this.expectedResponseType = expectedResponseType; } - @Override - public Object getBody() { - return body; - } - - public void setBody(Object body) { - this.body = body; - } - - @Override public long getTimeout() { return this.timeout; } @@ -119,13 +111,29 @@ public class HttpRequestFunctionProperties implements HttpRequestProperties { this.timeout = timeout; } - @Override - public HttpHeaders getHeaders() { - return headers; + public Expression getBodyExpression() { + return bodyExpression; } - public void setHeaders(HttpHeaders headers) { - this.headers = headers; + 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; } } 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 deleted file mode 100644 index a3d57557..00000000 --- a/functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestProperties.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * 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/HttpRequestFunctionTestApplicationTests.java b/functions/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java index bd7a01dd..bde2f390 100644 --- a/functions/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java +++ b/functions/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * 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. @@ -25,7 +25,6 @@ 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; @@ -45,22 +44,17 @@ import org.springframework.messaging.support.MessageBuilder; import static org.assertj.core.api.Assertions.assertThat; public class HttpRequestFunctionTestApplicationTests { - private MockWebServer server; + private MockWebServer server = new MockWebServer(); private ApplicationContextRunner runner; @BeforeEach void setup() { - this.server = new MockWebServer(); this.runner = new ApplicationContextRunner() .withUserConfiguration(HttpRequestFunctionTestApplication.class) .withPropertyValues( - "http.request.url=" + url()); - } - - @AfterEach - void shutdown() throws IOException { - this.server.shutdown(); + "http.request.reply-expression=#root", + "http.request.url-expression='" + url() + "'"); } @Test @@ -70,11 +64,12 @@ public class HttpRequestFunctionTestApplicationTests { .setResponseCode(HttpStatus.OK.value()) .setBody("hello")); - runner.run(context -> { + runner.withPropertyValues("http.request.http-method-expression='POST'").run(context -> { HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); Message message = MessageBuilder.withPayload("").build(); StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) - .assertNext((ResponseEntity r) -> { + .assertNext(o -> { + ResponseEntity r = (ResponseEntity) o; assertThat(r.getBody()).isEqualTo("hello"); assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); }) @@ -96,22 +91,24 @@ public class HttpRequestFunctionTestApplicationTests { } }); - 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"); - }); + runner.withPropertyValues("http.request.http-method-expression='POST'", + "http.request.headers-expression={'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(o -> { + ResponseEntity r = (ResponseEntity) o; + 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 @@ -127,15 +124,15 @@ public class HttpRequestFunctionTestApplicationTests { } }); - runner.withPropertyValues("http.request.http-method=POST", - "http.request.headers[Content-Type]=application/json", + runner.withPropertyValues("http.request.http-method-expression='POST'", "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) -> { + .assertNext(o -> { + ResponseEntity r = (ResponseEntity) o; assertThat(r.getBody()).isEqualTo(json); assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); assertThat(r.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_JSON); @@ -149,7 +146,6 @@ public class HttpRequestFunctionTestApplicationTests { @Test void shouldDelete() { - server.setDispatcher(new Dispatcher() { @Override public MockResponse dispatch(RecordedRequest recordedRequest) { @@ -160,13 +156,14 @@ public class HttpRequestFunctionTestApplicationTests { } }); - runner.withPropertyValues("http.request.http-method=DELETE", + runner.withPropertyValues("http.request.http-method-expression='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) -> { + .assertNext(o -> { + ResponseEntity r = (ResponseEntity) o; assertThat(r.getBody()).isNull(); assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); }) @@ -188,6 +185,106 @@ public class HttpRequestFunctionTestApplicationTests { }); } + @Test + void requestUsingExpressions() { + server.setDispatcher(new Dispatcher() { + @Override + public MockResponse dispatch(RecordedRequest recordedRequest) { + return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE, + recordedRequest.getHeader(HttpHeaders.ACCEPT)) + .setBody(recordedRequest.getBody()) + .setResponseCode(HttpStatus.OK.value()); + } + }); + + runner.withPropertyValues( + "http.request.url-expression=headers['url']", + "http.request.http-method-expression=headers['method']", + "http.request.body-expression=headers['body']", + "http.request.headers-expression={Accept:'application/json'}") + .run(context -> { + Message message = MessageBuilder.withPayload("") + .setHeader("url", url()) + .setHeader("method", "POST") + .setHeader("body", "{\"hello\":\"world\"}") + .build(); + + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) + .assertNext(o -> { + ResponseEntity responseEntity = (ResponseEntity) o; + assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(responseEntity.getBody()).isEqualTo(message.getHeaders().get("body")); + assertThat(responseEntity.getHeaders().getContentType()) + .isEqualTo(MediaType.APPLICATION_JSON); + }) + .expectComplete() + .verify(); + }); + } + + @Test + void requestUsingReturnType() { + 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.OK.value()); + } + }); + runner.withPropertyValues( + "http.request..http-method-expression='POST'", + "http.request.headers-expression={'Content-Type':'application/octet-stream'}", + "http.request.expected-response-type=byte[]") + .run(context -> { + Message message = MessageBuilder.withPayload("hello") + .build(); + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) + .assertNext(o -> { + ResponseEntity responseEntity = (ResponseEntity) o; + assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(responseEntity.getBody()).isEqualTo("hello".getBytes()); + assertThat(responseEntity.getHeaders().getContentType()) + .isEqualTo(MediaType.APPLICATION_OCTET_STREAM); + }) + .expectComplete() + .verify(); + }); + } + + @Test + void requestUsingJsonPathMethodExpression() { + server.setDispatcher(new Dispatcher() { + @Override + public MockResponse dispatch(RecordedRequest recordedRequest) { + return new MockResponse() + .setBody(recordedRequest.getBody()) + .setHeader("method", recordedRequest.getMethod()) + .setResponseCode(HttpStatus.OK.value()); + } + }); + runner.withPropertyValues( + "http.request.http-method-expression=#jsonPath(payload,'$.myMethod')") + .run(context -> { + Message message = MessageBuilder + .withPayload("{\"name\":\"Fred\",\"age\":41, \"myMethod\":\"POST\"}") + .build(); + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) + .assertNext(o -> { + ResponseEntity responseEntity = (ResponseEntity) o; + assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(responseEntity.getBody()).isEqualTo(message.getPayload()); + }) + .expectComplete() + .verify(); + + }); + } + private String url() { return String.format("http://localhost:%d", server.getPort()); }