Enable body filter codec encoding customization via CodecCustomizer

spring-cloud-gateway-server-webflux/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java

@@ -56,6 +56,7 @@
 import org.springframework.boot.autoconfigure.condition.NoneNestedConditions;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.boot.context.properties.PropertyMapper;
+import org.springframework.boot.http.codec.CodecCustomizer;
 import org.springframework.boot.reactor.netty.NettyReactiveWebServerFactory;
 import org.springframework.boot.reactor.netty.NettyServerCustomizer;
 import org.springframework.boot.reactor.netty.autoconfigure.NettyReactiveWebServerFactoryCustomizer;
@@ -578,8 +579,9 @@ public AddResponseHeaderGatewayFilterFactory addResponseHeaderGatewayFilterFacto
 	@Bean
 	@ConditionalOnEnabledFilter
 	public ModifyRequestBodyGatewayFilterFactory modifyRequestBodyGatewayFilterFactory(
-			ServerCodecConfigurer codecConfigurer) {
-		return new ModifyRequestBodyGatewayFilterFactory(codecConfigurer.getReaders());
+			ServerCodecConfigurer codecConfigurer, ObjectProvider<CodecCustomizer> codecCustomizers) {
+		return new ModifyRequestBodyGatewayFilterFactory(codecConfigurer.getReaders(),
+				codecCustomizers.orderedStream().toList());
 	}
 
 	@Bean
@@ -592,8 +594,9 @@ public DedupeResponseHeaderGatewayFilterFactory dedupeResponseHeaderGatewayFilte
 	@ConditionalOnEnabledFilter
 	public ModifyResponseBodyGatewayFilterFactory modifyResponseBodyGatewayFilterFactory(
 			ServerCodecConfigurer codecConfigurer, Set<MessageBodyDecoder> bodyDecoders,
-			Set<MessageBodyEncoder> bodyEncoders) {
-		return new ModifyResponseBodyGatewayFilterFactory(codecConfigurer.getReaders(), bodyDecoders, bodyEncoders);
+			Set<MessageBodyEncoder> bodyEncoders, ObjectProvider<CodecCustomizer> codecCustomizers) {
+		return new ModifyResponseBodyGatewayFilterFactory(codecConfigurer.getReaders(), bodyDecoders, bodyEncoders,
+				codecCustomizers.orderedStream().toList());
 	}
 
 	@Bean
