Skip to content

Commit af200a5

Browse files
spencergibbspencergibb
authored
Adds support for spring-cloud-function and spring-cloud-stream handlers (#3646)
* Add support for spring cloud function handler. * Add support for spring cloud stream handler. * Add support for path variable expansion in fn: and stream: URIs * Adds support for config based fn/stream HandlerFunctions * Starting support for webflux function support * Adds fn:functionName support for webflux server * Adds spring-cloud-starter-stream-rabbit to test scope * Adds StreamRoutingFilter to webflux server
1 parent ab5e61d commit af200a5

File tree

20 files changed

+1512
-3
lines changed

20 files changed

+1512
-3
lines changed

‎pom.xml

+16
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,8 @@
5757
<junit-pioneer.version>2.3.0</junit-pioneer.version>
5858
<spring-cloud-circuitbreaker.version>3.2.1-SNAPSHOT</spring-cloud-circuitbreaker.version>
5959
<spring-cloud-commons.version>4.3.0-SNAPSHOT</spring-cloud-commons.version>
60+
<spring-cloud-function.version>4.2.1-SNAPSHOT</spring-cloud-function.version>
61+
<spring-cloud-stream.version>4.2.1-SNAPSHOT</spring-cloud-stream.version>
6062
</properties>
6163

6264
<dependencyManagement>
@@ -75,6 +77,20 @@
7577
<type>pom</type>
7678
<scope>import</scope>
7779
</dependency>
80+
<dependency>
81+
<groupId>org.springframework.cloud</groupId>
82+
<artifactId>spring-cloud-function-dependencies</artifactId>
83+
<version>${spring-cloud-function.version}</version>
84+
<type>pom</type>
85+
<scope>import</scope>
86+
</dependency>
87+
<dependency>
88+
<groupId>org.springframework.cloud</groupId>
89+
<artifactId>spring-cloud-stream-dependencies</artifactId>
90+
<version>${spring-cloud-stream.version}</version>
91+
<type>pom</type>
92+
<scope>import</scope>
93+
</dependency>
7894
<dependency>
7995
<groupId>org.springframework.cloud</groupId>
8096
<artifactId>spring-cloud-test-support</artifactId>

‎spring-cloud-gateway-server-mvc/pom.xml

+30
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,16 @@
5252
<groupId>org.springframework.cloud</groupId>
5353
<artifactId>spring-cloud-commons</artifactId>
5454
</dependency>
55+
<dependency>
56+
<groupId>org.springframework.cloud</groupId>
57+
<artifactId>spring-cloud-function-context</artifactId>
58+
<optional>true</optional>
59+
</dependency>
60+
<dependency>
61+
<groupId>org.springframework.cloud</groupId>
62+
<artifactId>spring-cloud-stream</artifactId>
63+
<optional>true</optional>
64+
</dependency>
5565
<dependency>
5666
<groupId>org.springframework.cloud</groupId>
5767
<artifactId>spring-cloud-loadbalancer</artifactId>
@@ -89,7 +99,22 @@
8999
<artifactId>spring-boot-properties-migrator</artifactId>
90100
<scope>test</scope>
91101
</dependency>
102+
<dependency>
103+
<groupId>org.springframework.boot</groupId>
104+
<artifactId>spring-boot-testcontainers</artifactId>
105+
<scope>test</scope>
106+
</dependency>
107+
<dependency>
108+
<groupId>org.springframework.cloud</groupId>
109+
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
110+
<scope>test</scope>
111+
</dependency>
92112
<!-- Third party test dependencies -->
113+
<dependency>
114+
<groupId>org.awaitility</groupId>
115+
<artifactId>awaitility</artifactId>
116+
<scope>test</scope>
117+
</dependency>
93118
<dependency>
94119
<groupId>com.bucket4j</groupId>
95120
<artifactId>bucket4j_jdk17-caffeine</artifactId>
@@ -105,5 +130,10 @@
105130
<artifactId>junit-jupiter</artifactId>
106131
<scope>test</scope>
107132
</dependency>
133+
<dependency>
134+
<groupId>org.testcontainers</groupId>
135+
<artifactId>rabbitmq</artifactId>
136+
<scope>test</scope>
137+
</dependency>
108138
</dependencies>
109139
</project>

‎spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerHeaderUtils.java

+116
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
/*
2+
* Copyright 2012-2019 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.cloud.gateway.server.mvc.handler;
18+
19+
import java.util.Arrays;
20+
import java.util.Collection;
21+
import java.util.Collections;
22+
import java.util.LinkedHashMap;
23+
import java.util.List;
24+
import java.util.Locale;
25+
import java.util.Map;
26+
27+
import org.springframework.http.HttpHeaders;
28+
import org.springframework.messaging.MessageHeaders;
29+
30+
/**
31+
* @author Dave Syer
32+
* @author Oleg Zhurakousky
33+
*/
34+
public final class FunctionHandlerHeaderUtils {
35+
36+
/**
37+
* Message Header name which contains HTTP request parameters.
38+
*/
39+
public static final String HTTP_REQUEST_PARAM = "http_request_param";
40+
41+
private static HttpHeaders IGNORED = new HttpHeaders();
42+
43+
private static HttpHeaders REQUEST_ONLY = new HttpHeaders();
44+
45+
static {
46+
IGNORED.add(MessageHeaders.ID, "");
47+
IGNORED.add(HttpHeaders.CONTENT_LENGTH, "0");
48+
// Headers that would typically be added by a downstream client
49+
REQUEST_ONLY.add(HttpHeaders.ACCEPT, "");
50+
REQUEST_ONLY.add(HttpHeaders.CONTENT_LENGTH, "");
51+
REQUEST_ONLY.add(HttpHeaders.CONTENT_TYPE, "");
52+
REQUEST_ONLY.add(HttpHeaders.HOST, "");
53+
}
54+
55+
private FunctionHandlerHeaderUtils() {
56+
throw new IllegalStateException("Can't instantiate a utility class");
57+
}
58+
59+
public static HttpHeaders fromMessage(MessageHeaders headers, List<String> ignoredHeders) {
60+
HttpHeaders result = new HttpHeaders();
61+
for (String name : headers.keySet()) {
62+
Object value = headers.get(name);
63+
name = name.toLowerCase(Locale.ROOT);
64+
if (!IGNORED.containsKey(name) && !ignoredHeders.contains(name)) {
65+
Collection<?> values = multi(value);
66+
for (Object object : values) {
67+
result.set(name, object.toString());
68+
}
69+
}
70+
}
71+
return result;
72+
}
73+
74+
@SuppressWarnings("unchecked")
75+
public static HttpHeaders fromMessage(MessageHeaders headers) {
76+
return fromMessage(headers, Collections.EMPTY_LIST);
77+
}
78+
79+
public static HttpHeaders sanitize(HttpHeaders request, List<String> ignoredHeders,
80+
List<String> requestOnlyHeaders) {
81+
HttpHeaders result = new HttpHeaders();
82+
for (String name : request.keySet()) {
83+
List<String> value = request.get(name);
84+
name = name.toLowerCase(Locale.ROOT);
85+
if (!IGNORED.containsKey(name) && !REQUEST_ONLY.containsKey(name) && !ignoredHeders.contains(name)
86+
&& !requestOnlyHeaders.contains(name)) {
87+
result.put(name, value);
88+
}
89+
}
90+
return result;
91+
}
92+
93+
@SuppressWarnings("unchecked")
94+
public static HttpHeaders sanitize(HttpHeaders request) {
95+
return sanitize(request, Collections.EMPTY_LIST, Collections.EMPTY_LIST);
96+
}
97+
98+
public static MessageHeaders fromHttp(HttpHeaders headers) {
99+
Map<String, Object> map = new LinkedHashMap<>();
100+
for (String name : headers.keySet()) {
101+
Collection<?> values = multi(headers.get(name));
102+
name = name.toLowerCase(Locale.ROOT);
103+
Object value = values == null ? null : (values.size() == 1 ? values.iterator().next() : values);
104+
if (name.toLowerCase(Locale.ROOT).equals(HttpHeaders.CONTENT_TYPE.toLowerCase(Locale.ROOT))) {
105+
name = MessageHeaders.CONTENT_TYPE;
106+
}
107+
map.put(name, value);
108+
}
109+
return new MessageHeaders(map);
110+
}
111+
112+
private static Collection<?> multi(Object value) {
113+
return value instanceof Collection ? (Collection<?>) value : Arrays.asList(value);
114+
}
115+
116+
}

‎spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerRequestProcessingHelper.java

+130
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/*
2+
* Copyright 2019-2021 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.cloud.gateway.server.mvc.handler;
18+
19+
import java.util.List;
20+
import java.util.stream.Collectors;
21+
import java.util.stream.StreamSupport;
22+
23+
import org.apache.commons.logging.Log;
24+
import org.apache.commons.logging.LogFactory;
25+
26+
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
27+
import org.springframework.http.HttpHeaders;
28+
import org.springframework.http.HttpMethod;
29+
import org.springframework.messaging.Message;
30+
import org.springframework.messaging.support.MessageBuilder;
31+
import org.springframework.util.CollectionUtils;
32+
import org.springframework.web.servlet.function.ServerRequest;
33+
import org.springframework.web.servlet.function.ServerResponse;
34+
import org.springframework.web.servlet.function.ServerResponse.BodyBuilder;
35+
36+
import static org.springframework.cloud.gateway.server.mvc.handler.FunctionHandlerHeaderUtils.fromMessage;
37+
import static org.springframework.cloud.gateway.server.mvc.handler.FunctionHandlerHeaderUtils.sanitize;
38+
39+
/**
40+
* !INTERNAL USE ONLY!
41+
*
42+
* @author Oleg Zhurakousky
43+
*
44+
*/
45+
final class FunctionHandlerRequestProcessingHelper {
46+
47+
private static Log logger = LogFactory.getLog(FunctionHandlerRequestProcessingHelper.class);
48+
49+
private FunctionHandlerRequestProcessingHelper() {
50+
51+
}
52+
53+
@SuppressWarnings({ "rawtypes", "unchecked" })
54+
static ServerResponse processRequest(ServerRequest request, FunctionInvocationWrapper function, Object argument,
55+
boolean eventStream, List<String> ignoredHeaders, List<String> requestOnlyHeaders) {
56+
if (argument == null) {
57+
argument = "";
58+
}
59+
60+
if (function == null) {
61+
return ServerResponse.notFound().build();
62+
}
63+
64+
HttpHeaders headers = request.headers().asHttpHeaders();
65+
66+
Message<?> inputMessage = null;
67+
68+
MessageBuilder builder = MessageBuilder.withPayload(argument);
69+
if (!CollectionUtils.isEmpty(request.params())) {
70+
builder = builder.setHeader(FunctionHandlerHeaderUtils.HTTP_REQUEST_PARAM,
71+
request.params().toSingleValueMap());
72+
}
73+
inputMessage = builder.copyHeaders(headers.toSingleValueMap()).build();
74+
75+
if (function.isRoutingFunction()) {
76+
function.setSkipOutputConversion(true);
77+
}
78+
79+
Object result = function.apply(inputMessage);
80+
if (function.isConsumer()) {
81+
/*
82+
* if (result instanceof Publisher) { Mono.from((Publisher)
83+
* result).subscribe(); }
84+
*/
85+
return HttpMethod.DELETE.equals(request.method()) ? ServerResponse.ok().build()
86+
: ServerResponse.accepted()
87+
.headers(h -> h.addAll(sanitize(headers, ignoredHeaders, requestOnlyHeaders)))
88+
.build();
89+
// Mono.empty() :
90+
// Mono.just(ResponseEntity.accepted().headers(FunctionHandlerHeaderUtils.sanitize(headers,
91+
// ignoredHeaders, requestOnlyHeaders)).build());
92+
}
93+
94+
BodyBuilder responseOkBuilder = ServerResponse.ok()
95+
.headers(h -> h.addAll(sanitize(headers, ignoredHeaders, requestOnlyHeaders)));
96+
97+
// FIXME: Mono/Flux
98+
/*
99+
* Publisher pResult; if (result instanceof Publisher) { pResult = (Publisher)
100+
* result; if (eventStream) { return Flux.from(pResult); }
101+
*
102+
* if (pResult instanceof Flux) { pResult = ((Flux) pResult).onErrorContinue((e,
103+
* v) -> { logger.error("Failed to process value: " + v, (Throwable) e);
104+
* }).collectList(); } pResult = Mono.from(pResult); } else { pResult =
105+
* Mono.just(result); }
106+
*/
107+
108+
// return Mono.from(pResult).map(v -> {
109+
if (result instanceof Iterable i) {
110+
List aggregatedResult = (List) StreamSupport.stream(i.spliterator(), false).map(m -> {
111+
return m instanceof Message ? processMessage(responseOkBuilder, (Message<?>) m, ignoredHeaders) : m;
112+
}).collect(Collectors.toList());
113+
return responseOkBuilder.header("content-type", "application/json").body(aggregatedResult);
114+
}
115+
else if (result instanceof Message message) {
116+
return responseOkBuilder.body(processMessage(responseOkBuilder, message, ignoredHeaders));
117+
}
118+
else {
119+
return responseOkBuilder.body(result);
120+
}
121+
// });
122+
}
123+
124+
private static Object processMessage(BodyBuilder responseOkBuilder, Message<?> message,
125+
List<String> ignoredHeaders) {
126+
responseOkBuilder.headers(h -> h.addAll(fromMessage(message.getHeaders(), ignoredHeaders)));
127+
return message.getPayload();
128+
}
129+
130+
}

0 commit comments

Comments
 (0)