diff --git a/pom.xml b/pom.xml index 4ca15e53..d8937b9b 100644 --- a/pom.xml +++ b/pom.xml @@ -57,6 +57,8 @@ 2.3.0 3.2.1-SNAPSHOT 4.3.0-SNAPSHOT + 4.2.1-SNAPSHOT + 4.2.1-SNAPSHOT @@ -75,6 +77,20 @@ pom import + + org.springframework.cloud + spring-cloud-function-dependencies + ${spring-cloud-function.version} + pom + import + + + org.springframework.cloud + spring-cloud-stream-dependencies + ${spring-cloud-stream.version} + pom + import + org.springframework.cloud spring-cloud-test-support diff --git a/spring-cloud-gateway-server-mvc/pom.xml b/spring-cloud-gateway-server-mvc/pom.xml index 62aa2075..6f7ff6b3 100644 --- a/spring-cloud-gateway-server-mvc/pom.xml +++ b/spring-cloud-gateway-server-mvc/pom.xml @@ -52,6 +52,16 @@ org.springframework.cloud spring-cloud-commons + + org.springframework.cloud + spring-cloud-function-context + true + + + org.springframework.cloud + spring-cloud-stream + true + org.springframework.cloud spring-cloud-loadbalancer @@ -89,7 +99,22 @@ spring-boot-properties-migrator test + + org.springframework.boot + spring-boot-testcontainers + test + + + org.springframework.cloud + spring-cloud-starter-stream-rabbit + test + + + org.awaitility + awaitility + test + com.bucket4j bucket4j_jdk17-caffeine @@ -105,5 +130,10 @@ junit-jupiter test + + org.testcontainers + rabbitmq + test + \ No newline at end of file diff --git a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerHeaderUtils.java b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerHeaderUtils.java new file mode 100644 index 00000000..891f05c1 --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerHeaderUtils.java @@ -0,0 +1,116 @@ +/* + * Copyright 2012-2019 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.gateway.server.mvc.handler; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import org.springframework.http.HttpHeaders; +import org.springframework.messaging.MessageHeaders; + +/** + * @author Dave Syer + * @author Oleg Zhurakousky + */ +public final class FunctionHandlerHeaderUtils { + + /** + * Message Header name which contains HTTP request parameters. + */ + public static final String HTTP_REQUEST_PARAM = "http_request_param"; + + private static HttpHeaders IGNORED = new HttpHeaders(); + + private static HttpHeaders REQUEST_ONLY = new HttpHeaders(); + + static { + IGNORED.add(MessageHeaders.ID, ""); + IGNORED.add(HttpHeaders.CONTENT_LENGTH, "0"); + // Headers that would typically be added by a downstream client + REQUEST_ONLY.add(HttpHeaders.ACCEPT, ""); + REQUEST_ONLY.add(HttpHeaders.CONTENT_LENGTH, ""); + REQUEST_ONLY.add(HttpHeaders.CONTENT_TYPE, ""); + REQUEST_ONLY.add(HttpHeaders.HOST, ""); + } + + private FunctionHandlerHeaderUtils() { + throw new IllegalStateException("Can't instantiate a utility class"); + } + + public static HttpHeaders fromMessage(MessageHeaders headers, List ignoredHeders) { + HttpHeaders result = new HttpHeaders(); + for (String name : headers.keySet()) { + Object value = headers.get(name); + name = name.toLowerCase(Locale.ROOT); + if (!IGNORED.containsKey(name) && !ignoredHeders.contains(name)) { + Collection values = multi(value); + for (Object object : values) { + result.set(name, object.toString()); + } + } + } + return result; + } + + @SuppressWarnings("unchecked") + public static HttpHeaders fromMessage(MessageHeaders headers) { + return fromMessage(headers, Collections.EMPTY_LIST); + } + + public static HttpHeaders sanitize(HttpHeaders request, List ignoredHeders, + List requestOnlyHeaders) { + HttpHeaders result = new HttpHeaders(); + for (String name : request.keySet()) { + List value = request.get(name); + name = name.toLowerCase(Locale.ROOT); + if (!IGNORED.containsKey(name) && !REQUEST_ONLY.containsKey(name) && !ignoredHeders.contains(name) + && !requestOnlyHeaders.contains(name)) { + result.put(name, value); + } + } + return result; + } + + @SuppressWarnings("unchecked") + public static HttpHeaders sanitize(HttpHeaders request) { + return sanitize(request, Collections.EMPTY_LIST, Collections.EMPTY_LIST); + } + + public static MessageHeaders fromHttp(HttpHeaders headers) { + Map map = new LinkedHashMap<>(); + for (String name : headers.keySet()) { + Collection values = multi(headers.get(name)); + name = name.toLowerCase(Locale.ROOT); + Object value = values == null ? null : (values.size() == 1 ? values.iterator().next() : values); + if (name.toLowerCase(Locale.ROOT).equals(HttpHeaders.CONTENT_TYPE.toLowerCase(Locale.ROOT))) { + name = MessageHeaders.CONTENT_TYPE; + } + map.put(name, value); + } + return new MessageHeaders(map); + } + + private static Collection multi(Object value) { + return value instanceof Collection ? (Collection) value : Arrays.asList(value); + } + +} diff --git a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerRequestProcessingHelper.java b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerRequestProcessingHelper.java new file mode 100644 index 00000000..1990991b --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerRequestProcessingHelper.java @@ -0,0 +1,130 @@ +/* + * Copyright 2019-2021 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.gateway.server.mvc.handler; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.CollectionUtils; +import org.springframework.web.servlet.function.ServerRequest; +import org.springframework.web.servlet.function.ServerResponse; +import org.springframework.web.servlet.function.ServerResponse.BodyBuilder; + +import static org.springframework.cloud.gateway.server.mvc.handler.FunctionHandlerHeaderUtils.fromMessage; +import static org.springframework.cloud.gateway.server.mvc.handler.FunctionHandlerHeaderUtils.sanitize; + +/** + * !INTERNAL USE ONLY! + * + * @author Oleg Zhurakousky + * + */ +final class FunctionHandlerRequestProcessingHelper { + + private static Log logger = LogFactory.getLog(FunctionHandlerRequestProcessingHelper.class); + + private FunctionHandlerRequestProcessingHelper() { + + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + static ServerResponse processRequest(ServerRequest request, FunctionInvocationWrapper function, Object argument, + boolean eventStream, List ignoredHeaders, List requestOnlyHeaders) { + if (argument == null) { + argument = ""; + } + + if (function == null) { + return ServerResponse.notFound().build(); + } + + HttpHeaders headers = request.headers().asHttpHeaders(); + + Message inputMessage = null; + + MessageBuilder builder = MessageBuilder.withPayload(argument); + if (!CollectionUtils.isEmpty(request.params())) { + builder = builder.setHeader(FunctionHandlerHeaderUtils.HTTP_REQUEST_PARAM, + request.params().toSingleValueMap()); + } + inputMessage = builder.copyHeaders(headers.toSingleValueMap()).build(); + + if (function.isRoutingFunction()) { + function.setSkipOutputConversion(true); + } + + Object result = function.apply(inputMessage); + if (function.isConsumer()) { + /* + * if (result instanceof Publisher) { Mono.from((Publisher) + * result).subscribe(); } + */ + return HttpMethod.DELETE.equals(request.method()) ? ServerResponse.ok().build() + : ServerResponse.accepted() + .headers(h -> h.addAll(sanitize(headers, ignoredHeaders, requestOnlyHeaders))) + .build(); + // Mono.empty() : + // Mono.just(ResponseEntity.accepted().headers(FunctionHandlerHeaderUtils.sanitize(headers, + // ignoredHeaders, requestOnlyHeaders)).build()); + } + + BodyBuilder responseOkBuilder = ServerResponse.ok() + .headers(h -> h.addAll(sanitize(headers, ignoredHeaders, requestOnlyHeaders))); + + // FIXME: Mono/Flux + /* + * Publisher pResult; if (result instanceof Publisher) { pResult = (Publisher) + * result; if (eventStream) { return Flux.from(pResult); } + * + * if (pResult instanceof Flux) { pResult = ((Flux) pResult).onErrorContinue((e, + * v) -> { logger.error("Failed to process value: " + v, (Throwable) e); + * }).collectList(); } pResult = Mono.from(pResult); } else { pResult = + * Mono.just(result); } + */ + + // return Mono.from(pResult).map(v -> { + if (result instanceof Iterable i) { + List aggregatedResult = (List) StreamSupport.stream(i.spliterator(), false).map(m -> { + return m instanceof Message ? processMessage(responseOkBuilder, (Message) m, ignoredHeaders) : m; + }).collect(Collectors.toList()); + return responseOkBuilder.header("content-type", "application/json").body(aggregatedResult); + } + else if (result instanceof Message message) { + return responseOkBuilder.body(processMessage(responseOkBuilder, message, ignoredHeaders)); + } + else { + return responseOkBuilder.body(result); + } + // }); + } + + private static Object processMessage(BodyBuilder responseOkBuilder, Message message, + List ignoredHeaders) { + responseOkBuilder.headers(h -> h.addAll(fromMessage(message.getHeaders(), ignoredHeaders))); + return message.getPayload(); + } + +} diff --git a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/HandlerFunctions.java b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/HandlerFunctions.java index 9b3d118b..4b35cc44 100644 --- a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/HandlerFunctions.java +++ b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/HandlerFunctions.java @@ -21,22 +21,77 @@ import java.lang.reflect.Method; import java.net.URI; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.concurrent.atomic.AtomicReference; import jakarta.servlet.ServletException; +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.gateway.server.mvc.common.MvcUtils; import org.springframework.cloud.gateway.server.mvc.config.RouteProperties; +import org.springframework.cloud.stream.function.StreamOperations; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.Assert; +import org.springframework.util.MimeType; import org.springframework.web.servlet.function.HandlerFunction; import org.springframework.web.servlet.function.ServerRequest; import org.springframework.web.servlet.function.ServerResponse; +import static org.springframework.cloud.gateway.server.mvc.handler.FunctionHandlerRequestProcessingHelper.processRequest; + public abstract class HandlerFunctions { private HandlerFunctions() { } + // for properties + public static HandlerFunction fn(RouteProperties routeProperties) { + // fn:fnName + return fn(routeProperties.getUri().getSchemeSpecificPart()); + } + + public static HandlerFunction fn(String functionName) { + Assert.hasText(functionName, "'functionName' must not be empty"); + return request -> { + String expandedFunctionName = MvcUtils.expand(request, functionName); + FunctionCatalog functionCatalog = MvcUtils.getApplicationContext(request).getBean(FunctionCatalog.class); + FunctionInvocationWrapper function = functionCatalog.lookup(expandedFunctionName, + request.headers().accept().stream().map(MimeType::toString).toArray(String[]::new)); + if (function != null) { + Object body = request.body(function.getRawInputType()); + return processRequest(request, function, body, false, Collections.emptyList(), Collections.emptyList()); + } + return ServerResponse.notFound().build(); + }; + } + + // for properties + public static HandlerFunction stream(RouteProperties routeProperties) { + // stream:bindingName + return stream(routeProperties.getUri().getSchemeSpecificPart()); + } + + public static HandlerFunction stream(String bindingName) { + Assert.hasText(bindingName, "'bindingName' must not be empty"); + // TODO: validate bindingName + return request -> { + String expandedBindingName = MvcUtils.expand(request, bindingName); + StreamOperations streamOps = MvcUtils.getApplicationContext(request).getBean(StreamOperations.class); + byte[] body = request.body(byte[].class); + MessageHeaders messageHeaders = FunctionHandlerHeaderUtils + .fromHttp(FunctionHandlerHeaderUtils.sanitize(request.headers().asHttpHeaders())); + boolean send = streamOps.send(expandedBindingName, MessageBuilder.createMessage(body, messageHeaders)); + if (send) { + return ServerResponse.accepted().build(); + } + return ServerResponse.badRequest().build(); + }; + } + + // for properties public static HandlerFunction forward(RouteProperties routeProperties) { return forward(routeProperties.getUri().getPath()); } diff --git a/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/config/FunctionHandlerConfigTests.java b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/config/FunctionHandlerConfigTests.java new file mode 100644 index 00000000..21fe10f2 --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/config/FunctionHandlerConfigTests.java @@ -0,0 +1,77 @@ +/* + * Copyright 2013-2024 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.gateway.server.mvc.config; + +import java.util.Locale; +import java.util.function.Function; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.gateway.server.mvc.test.client.TestRestClient; +import org.springframework.context.annotation.Bean; +import org.springframework.http.MediaType; +import org.springframework.test.context.ActiveProfiles; + +@SpringBootTest(properties = {}, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +@ActiveProfiles("functionhandlerconfigtests") +public class FunctionHandlerConfigTests { + + @Autowired + private TestRestClient restClient; + + @Test + public void testSimpleFunctionWorks() { + restClient.post() + .uri("/simplefunction") + .accept(MediaType.TEXT_PLAIN) + .bodyValue("hello") + .exchange() + .expectStatus() + .isOk() + .expectBody(String.class) + .isEqualTo("HELLO"); + } + + @Test + public void testTemplatedFunctionWorks() { + restClient.post() + .uri("/templatedfunction/upper") + .accept(MediaType.TEXT_PLAIN) + .bodyValue("hello") + .exchange() + .expectStatus() + .isOk() + .expectBody(String.class) + .isEqualTo("HELLO"); + } + + @SpringBootConfiguration + @EnableAutoConfiguration + protected static class TestConfiguration { + + @Bean + Function upper() { + return s -> s.toUpperCase(Locale.ROOT); + } + + } + +} diff --git a/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/config/StreamHandlerConfigTests.java b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/config/StreamHandlerConfigTests.java new file mode 100644 index 00000000..5227cc70 --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/config/StreamHandlerConfigTests.java @@ -0,0 +1,107 @@ +/* + * Copyright 2013-2024 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.gateway.server.mvc.config; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.RabbitMQContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.cloud.gateway.server.mvc.test.client.TestRestClient; +import org.springframework.context.annotation.Bean; +import org.springframework.http.MediaType; +import org.springframework.test.context.ActiveProfiles; + +import static org.assertj.core.api.Assertions.assertThat; + +@SpringBootTest( + properties = { "spring.cloud.function.definition=consumeHello", + "spring.cloud.stream.bindings.consumeHello-in-0.destination=hello-out-0" }, + webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +@ActiveProfiles("streamhandlerconfigtests") +@Testcontainers +public class StreamHandlerConfigTests { + + @Container + @ServiceConnection + public static RabbitMQContainer rabbitmq = new RabbitMQContainer("rabbitmq:3.7.25-management-alpine"); + + @Autowired + private TestRestClient restClient; + + @Autowired + private AtomicBoolean helloConsumed; + + @Test + public void testSimpleStreamWorks() { + helloConsumed.set(false); + restClient.post() + .uri("/simplestream") + .accept(MediaType.TEXT_PLAIN) + .bodyValue("hello") + .exchange() + .expectStatus() + .isAccepted(); + + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> helloConsumed.get()); + assertThat(helloConsumed).isTrue(); + } + + @Test + public void testTemplatedStreamWorks() { + helloConsumed.set(false); + restClient.post() + .uri("/templatedstream/hello") + .accept(MediaType.TEXT_PLAIN) + .bodyValue("hello") + .exchange() + .expectStatus() + .isAccepted(); + + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> helloConsumed.get()); + assertThat(helloConsumed).isTrue(); + } + + @SpringBootConfiguration + @EnableAutoConfiguration + protected static class TestConfiguration { + + @Bean + public AtomicBoolean helloConsumed() { + return new AtomicBoolean(false); + } + + @Bean + public Consumer consumeHello(AtomicBoolean helloConsumed) { + return message -> { + helloConsumed.compareAndSet(false, true); + }; + } + + } + +} diff --git a/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerTests.java b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerTests.java new file mode 100644 index 00000000..28d17153 --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/handler/FunctionHandlerTests.java @@ -0,0 +1,92 @@ +/* + * Copyright 2013-2024 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.gateway.server.mvc.handler; + +import java.util.Locale; +import java.util.function.Function; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.gateway.server.mvc.test.client.TestRestClient; +import org.springframework.context.annotation.Bean; +import org.springframework.http.MediaType; +import org.springframework.web.servlet.function.RouterFunction; +import org.springframework.web.servlet.function.ServerResponse; + +import static org.springframework.cloud.gateway.server.mvc.handler.GatewayRouterFunctions.route; +import static org.springframework.cloud.gateway.server.mvc.handler.HandlerFunctions.fn; + +@SpringBootTest(properties = {}, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +public class FunctionHandlerTests { + + @Autowired + private TestRestClient restClient; + + @Test + public void testSimpleFunctionWorks() { + restClient.post() + .uri("/simplefunction") + .accept(MediaType.TEXT_PLAIN) + .bodyValue("hello") + .exchange() + .expectStatus() + .isOk() + .expectBody(String.class) + .isEqualTo("HELLO"); + } + + @Test + public void testTemplatedFunctionWorks() { + restClient.post() + .uri("/templatedfunction/upper") + .accept(MediaType.TEXT_PLAIN) + .bodyValue("hello") + .exchange() + .expectStatus() + .isOk() + .expectBody(String.class) + .isEqualTo("HELLO"); + } + + @SpringBootConfiguration + @EnableAutoConfiguration + protected static class TestConfiguration { + + @Bean + Function upper() { + return s -> s.toUpperCase(Locale.ROOT); + } + + @Bean + public RouterFunction gatewayRouterFunctionsSimpleFunction() { + // @formatter:off + return route("testsimplefunction") + .POST("/simplefunction", fn("upper")) + .build() + .and(route("testtemplatedfunction") + .POST("/templatedfunction/{fnName}", fn("{fnName}")) + .build()); + // @formatter:on + } + + } + +} diff --git a/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/handler/StreamHandlerTests.java b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/handler/StreamHandlerTests.java new file mode 100644 index 00000000..aa89417d --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/handler/StreamHandlerTests.java @@ -0,0 +1,119 @@ +/* + * Copyright 2013-2024 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.gateway.server.mvc.handler; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.RabbitMQContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.cloud.gateway.server.mvc.test.client.TestRestClient; +import org.springframework.context.annotation.Bean; +import org.springframework.http.MediaType; +import org.springframework.web.servlet.function.RouterFunction; +import org.springframework.web.servlet.function.ServerResponse; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.cloud.gateway.server.mvc.handler.GatewayRouterFunctions.route; +import static org.springframework.cloud.gateway.server.mvc.handler.HandlerFunctions.stream; + +@SpringBootTest( + properties = { "spring.cloud.function.definition=consumeHello", + "spring.cloud.stream.bindings.consumeHello-in-0.destination=hello-out-0" }, + webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +@Testcontainers +public class StreamHandlerTests { + + @Container + @ServiceConnection + public static RabbitMQContainer rabbitmq = new RabbitMQContainer("rabbitmq:3.7.25-management-alpine"); + + @Autowired + private TestRestClient restClient; + + @Autowired + private AtomicBoolean helloConsumed; + + @Test + public void testSimpleStreamWorks() { + helloConsumed.set(false); + restClient.post() + .uri("/simplestream") + .accept(MediaType.TEXT_PLAIN) + .bodyValue("hello") + .exchange() + .expectStatus() + .isAccepted(); + + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> helloConsumed.get()); + assertThat(helloConsumed).isTrue(); + } + + @Test + public void testTemplatedStreamWorks() { + helloConsumed.set(false); + restClient.post() + .uri("/templatedstream/hello") + .accept(MediaType.TEXT_PLAIN) + .bodyValue("hello") + .exchange() + .expectStatus() + .isAccepted(); + + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> helloConsumed.get()); + assertThat(helloConsumed).isTrue(); + } + + @SpringBootConfiguration + @EnableAutoConfiguration + protected static class TestConfiguration { + + @Bean + public RouterFunction gatewayRouterFunctionsSimpleStream() { + // @formatter:off + return route("testsimplestream") + .POST("/simplestream", stream("hello-out-0")) + .build() + .and(route("testtemplatedstream") + .POST("/templatedstream/{name}", stream("{name}-out-0")) + .build()); + // @formatter:on + } + + @Bean + public AtomicBoolean helloConsumed() { + return new AtomicBoolean(false); + } + + @Bean + public Consumer consumeHello(AtomicBoolean helloConsumed) { + return message -> helloConsumed.compareAndSet(false, true); + } + + } + +} diff --git a/spring-cloud-gateway-server-mvc/src/test/resources/application-functionhandlerconfigtests.yml b/spring-cloud-gateway-server-mvc/src/test/resources/application-functionhandlerconfigtests.yml new file mode 100644 index 00000000..0690e34b --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/test/resources/application-functionhandlerconfigtests.yml @@ -0,0 +1,16 @@ +spring.cloud.gateway.mvc: + routesMap: + testsimplefunction: + uri: fn:upper + predicates: + - Path=/simplefunction + - Method=POST + testtemplatedfunction: + uri: fn:{fnName} + predicates: + - Path=/templatedfunction/{fnName} + - Method=POST + +logging: + level: + org.springframework.cloud.gateway.server.mvc: TRACE diff --git a/spring-cloud-gateway-server-mvc/src/test/resources/application-streamhandlerconfigtests.yml b/spring-cloud-gateway-server-mvc/src/test/resources/application-streamhandlerconfigtests.yml new file mode 100644 index 00000000..3e97fc80 --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/test/resources/application-streamhandlerconfigtests.yml @@ -0,0 +1,16 @@ +spring.cloud.gateway.mvc: + routesMap: + testsimplestream: + uri: stream:hello-out-0 + predicates: + - Path=/simplestream + - Method=POST + testtemplatedstream: + uri: stream:{name}-out-0 + predicates: + - Path=/templatedstream/{name} + - Method=POST + +logging: + level: + org.springframework.cloud.gateway.server.mvc: TRACE diff --git a/spring-cloud-gateway-server/pom.xml b/spring-cloud-gateway-server/pom.xml index 8ba0bd52..e405dd49 100644 --- a/spring-cloud-gateway-server/pom.xml +++ b/spring-cloud-gateway-server/pom.xml @@ -59,6 +59,16 @@ spring-cloud-loadbalancer true + + org.springframework.cloud + spring-cloud-function-context + true + + + org.springframework.cloud + spring-cloud-stream + true + org.springframework.boot spring-boot-devtools @@ -182,6 +192,16 @@ spring-boot-starter-test test + + org.springframework.boot + spring-boot-testcontainers + test + + + org.springframework.cloud + spring-cloud-starter-stream-rabbit + test + org.junit-pioneer junit-pioneer @@ -192,6 +212,11 @@ spring-cloud-test-support test + + org.springframework.cloud + spring-cloud-starter-stream-rabbit + test + io.projectreactor reactor-test @@ -217,12 +242,17 @@ junit-jupiter test + + org.testcontainers + rabbitmq + test + org.junit.platform junit-platform-launcher - test - - + test + + org.openjdk.jmh jmh-core 1.20 diff --git a/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/config/GatewayFunctionAutoConfiguration.java b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/config/GatewayFunctionAutoConfiguration.java new file mode 100644 index 00000000..0afedfbe --- /dev/null +++ b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/config/GatewayFunctionAutoConfiguration.java @@ -0,0 +1,52 @@ +/* + * Copyright 2013-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.gateway.config; + +import java.util.Set; + +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.web.reactive.HttpHandlerAutoConfiguration; +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration; +import org.springframework.cloud.gateway.config.conditional.ConditionalOnEnabledGlobalFilter; +import org.springframework.cloud.gateway.filter.FunctionRoutingFilter; +import org.springframework.cloud.gateway.filter.factory.rewrite.MessageBodyEncoder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.codec.ServerCodecConfigurer; +import org.springframework.web.reactive.DispatcherHandler; + +@Configuration(proxyBeanMethods = false) +@AutoConfigureAfter(ContextFunctionCatalogAutoConfiguration.class) +@AutoConfigureBefore({ HttpHandlerAutoConfiguration.class, GatewayAutoConfiguration.class }) +@ConditionalOnClass({ FunctionCatalog.class, DispatcherHandler.class }) +@ConditionalOnProperty(name = "spring.cloud.gateway.function.enabled", matchIfMissing = true) +class GatewayFunctionAutoConfiguration { + + @Bean + @ConditionalOnEnabledGlobalFilter + @ConditionalOnBean(FunctionCatalog.class) + public FunctionRoutingFilter functionRoutingFilter(FunctionCatalog functionCatalog, + ServerCodecConfigurer codecConfigurer, Set messageBodyEncoders) { + return new FunctionRoutingFilter(functionCatalog, codecConfigurer.getReaders(), messageBodyEncoders); + } + +} diff --git a/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/config/GatewayStreamAutoConfiguration.java b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/config/GatewayStreamAutoConfiguration.java new file mode 100644 index 00000000..ba5dcbb4 --- /dev/null +++ b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/config/GatewayStreamAutoConfiguration.java @@ -0,0 +1,48 @@ +/* + * Copyright 2013-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.gateway.config; + +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.web.reactive.HttpHandlerAutoConfiguration; +import org.springframework.cloud.gateway.config.conditional.ConditionalOnEnabledGlobalFilter; +import org.springframework.cloud.gateway.filter.StreamRoutingFilter; +import org.springframework.cloud.stream.config.BindingServiceConfiguration; +import org.springframework.cloud.stream.function.StreamBridge; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.codec.ServerCodecConfigurer; +import org.springframework.web.reactive.DispatcherHandler; + +@Configuration(proxyBeanMethods = false) +@AutoConfigureAfter(BindingServiceConfiguration.class) +@AutoConfigureBefore({ HttpHandlerAutoConfiguration.class, GatewayAutoConfiguration.class }) +@ConditionalOnClass({ StreamBridge.class, DispatcherHandler.class }) +@ConditionalOnProperty(name = "spring.cloud.gateway.stream.enabled", matchIfMissing = true) +class GatewayStreamAutoConfiguration { + + @Bean + @ConditionalOnEnabledGlobalFilter + @ConditionalOnBean(StreamBridge.class) + public StreamRoutingFilter streamRoutingFilter(StreamBridge streamBridge, ServerCodecConfigurer codecConfigurer) { + return new StreamRoutingFilter(streamBridge, codecConfigurer.getReaders()); + } + +} diff --git a/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/filter/FunctionRoutingFilter.java b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/filter/FunctionRoutingFilter.java new file mode 100644 index 00000000..e35227dd --- /dev/null +++ b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/filter/FunctionRoutingFilter.java @@ -0,0 +1,198 @@ +/* + * Copyright 2013-2024 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.gateway.filter; + +import java.net.URI; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; +import org.springframework.cloud.gateway.filter.factory.rewrite.CachedBodyOutputMessage; +import org.springframework.cloud.gateway.filter.factory.rewrite.MessageBodyEncoder; +import org.springframework.cloud.gateway.support.BodyInserterContext; +import org.springframework.cloud.gateway.support.MessageHeaderUtils; +import org.springframework.cloud.gateway.support.NotFoundException; +import org.springframework.core.Ordered; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferFactory; +import org.springframework.core.io.buffer.DataBufferUtils; +import org.springframework.http.HttpHeaders; +import org.springframework.http.codec.HttpMessageReader; +import org.springframework.http.server.reactive.ServerHttpRequest; +import org.springframework.http.server.reactive.ServerHttpResponse; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.CollectionUtils; +import org.springframework.util.MimeType; +import org.springframework.web.reactive.function.BodyInserter; +import org.springframework.web.reactive.function.BodyInserters; +import org.springframework.web.reactive.function.server.ServerRequest; +import org.springframework.web.server.ServerWebExchange; + +import static java.util.function.Function.identity; +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR; +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.isAlreadyRouted; +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.setAlreadyRouted; + +/** + * @author Spencer Gibb + */ +public class FunctionRoutingFilter implements GlobalFilter, Ordered { + + private static Log logger = LogFactory.getLog(FunctionRoutingFilter.class); + + private final FunctionCatalog functionCatalog; + + private final List> messageReaders; + + private final Map messageBodyEncoders; + + public FunctionRoutingFilter(FunctionCatalog functionCatalog, List> messageReaders, + Set messageBodyEncoders) { + this.functionCatalog = functionCatalog; + this.messageReaders = messageReaders; + this.messageBodyEncoders = messageBodyEncoders.stream() + .collect(Collectors.toMap(MessageBodyEncoder::encodingType, identity())); + } + + @Override + public int getOrder() { + return RouteToRequestUrlFilter.ROUTE_TO_URL_FILTER_ORDER + 10; + } + + @Override + public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { + URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR); + String scheme = requestUrl.getScheme(); + + if (isAlreadyRouted(exchange) || !"fn".equals(scheme)) { + return chain.filter(exchange); + } + setAlreadyRouted(exchange); + + FunctionInvocationWrapper function = functionCatalog.lookup(requestUrl.getHost(), + exchange.getRequest().getHeaders().getAccept().stream().map(MimeType::toString).toArray(String[]::new)); + if (function != null) { + return processRequest(exchange, function, messageReaders, messageBodyEncoders).then(chain.filter(exchange)); + } + + return Mono.error(new NotFoundException("No route for uri " + requestUrl)); + } + + protected Mono processRequest(ServerWebExchange exchange, FunctionInvocationWrapper function, + List> messageReaders, Map messageBodyEncoders) { + // 1- convert request body to function input type + // 2- call function + // 3- convert function return to raw data for response + ServerRequest serverRequest = ServerRequest.create(exchange, messageReaders); + return serverRequest.bodyToMono(function.getRawInputType()).flatMap(requestBody -> { + ServerHttpRequest request = exchange.getRequest(); + HttpHeaders headers = request.getHeaders(); + + Message inputMessage = null; + + MessageBuilder builder = MessageBuilder.withPayload(requestBody); + if (!CollectionUtils.isEmpty(request.getQueryParams())) { + builder = builder.setHeader(MessageHeaderUtils.HTTP_REQUEST_PARAM, + request.getQueryParams().toSingleValueMap()); + } + inputMessage = builder.copyHeaders(headers.toSingleValueMap()).build(); + + if (function.isRoutingFunction()) { + function.setSkipOutputConversion(true); + } + + List ignoredHeaders = Collections.emptyList(); + HttpHeaders newResponseHeaders = new HttpHeaders(); + + Object functionResult = function.apply(inputMessage); + if (functionResult instanceof Message message) { + newResponseHeaders.addAll(MessageHeaderUtils.fromMessage(message.getHeaders(), ignoredHeaders)); + functionResult = message.getPayload(); + } + Publisher result; + if (functionResult instanceof Publisher publisher) { + // TODO: deal with eventStream + result = publisher; + } + else { + result = Mono.just(functionResult); + } + + Class outClass = byte[].class; + + BodyInserter bodyInserter = BodyInserters.fromPublisher(result, outClass); + CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, + exchange.getResponse().getHeaders()); + + return bodyInserter.insert(outputMessage, new BodyInserterContext()).then(Mono.defer(() -> { + ServerHttpResponse response = exchange.getResponse(); + Mono messageBody = writeBody(response, outputMessage, outClass); + HttpHeaders responseHeaders = response.getHeaders(); + if (!responseHeaders.containsKey(HttpHeaders.TRANSFER_ENCODING) + || responseHeaders.containsKey(HttpHeaders.CONTENT_LENGTH)) { + messageBody = messageBody.doOnNext(data -> headers.setContentLength(data.readableByteCount())); + } + responseHeaders.addAll(newResponseHeaders); + + // TODO: deal with content type + /* + * if (StringUtils.hasText(config.newContentType)) { + * headers.set(HttpHeaders.CONTENT_TYPE, config.newContentType); } + */ + + // TODO: fail if isStreamingMediaType? + return response.writeWith(messageBody); + })); + }); + } + + private Mono writeBody(ServerHttpResponse httpResponse, CachedBodyOutputMessage message, + Class outClass) { + Mono response = DataBufferUtils.join(message.getBody()); + if (byte[].class.isAssignableFrom(outClass)) { + return response; + } + + List encodingHeaders = httpResponse.getHeaders().getOrEmpty(HttpHeaders.CONTENT_ENCODING); + for (String encoding : encodingHeaders) { + MessageBodyEncoder encoder = messageBodyEncoders.get(encoding); + if (encoder != null) { + DataBufferFactory dataBufferFactory = httpResponse.bufferFactory(); + response = response.publishOn(Schedulers.parallel()).map(buffer -> { + byte[] encodedResponse = encoder.encode(buffer); + DataBufferUtils.release(buffer); + return encodedResponse; + }).map(dataBufferFactory::wrap); + break; + } + } + + return response; + } + +} diff --git a/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/filter/StreamRoutingFilter.java b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/filter/StreamRoutingFilter.java new file mode 100644 index 00000000..8070ba05 --- /dev/null +++ b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/filter/StreamRoutingFilter.java @@ -0,0 +1,103 @@ +/* + * Copyright 2013-2024 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.gateway.filter; + +import java.net.URI; +import java.util.List; + +import reactor.core.publisher.Mono; + +import org.springframework.cloud.gateway.filter.factory.SetStatusGatewayFilterFactory; +import org.springframework.cloud.gateway.support.MessageHeaderUtils; +import org.springframework.cloud.stream.function.StreamBridge; +import org.springframework.core.Ordered; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.http.codec.HttpMessageReader; +import org.springframework.http.server.reactive.ServerHttpRequest; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.CollectionUtils; +import org.springframework.web.reactive.function.server.ServerRequest; +import org.springframework.web.server.ServerWebExchange; + +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR; +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.isAlreadyRouted; +import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.setAlreadyRouted; + +/** + * @author Spencer Gibb + */ +public class StreamRoutingFilter implements GlobalFilter, Ordered { + + private final StreamBridge streamBridge; + + private final List> messageReaders; + + private final SetStatusGatewayFilterFactory setStatusFilter; + + public StreamRoutingFilter(StreamBridge streamBridge, List> messageReaders) { + this.streamBridge = streamBridge; + this.messageReaders = messageReaders; + // TODO: is this the right place for this? + this.streamBridge.setAsync(true); + setStatusFilter = new SetStatusGatewayFilterFactory(); + } + + @Override + public int getOrder() { + return RouteToRequestUrlFilter.ROUTE_TO_URL_FILTER_ORDER + 10; + } + + @Override + public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { + URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR); + String scheme = requestUrl.getScheme(); + + if (isAlreadyRouted(exchange) || !"stream".equals(scheme)) { + return chain.filter(exchange); + } + setAlreadyRouted(exchange); + + ServerRequest serverRequest = ServerRequest.create(exchange, messageReaders); + return serverRequest.bodyToMono(byte[].class).flatMap(requestBody -> { + ServerHttpRequest request = exchange.getRequest(); + HttpHeaders headers = request.getHeaders(); + + Message inputMessage = null; + + MessageBuilder builder = MessageBuilder.withPayload(requestBody); + if (!CollectionUtils.isEmpty(request.getQueryParams())) { + // TODO: move HeaderUtils + builder = builder.setHeader(MessageHeaderUtils.HTTP_REQUEST_PARAM, + request.getQueryParams().toSingleValueMap()); + // TODO: sanitize? + } + inputMessage = builder.copyHeaders(headers.toSingleValueMap()).build(); + + // TODO: output content type + boolean send = streamBridge.send(requestUrl.getHost(), inputMessage); + + HttpStatus responseStatus = (send) ? HttpStatus.OK : HttpStatus.BAD_REQUEST; + + SetStatusGatewayFilterFactory.Config config = new SetStatusGatewayFilterFactory.Config(); + config.setStatus(responseStatus.name()); + return setStatusFilter.apply(config).filter(exchange, chain); + }); + } + +} diff --git a/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/support/MessageHeaderUtils.java b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/support/MessageHeaderUtils.java new file mode 100644 index 00000000..7916d77d --- /dev/null +++ b/spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/support/MessageHeaderUtils.java @@ -0,0 +1,116 @@ +/* + * Copyright 2013-2024 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.gateway.support; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import org.springframework.http.HttpHeaders; +import org.springframework.messaging.MessageHeaders; + +/** + * @author Dave Syer + * @author Oleg Zhurakousky + */ +public final class MessageHeaderUtils { + + /** + * Message Header name which contains HTTP request parameters. + */ + public static final String HTTP_REQUEST_PARAM = "http_request_param"; + + private static final HttpHeaders IGNORED = new HttpHeaders(); + + private static final HttpHeaders REQUEST_ONLY = new HttpHeaders(); + + static { + IGNORED.add(MessageHeaders.ID, ""); + IGNORED.add(HttpHeaders.CONTENT_LENGTH, "0"); + // Headers that would typically be added by a downstream client + REQUEST_ONLY.add(HttpHeaders.ACCEPT, ""); + REQUEST_ONLY.add(HttpHeaders.CONTENT_LENGTH, ""); + REQUEST_ONLY.add(HttpHeaders.CONTENT_TYPE, ""); + REQUEST_ONLY.add(HttpHeaders.HOST, ""); + } + + private MessageHeaderUtils() { + throw new IllegalStateException("Can't instantiate a utility class"); + } + + public static HttpHeaders fromMessage(MessageHeaders headers, List ignoredHeders) { + HttpHeaders result = new HttpHeaders(); + for (String name : headers.keySet()) { + Object value = headers.get(name); + name = name.toLowerCase(Locale.ROOT); + if (!IGNORED.containsKey(name) && !ignoredHeders.contains(name)) { + Collection values = multi(value); + for (Object object : values) { + result.set(name, object.toString()); + } + } + } + return result; + } + + @SuppressWarnings("unchecked") + public static HttpHeaders fromMessage(MessageHeaders headers) { + return fromMessage(headers, Collections.EMPTY_LIST); + } + + public static HttpHeaders sanitize(HttpHeaders request, List ignoredHeders, + List requestOnlyHeaders) { + HttpHeaders result = new HttpHeaders(); + for (String name : request.keySet()) { + List value = request.get(name); + name = name.toLowerCase(Locale.ROOT); + if (!IGNORED.containsKey(name) && !REQUEST_ONLY.containsKey(name) && !ignoredHeders.contains(name) + && !requestOnlyHeaders.contains(name)) { + result.put(name, value); + } + } + return result; + } + + @SuppressWarnings("unchecked") + public static HttpHeaders sanitize(HttpHeaders request) { + return sanitize(request, Collections.EMPTY_LIST, Collections.EMPTY_LIST); + } + + public static MessageHeaders fromHttp(HttpHeaders headers) { + Map map = new LinkedHashMap<>(); + for (String name : headers.keySet()) { + Collection values = multi(headers.get(name)); + name = name.toLowerCase(Locale.ROOT); + Object value = values == null ? null : (values.size() == 1 ? values.iterator().next() : values); + if (name.toLowerCase(Locale.ROOT).equals(HttpHeaders.CONTENT_TYPE.toLowerCase(Locale.ROOT))) { + name = MessageHeaders.CONTENT_TYPE; + } + map.put(name, value); + } + return new MessageHeaders(map); + } + + private static Collection multi(Object value) { + return value instanceof Collection ? (Collection) value : Arrays.asList(value); + } + +} diff --git a/spring-cloud-gateway-server/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring-cloud-gateway-server/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index 51f1dfb4..ed8ffb9d 100644 --- a/spring-cloud-gateway-server/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/spring-cloud-gateway-server/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -2,8 +2,10 @@ org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguratio org.springframework.cloud.gateway.config.GatewayAutoConfiguration org.springframework.cloud.gateway.config.GatewayResilience4JCircuitBreakerAutoConfiguration org.springframework.cloud.gateway.config.GatewayNoLoadBalancerClientAutoConfiguration +org.springframework.cloud.gateway.config.GatewayFunctionAutoConfiguration org.springframework.cloud.gateway.config.GatewayMetricsAutoConfiguration org.springframework.cloud.gateway.config.GatewayRedisAutoConfiguration +org.springframework.cloud.gateway.config.GatewayStreamAutoConfiguration org.springframework.cloud.gateway.discovery.GatewayDiscoveryClientAutoConfiguration org.springframework.cloud.gateway.config.SimpleUrlHandlerMappingGlobalCorsAutoConfiguration org.springframework.cloud.gateway.config.GatewayReactiveLoadBalancerClientAutoConfiguration diff --git a/spring-cloud-gateway-server/src/test/java/org/springframework/cloud/gateway/filter/FunctionRoutingFilterTests.java b/spring-cloud-gateway-server/src/test/java/org/springframework/cloud/gateway/filter/FunctionRoutingFilterTests.java new file mode 100644 index 00000000..3e889c44 --- /dev/null +++ b/spring-cloud-gateway-server/src/test/java/org/springframework/cloud/gateway/filter/FunctionRoutingFilterTests.java @@ -0,0 +1,80 @@ +/* + * Copyright 2013-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.gateway.filter; + +import java.net.URI; +import java.util.Locale; +import java.util.function.Function; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.gateway.route.RouteLocator; +import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder; +import org.springframework.cloud.gateway.test.BaseWebClientTests; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.http.MediaType; +import org.springframework.web.util.UriComponentsBuilder; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT; + +/** + * @author Spencer Gibb + */ +@SpringBootTest(webEnvironment = RANDOM_PORT, + properties = { "debug=false", "spring.cloud.function.definition=upper", "spring.codec.max-in-memory-size=40" }) +public class FunctionRoutingFilterTests extends BaseWebClientTests { + + @Test + public void functionRoutingFilterWorks() { + URI uri = UriComponentsBuilder.fromUriString(this.baseUri + "/").build(true).toUri(); + + testClient.post() + .uri(uri) + .bodyValue("hello") + .header("Host", "www.functionroutingfilterjava.org") + .accept(MediaType.TEXT_PLAIN) + .exchange() + .expectBody(String.class) + .consumeWith(res -> assertThat(res.getResponseBody()).isEqualTo("HELLO")); + } + + @EnableAutoConfiguration + @SpringBootConfiguration + @Import(DefaultTestConfig.class) + public static class TestConfig { + + @Bean + Function upper() { + return s -> s.toUpperCase(Locale.ROOT); + } + + @Bean + public RouteLocator testRouteLocator(RouteLocatorBuilder builder) { + return builder.routes() + .route("function_routing_filter_java_test", + r -> r.path("/").and().host("www.functionroutingfilterjava.org").uri("fn://upper")) + .build(); + } + + } + +} diff --git a/spring-cloud-gateway-server/src/test/java/org/springframework/cloud/gateway/filter/StreamRoutingFilterTests.java b/spring-cloud-gateway-server/src/test/java/org/springframework/cloud/gateway/filter/StreamRoutingFilterTests.java new file mode 100644 index 00000000..2c1abdf3 --- /dev/null +++ b/spring-cloud-gateway-server/src/test/java/org/springframework/cloud/gateway/filter/StreamRoutingFilterTests.java @@ -0,0 +1,106 @@ +/* + * Copyright 2013-2024 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.gateway.filter; + +import java.net.URI; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.RabbitMQContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.cloud.gateway.route.RouteLocator; +import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder; +import org.springframework.cloud.gateway.test.BaseWebClientTests; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.http.MediaType; +import org.springframework.web.util.UriComponentsBuilder; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT; + +/** + * @author Spencer Gibb + */ +@SpringBootTest(webEnvironment = RANDOM_PORT, + properties = { "debug=false", "spring.cloud.function.definition=consumeHello", + "spring.cloud.stream.bindings.consumeHello-in-0.destination=hello-out-0" }) +@Testcontainers +public class StreamRoutingFilterTests extends BaseWebClientTests { + + @Container + @ServiceConnection + public static RabbitMQContainer rabbitmq = new RabbitMQContainer("rabbitmq:3.7.25-management-alpine"); + + @Autowired + private AtomicBoolean helloConsumed; + + @Test + public void streamRoutingFilterWorks() { + helloConsumed.set(false); + + URI uri = UriComponentsBuilder.fromUriString(this.baseUri + "/").build(true).toUri(); + + testClient.post() + .uri(uri) + .bodyValue("hello") + .header("Host", "www.streamroutingfilterjava.org") + .accept(MediaType.TEXT_PLAIN) + .exchange() + .expectStatus() + .isOk(); + + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> helloConsumed.get()); + assertThat(helloConsumed).isTrue(); + } + + @EnableAutoConfiguration + @SpringBootConfiguration + @Import(DefaultTestConfig.class) + public static class TestConfig { + + @Bean + public RouteLocator testRouteLocator(RouteLocatorBuilder builder) { + return builder.routes() + .route("stream_routing_filter_java_test", + r -> r.path("/").and().host("www.streamroutingfilterjava.org").uri("stream://hello-out-0")) + .build(); + } + + @Bean + public AtomicBoolean helloConsumed() { + return new AtomicBoolean(false); + } + + @Bean + public Consumer consumeHello(AtomicBoolean helloConsumed) { + return message -> helloConsumed.compareAndSet(false, true); + } + + } + +}