SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor to use SpEL in function properties

Browse Source 
* Clean up per review
This commit is contained in:
David Turanski
2020-08-12 15:39:56 -04:00
committed by GitHub
parent 23aabb1d63
commit eee707da00
14 changed files with 279 additions and 544 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
applications/processor/http-request-processor/README.adoc
View File

@@ -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: `$$<none>$$`)*
$$http.request.processor.body-expression$$:: $$A SpEL expression to derive the request body from the incoming message.$$ *($$Expression$$, default: `$$<none>$$`)*
$$http.request.processor.expected-response-type$$:: $$The type used to interpret the response.$$ *($$Class<?>$$, default: `$$<none>$$`)*
$$http.request.processor.headers$$:: $$A Map of HTTP request headers.$$ *($$HttpHeaders$$, default: `$$<none>$$`)*
$$http.request.processor.headers-expression$$:: $$A SpEL expression used to derive the http headers map to use.$$ *($$Expression$$, default: `$$<none>$$`)*
$$http.request.processor.http-method$$:: $$The kind of http method to use.$$ *($$HttpMethod$$, default: `$$<none>$$`, 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: `$$<none>$$`)*
$$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: `$$<none>$$`)*
$$http.request.processor.url-expression$$:: $$A SpEL expression against incoming message to determine the URL to use.$$ *($$Expression$$, default: `$$<none>$$`)*
$$http.request.body-expression$$:: $$A SpEL expression to derive the request body from the incoming message.$$ *($$Expression$$, default: `$$<none>$$`)*
$$http.request.expected-response-type$$:: $$The type used to interpret the response.$$ *($$Class<?>$$, default: `$$<none>$$`)*
$$http.request.headers-expression$$:: $$A SpEL expression used to derive the http headers map to use.$$ *($$Expression$$, default: `$$<none>$$`)*
$$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: `$$<none>$$`)*
//end::configuration-properties[]
//end::ref-doc[]

8
applications/processor/http-request-processor/pom.xml
View File

@@ -60,15 +60,15 @@
<name>http-request</name>
<type>processor</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.stream.app.processor.http.request.HttpRequestProcessorConfiguration.class
<configClass>org.springframework.cloud.fn.http.request.HttpRequestFunctionConfiguration.class
</configClass>
<functionDefinition>httpRequestProcessor</functionDefinition>
<functionDefinition>httpRequestFunction</functionDefinition>
</generatedApp>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>http-request-processor</artifactId>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>http-request-function</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>

86
applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorConfiguration.java
View File

@@ -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<Message<?>, ?> 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));
}
}
}

254
applications/processor/http-request-processor/src/main/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorProperties.java
View File

@@ -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;
}
}

3
applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties
View File

@@ -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

3
applications/processor/http-request-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties
View File

@@ -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

74
applications/processor/http-request-processor/src/test/java/org/springframework/cloud/stream/app/processor/http/request/HttpRequestProcessorTests.java
View File

@@ -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<byte[]> reply = outputDestination.receive(100);
Message<byte[]> 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<byte[]> reply = outputDestination.receive(100);
Message<byte[]> 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<byte[]> reply = outputDestination.receive(100);
Message<byte[]> 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 {
}
}

10
applications/sink/cassandra-sink/README.adoc
View File

@@ -16,17 +16,17 @@ The **$$cassandra$$** $$sink$$ has the following options:
//tag::configuration-properties[]
$$spring.data.cassandra.cluster-name$$:: $$<documentation missing>$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.cassandra.compression$$:: $$Compression supported by the Cassandra binary protocol.$$ *($$Compression$$, default: `$$none$$`, possible values: `LZ4`,`SNAPPY`,`NONE`)*
$$spring.data.cassandra.connect-timeout$$:: $$<documentation missing>$$ *($$Duration$$, default: `$$<none>$$`)*
$$spring.data.cassandra.consistency-level$$:: $$<documentation missing>$$ *($$DefaultConsistencyLevel$$, default: `$$<none>$$`, 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: `$$<none>$$`)*
$$spring.data.cassandra.consistency-level$$:: $$Queries consistency level.$$ *($$DefaultConsistencyLevel$$, default: `$$<none>$$`, 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<String>$$, default: `$$[127.0.0.1:9042]$$`)*
$$spring.data.cassandra.fetch-size$$:: $$<documentation missing>$$ *($$Integer$$, default: `$$<none>$$`)*
$$spring.data.cassandra.keyspace-name$$:: $$Keyspace name to use.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.cassandra.local-datacenter$$:: $$Datacenter that is considered "local". Contact points should be from this datacenter.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.cassandra.page-size$$:: $$Queries default page size.$$ *($$Integer$$, default: `$$5000$$`)*
$$spring.data.cassandra.password$$:: $$Login password of the server.$$ *($$String$$, default: `$$<none>$$`)*
$$spring.data.cassandra.port$$:: $$Port to use if a contact point does not specify one.$$ *($$Integer$$, default: `$$9042$$`)*
$$spring.data.cassandra.read-timeout$$:: $$<documentation missing>$$ *($$Duration$$, default: `$$<none>$$`)*
$$spring.data.cassandra.read-timeout$$:: $$Socket option: read time out.$$ *($$Duration$$, default: `$$<none>$$`)*
$$spring.data.cassandra.schema-action$$:: $$Schema action to take at startup.$$ *($$String$$, default: `$$none$$`)*
$$spring.data.cassandra.serial-consistency-level$$:: $$<documentation missing>$$ *($$DefaultConsistencyLevel$$, default: `$$<none>$$`, 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: `$$<none>$$`, 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: `$$<none>$$`)*
$$spring.data.cassandra.ssl$$:: $$Enable SSL support.$$ *($$Boolean$$, default: `$$false$$`)*
$$spring.data.cassandra.username$$:: $$Login user of the server.$$ *($$String$$, default: `$$<none>$$`)*

2
functions/function/http-request-function/README.adoc
View File

@@ -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.

5
functions/function/http-request-function/pom.xml
View File

@@ -19,6 +19,11 @@
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>config-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>

50
functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java
View File

@@ -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<Message<?>>} containing body and headers and
* returns a {@code Flux<ResponseEntity<?>>}.
*/
public static class HttpRequestFunction implements Function<Flux<Message<?>>, Flux<ResponseEntity<?>>> {
public static class HttpRequestFunction implements Function<Flux<Message<?>>, 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<ResponseEntity<?>> apply(Flux<Message<?>> messageFlux) {
public Flux<?> apply(Flux<Message<?>> messageFlux) {
return messageFlux.flatMap(message -> this.webClient
.method(properties.getHttpMethod())
.uri(uriBuilderFactory.uriString(properties.getUrl()).build())
.bodyValue(properties.getBody() == null ? message.getPayload() : properties.getBody())
.headers(httpHeaders -> httpHeaders.addAll(properties.getHeaders()))
.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;
}
}
}

116
functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionProperties.java
View File

@@ -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>(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;
}
}

29
functions/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestProperties.java
View File

@@ -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();
}

165
functions/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java
View File

@@ -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<String, String> 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());
}