@@ -625,9 +628,9 @@ public RedirectToGatewayFilterFactory redirectToGatewayFilterFactory() {
 	@ConditionalOnEnabledFilter
 	public RemoveJsonAttributesResponseBodyGatewayFilterFactory removeJsonAttributesResponseBodyGatewayFilterFactory(
 			ServerCodecConfigurer codecConfigurer, Set<MessageBodyDecoder> bodyDecoders,
-			Set<MessageBodyEncoder> bodyEncoders) {
-		return new RemoveJsonAttributesResponseBodyGatewayFilterFactory(
-				new ModifyResponseBodyGatewayFilterFactory(codecConfigurer.getReaders(), bodyDecoders, bodyEncoders));
+			Set<MessageBodyEncoder> bodyEncoders, ObjectProvider<CodecCustomizer> codecCustomizers) {
+		return new RemoveJsonAttributesResponseBodyGatewayFilterFactory(new ModifyResponseBodyGatewayFilterFactory(
+				codecConfigurer.getReaders(), bodyDecoders, bodyEncoders, codecCustomizers.orderedStream().toList()));
 	}
 
 	@Bean

spring-cloud-gateway-server-webflux/src/main/java/org/springframework/cloud/gateway/config/GatewayFunctionAutoConfiguration.java

@@ -18,11 +18,13 @@
 
 import java.util.Set;
 
+import org.springframework.beans.factory.ObjectProvider;
 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.http.codec.CodecCustomizer;
 import org.springframework.boot.webflux.autoconfigure.HttpHandlerAutoConfiguration;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration;
@@ -45,8 +47,10 @@ class GatewayFunctionAutoConfiguration {
 	@ConditionalOnEnabledGlobalFilter
 	@ConditionalOnBean(FunctionCatalog.class)
 	public FunctionRoutingFilter functionRoutingFilter(FunctionCatalog functionCatalog,
-			ServerCodecConfigurer codecConfigurer, Set<MessageBodyEncoder> messageBodyEncoders) {
-		return new FunctionRoutingFilter(functionCatalog, codecConfigurer.getReaders(), messageBodyEncoders);
+			ServerCodecConfigurer codecConfigurer, Set<MessageBodyEncoder> messageBodyEncoders,
+			ObjectProvider<CodecCustomizer> codecCustomizers) {
+		return new FunctionRoutingFilter(functionCatalog, codecConfigurer.getReaders(), messageBodyEncoders,
+				codecCustomizers.orderedStream().toList());
 	}
 
 }

spring-cloud-gateway-server-webflux/src/main/java/org/springframework/cloud/gateway/filter/FunctionRoutingFilter.java

@@ -29,6 +29,7 @@
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
+import org.springframework.boot.http.codec.CodecCustomizer;
 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;
@@ -50,6 +51,7 @@
 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.client.ExchangeStrategies;
 import org.springframework.web.reactive.function.server.ServerRequest;
 import org.springframework.web.server.ServerWebExchange;
 
@@ -71,12 +73,20 @@ public class FunctionRoutingFilter implements GlobalFilter, Ordered {
 
 	private final Map<String, MessageBodyEncoder> messageBodyEncoders;
 
+	private final ExchangeStrategies exchangeStrategies;
+
 	public FunctionRoutingFilter(FunctionCatalog functionCatalog, List<HttpMessageReader<?>> messageReaders,
 			Set<MessageBodyEncoder> messageBodyEncoders) {
+		this(functionCatalog, messageReaders, messageBodyEncoders, List.of());
+	}
+
+	public FunctionRoutingFilter(FunctionCatalog functionCatalog, List<HttpMessageReader<?>> messageReaders,
+			Set<MessageBodyEncoder> messageBodyEncoders, List<CodecCustomizer> codecCustomizers) {
 		this.functionCatalog = functionCatalog;
 		this.messageReaders = messageReaders;
 		this.messageBodyEncoders = messageBodyEncoders.stream()
 			.collect(Collectors.toMap(MessageBodyEncoder::encodingType, identity()));
+		this.exchangeStrategies = BodyInserterContext.buildExchangeStrategies(codecCustomizers);
 	}
 
 	@Override
@@ -149,7 +159,7 @@ protected Mono<Void> processRequest(ServerWebExchange exchange, FunctionInvocati
 			CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange,
 					exchange.getResponse().getHeaders());
 
-			return bodyInserter.insert(outputMessage, new BodyInserterContext()).then(Mono.defer(() -> {
+			return bodyInserter.insert(outputMessage, new BodyInserterContext(exchangeStrategies)).then(Mono.defer(() -> {
 				ServerHttpResponse response = exchange.getResponse();
 				Mono<DataBuffer> messageBody = writeBody(response, outputMessage, outClass);
 				HttpHeaders responseHeaders = response.getHeaders();

spring-cloud-gateway-server-webflux/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyRequestBodyGatewayFilterFactory.java

@@ -24,6 +24,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.boot.http.codec.CodecCustomizer;
 import org.springframework.cloud.gateway.filter.GatewayFilter;
 import org.springframework.cloud.gateway.filter.GatewayFilterChain;
 import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
@@ -37,6 +38,7 @@
 import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
 import org.springframework.web.reactive.function.BodyInserter;
 import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.ExchangeStrategies;
 import org.springframework.web.reactive.function.server.HandlerStrategies;
 import org.springframework.web.reactive.function.server.ServerRequest;
 import org.springframework.web.server.ServerWebExchange;
@@ -55,14 +57,21 @@ public class ModifyRequestBodyGatewayFilterFactory
 
 	private final List<HttpMessageReader<?>> messageReaders;
 
+	private final ExchangeStrategies exchangeStrategies;
+
 	public ModifyRequestBodyGatewayFilterFactory() {
-		super(Config.class);
-		this.messageReaders = HandlerStrategies.withDefaults().messageReaders();
+		this(HandlerStrategies.withDefaults().messageReaders(), List.of());
 	}
 
 	public ModifyRequestBodyGatewayFilterFactory(List<HttpMessageReader<?>> messageReaders) {
+		this(messageReaders, List.of());
+	}
+
+	public ModifyRequestBodyGatewayFilterFactory(List<HttpMessageReader<?>> messageReaders,
+			List<CodecCustomizer> codecCustomizers) {
 		super(Config.class);
 		this.messageReaders = messageReaders;
+		this.exchangeStrategies = BodyInserterContext.buildExchangeStrategies(codecCustomizers);
 	}
 
 	@Override
@@ -98,7 +107,7 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 					headers.set(HttpHeaders.CONTENT_TYPE, config.getContentType());
 				}
 				CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, headers);
-				return bodyInserter.insert(outputMessage, new BodyInserterContext())
+				return bodyInserter.insert(outputMessage, new BodyInserterContext(exchangeStrategies))
 					// .log("modify_request", Level.INFO)
 					.then(Mono.defer(() -> {
 						ServerHttpRequest decorator = decorate(exchange, headers, outputMessage);

spring-cloud-gateway-server-webflux/src/main/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyResponseBodyGatewayFilterFactory.java

@@ -28,6 +28,7 @@
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
+import org.springframework.boot.http.codec.CodecCustomizer;
 import org.springframework.cloud.gateway.filter.GatewayFilter;
 import org.springframework.cloud.gateway.filter.GatewayFilterChain;
 import org.springframework.cloud.gateway.filter.NettyWriteResponseFilter;
@@ -46,6 +47,7 @@
 import org.springframework.web.reactive.function.BodyInserter;
 import org.springframework.web.reactive.function.BodyInserters;
 import org.springframework.web.reactive.function.client.ClientResponse;
+import org.springframework.web.reactive.function.client.ExchangeStrategies;
 import org.springframework.web.server.ServerWebExchange;
 
 import static java.util.function.Function.identity;
@@ -68,14 +70,23 @@ public class ModifyResponseBodyGatewayFilterFactory
 
 	private final List<HttpMessageReader<?>> messageReaders;
 
+	private final ExchangeStrategies exchangeStrategies;
+
 	public ModifyResponseBodyGatewayFilterFactory(List<HttpMessageReader<?>> messageReaders,
 			Set<MessageBodyDecoder> messageBodyDecoders, Set<MessageBodyEncoder> messageBodyEncoders) {
+		this(messageReaders, messageBodyDecoders, messageBodyEncoders, List.of());
+	}
+
+	public ModifyResponseBodyGatewayFilterFactory(List<HttpMessageReader<?>> messageReaders,
+			Set<MessageBodyDecoder> messageBodyDecoders, Set<MessageBodyEncoder> messageBodyEncoders,
+			List<CodecCustomizer> codecCustomizers) {
 		super(Config.class);
 		this.messageReaders = messageReaders;
 		this.messageBodyDecoders = messageBodyDecoders.stream()
 			.collect(Collectors.toMap(MessageBodyDecoder::encodingType, identity()));
 		this.messageBodyEncoders = messageBodyEncoders.stream()
 			.collect(Collectors.toMap(MessageBodyEncoder::encodingType, identity()));
+		this.exchangeStrategies = BodyInserterContext.buildExchangeStrategies(codecCustomizers);
 	}
 
 	@Override
@@ -237,7 +248,7 @@ public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
 			BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, outClass);
 			CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange,
 					exchange.getResponse().getHeaders());
-			return bodyInserter.insert(outputMessage, new BodyInserterContext()).then(Mono.defer(() -> {
+			return bodyInserter.insert(outputMessage, new BodyInserterContext(exchangeStrategies)).then(Mono.defer(() -> {
 				Mono<DataBuffer> messageBody = writeBody(getDelegate(), outputMessage, outClass);
 				HttpHeaders headers = getDelegate().getHeaders();
 				if (!headers.containsHeader(HttpHeaders.TRANSFER_ENCODING)

spring-cloud-gateway-server-webflux/src/main/java/org/springframework/cloud/gateway/support/BodyInserterContext.java

@@ -21,8 +21,10 @@
 import java.util.Map;
 import java.util.Optional;
 
+import org.springframework.boot.http.codec.CodecCustomizer;
 import org.springframework.http.codec.HttpMessageWriter;
 import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.util.ObjectUtils;
 import org.springframework.web.reactive.function.BodyInserter;
 import org.springframework.web.reactive.function.client.ExchangeStrategies;
 
@@ -35,7 +37,21 @@ public BodyInserterContext() {
 	}
 
 	public BodyInserterContext(ExchangeStrategies exchangeStrategies) {
-		this.exchangeStrategies = exchangeStrategies; // TODO: support custom strategies
+		this.exchangeStrategies = exchangeStrategies;
+	}
+
+	/**
+	 * Build an {@link ExchangeStrategies} instance and apply all registered
+	 * {@link CodecCustomizer CodecCustomizers}.
+	 */
+	public static ExchangeStrategies buildExchangeStrategies(List<CodecCustomizer> codecCustomizers) {
+		if (ObjectUtils.isEmpty(codecCustomizers)) {
+			return ExchangeStrategies.withDefaults();
+		}
+		ExchangeStrategies.Builder exchangeStrategiesBuilder = ExchangeStrategies.builder();
+		exchangeStrategiesBuilder
+			.codecs((codecs) -> codecCustomizers.forEach((customizer) -> customizer.customize(codecs)));
+		return exchangeStrategiesBuilder.build();
 	}
 
 	@Override

spring-cloud-gateway-server-webflux/src/test/java/org/springframework/cloud/gateway/filter/FunctionRoutingFilterTests.java

@@ -18,6 +18,7 @@
 
 import java.net.URI;
 import java.util.Locale;
+import java.util.Map;
 import java.util.function.Function;
 
 import org.junit.jupiter.api.Test;
@@ -28,6 +29,7 @@
 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.cloud.gateway.test.TestCodecCustomizerConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 import org.springframework.http.MediaType;
@@ -57,21 +59,42 @@ public void functionRoutingFilterWorks() {
 			.consumeWith(res -> assertThat(res.getResponseBody()).isEqualTo("HELLO"));
 	}
 
+	@Test
+	public void codecCustomizerWorksWithFunctionRouting() {
+		URI uri = UriComponentsBuilder.fromUriString(this.baseUri + "/").build(true).toUri();
+
+		testClient.post()
+			.uri(uri)
+			.bodyValue(Map.of("codec", "v1"))
+			.header("Host", "www.codeccustomizerworks.org")
+			.accept(MediaType.APPLICATION_JSON)
+			.exchange()
+			.expectBody(String.class)
+			.consumeWith(res -> assertThat(res.getResponseBody()).isEqualTo("{\"codec\":\"v2\"}"));
+	}
+
 	@EnableAutoConfiguration
 	@SpringBootConfiguration
-	@Import(DefaultTestConfig.class)
+	@Import({ DefaultTestConfig.class, TestCodecCustomizerConfiguration.class })
 	public static class TestConfig {
 
 		@Bean
 		Function<String, String> upper() {
 			return s -> s.toUpperCase(Locale.ROOT);
 		}
 
+		@Bean
+		Function<Object, Object> passThrough() {
+			return o -> o;
+		}
+
 		@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"))
+				.route("function_routing_codec_customizer_test",
+						r -> r.path("/").and().host("www.codeccustomizerworks.org").uri("fn://passThrough"))
 				.build();
 		}
 

spring-cloud-gateway-server-webflux/src/test/java/org/springframework/cloud/gateway/filter/factory/rewrite/ModifyRequestBodyGatewayFilterFactoryTests.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.gateway.filter.factory.rewrite;
 
 import java.util.Locale;
+import java.util.Map;
 
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
@@ -28,6 +29,7 @@
 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.cloud.gateway.test.TestCodecCustomizerConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 import org.springframework.core.ParameterizedTypeReference;
@@ -42,7 +44,7 @@
 /**
  * @author Junghoon Song
  */
-@SpringBootTest(webEnvironment = RANDOM_PORT, properties = "spring.http.codecs.max-in-memory-size=13")
+@SpringBootTest(webEnvironment = RANDOM_PORT, properties = "spring.http.codecs.max-in-memory-size=15")
 @DirtiesContext
 public class ModifyRequestBodyGatewayFilterFactoryTests extends BaseWebClientTests {
 
@@ -109,9 +111,26 @@ public void modifyRequestBodyParameterizedTypeReference() {
 			.isEqualTo("FOO_BAR_BAZ");
 	}
 
+	@Test
+	public void codecCustomizerWorksWithModifyRequestBody() {
+		testClient.post()
+			.uri("/post")
+			.header("Host", "www.codeccustomizerworks.org")
+			.header(HttpHeaders.CONTENT_TYPE, MediaType.TEXT_PLAIN_VALUE)
+			.body(BodyInserters.fromValue("request"))
+			.exchange()
+			.expectStatus()
+			.isEqualTo(HttpStatus.OK)
+			.expectBody()
+			.jsonPath("headers.Content-Type")
+			.isEqualTo(MediaType.APPLICATION_JSON_VALUE)
+			.jsonPath("data")
+			.isEqualTo("{\"codec\":\"v2\"}");
+	}
+
 	@EnableAutoConfiguration
 	@SpringBootConfiguration
-	@Import(DefaultTestConfig.class)
+	@Import({ DefaultTestConfig.class, TestCodecCustomizerConfiguration.class })
 	public static class TestConfig {
 
 		@Value("${test.uri}")
@@ -157,6 +176,14 @@ public RouteLocator testRouteLocator(RouteLocatorBuilder builder) {
 								return Mono.just(body.replaceAll(" ", "_").toUpperCase(Locale.ROOT));
 							}))
 							.uri(uri))
+				.route("modify_request_body_codec_customizer_test",
+						r -> r.order(-1)
+							.host("**.codeccustomizerworks.org")
+							.filters(f -> f.modifyRequestBody(String.class, Object.class,
+									MediaType.APPLICATION_JSON_VALUE, (serverWebExchange, body) -> {
+										return Mono.just(Map.of("codec", "v1"));
+									}))
+							.uri(uri))
 				.build();
 		}