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

pom.xml

@@ -57,6 +57,8 @@
 		<junit-pioneer.version>2.3.0</junit-pioneer.version>
 		<spring-cloud-circuitbreaker.version>3.2.1-SNAPSHOT</spring-cloud-circuitbreaker.version>
 		<spring-cloud-commons.version>4.3.0-SNAPSHOT</spring-cloud-commons.version>
+		<spring-cloud-function.version>4.2.1-SNAPSHOT</spring-cloud-function.version>
+		<spring-cloud-stream.version>4.2.1-SNAPSHOT</spring-cloud-stream.version>
 	</properties>
 
 	<dependencyManagement>
@@ -75,6 +77,20 @@
 				<type>pom</type>
 				<scope>import</scope>
 			</dependency>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-function-dependencies</artifactId>
+				<version>${spring-cloud-function.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-stream-dependencies</artifactId>
+				<version>${spring-cloud-stream.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
 			<dependency>
 				<groupId>org.springframework.cloud</groupId>
 				<artifactId>spring-cloud-test-support</artifactId>

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

@@ -52,6 +52,16 @@
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-commons</artifactId>
         </dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-function-context</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream</artifactId>
+			<optional>true</optional>
+		</dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-loadbalancer</artifactId>
@@ -89,7 +99,22 @@
 			<artifactId>spring-boot-properties-migrator</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-testcontainers</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+			<scope>test</scope>
+		</dependency>
         <!-- Third party test dependencies -->
+		<dependency>
+			<groupId>org.awaitility</groupId>
+			<artifactId>awaitility</artifactId>
+			<scope>test</scope>
+		</dependency>
         <dependency>
             <groupId>com.bucket4j</groupId>
             <artifactId>bucket4j_jdk17-caffeine</artifactId>
@@ -105,5 +130,10 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>rabbitmq</artifactId>
+			<scope>test</scope>
+		</dependency>
     </dependencies>
 </project>

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<String> 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<String> ignoredHeders,
+			List<String> requestOnlyHeaders) {
+		HttpHeaders result = new HttpHeaders();
+		for (String name : request.keySet()) {
+			List<String> 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<String, Object> 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);
+	}
+
+}

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<String> ignoredHeaders, List<String> 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<String> ignoredHeaders) {
+		responseOkBuilder.headers(h -> h.addAll(fromMessage(message.getHeaders(), ignoredHeaders)));
+		return message.getPayload();
+	}
+
+}

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<ServerResponse> fn(RouteProperties routeProperties) {
+		// fn:fnName
+		return fn(routeProperties.getUri().getSchemeSpecificPart());
+	}
+
+	public static HandlerFunction<ServerResponse> 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<ServerResponse> stream(RouteProperties routeProperties) {
+		// stream:bindingName
+		return stream(routeProperties.getUri().getSchemeSpecificPart());
+	}
+
+	public static HandlerFunction<ServerResponse> 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<ServerResponse> forward(RouteProperties routeProperties) {
 		return forward(routeProperties.getUri().getPath());
 	}

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<String, String> upper() {
+			return s -> s.toUpperCase(Locale.ROOT);
+		}
+
+	}
+
+}

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<String> consumeHello(AtomicBoolean helloConsumed) {
+			return message -> {
+				helloConsumed.compareAndSet(false, true);
+			};
+		}
+
+	}
+
+}

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<String, String> upper() {
+			return s -> s.toUpperCase(Locale.ROOT);
+		}
+
+		@Bean
+		public RouterFunction<ServerResponse> gatewayRouterFunctionsSimpleFunction() {
+			// @formatter:off
+			return route("testsimplefunction")
+					.POST("/simplefunction", fn("upper"))
+					.build()
+				.and(route("testtemplatedfunction")
+					.POST("/templatedfunction/{fnName}", fn("{fnName}"))
+					.build());
+			// @formatter:on
+		}
+
+	}
+
+}

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<ServerResponse> 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<String> consumeHello(AtomicBoolean helloConsumed) {
+			return message -> helloConsumed.compareAndSet(false, true);
+		}
+
+	}
+
+}

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

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

spring-cloud-gateway-server/pom.xml

@@ -59,6 +59,16 @@
 			<artifactId>spring-cloud-loadbalancer</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-function-context</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-devtools</artifactId>
@@ -182,6 +192,16 @@
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-testcontainers</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.junit-pioneer</groupId>
 			<artifactId>junit-pioneer</artifactId>
@@ -192,6 +212,11 @@
 			<artifactId>spring-cloud-test-support</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>io.projectreactor</groupId>
 			<artifactId>reactor-test</artifactId>
@@ -217,12 +242,17 @@
 			<artifactId>junit-jupiter</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>rabbitmq</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.junit.platform</groupId>
 			<artifactId>junit-platform-launcher</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
 			<groupId>org.openjdk.jmh</groupId>
 			<artifactId>jmh-core</artifactId>
 			<version>1.20</version>

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<MessageBodyEncoder> messageBodyEncoders) {
+		return new FunctionRoutingFilter(functionCatalog, codecConfigurer.getReaders(), messageBodyEncoders);
+	}
+
+}

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

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<HttpMessageReader<?>> messageReaders;
+
+	private final Map<String, MessageBodyEncoder> messageBodyEncoders;
+
+	public FunctionRoutingFilter(FunctionCatalog functionCatalog, List<HttpMessageReader<?>> messageReaders,
+			Set<MessageBodyEncoder> 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<Void> 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<Void> processRequest(ServerWebExchange exchange, FunctionInvocationWrapper function,
+			List<HttpMessageReader<?>> messageReaders, Map<String, MessageBodyEncoder> 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<String> 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<DataBuffer> 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<DataBuffer> writeBody(ServerHttpResponse httpResponse, CachedBodyOutputMessage message,
+			Class<?> outClass) {
+		Mono<DataBuffer> response = DataBufferUtils.join(message.getBody());
+		if (byte[].class.isAssignableFrom(outClass)) {
+			return response;
+		}
+
+		List<String> 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;
+	}
+
+}

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<HttpMessageReader<?>> messageReaders;
+
+	private final SetStatusGatewayFilterFactory setStatusFilter;
+
+	public StreamRoutingFilter(StreamBridge streamBridge, List<HttpMessageReader<?>> 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<Void> 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);
+		});
+	}
+
+}

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<String> 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<String> ignoredHeders,
+			List<String> requestOnlyHeaders) {
+		HttpHeaders result = new HttpHeaders();
+		for (String name : request.keySet()) {
+			List<String> 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<String, Object> 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);
+	}
+
+}

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

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

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<String> consumeHello(AtomicBoolean helloConsumed) {
+			return message -> helloConsumed.compareAndSet(false, true);
+		}
+
+	}
+
+}