Adds BodyFilterFunctions.modifyRequestBody()

Adds MvcUtils.cacheAndReadBody(), MvcUtils.readBody() and MvcUtils.cacheAndReadBody(). This is generalized from GatewayRequestPredicates.readBody().

See gh-2949

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/common/MvcUtils.java

@@ -16,15 +16,24 @@
 
 package org.springframework.cloud.gateway.server.mvc.common;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
 import java.net.URI;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import org.springframework.context.ApplicationContext;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpInputMessage;
+import org.springframework.http.converter.HttpMessageConverter;
 import org.springframework.util.Assert;
+import org.springframework.util.StreamUtils;
 import org.springframework.web.context.WebApplicationContext;
 import org.springframework.web.servlet.function.ServerRequest;
 import org.springframework.web.servlet.support.RequestContextUtils;
@@ -77,6 +86,23 @@ public abstract class MvcUtils {
 		return "GatewayServerMvc." + attr;
 	}
 
+	public static <T> Optional<T> cacheAndReadBody(ServerRequest request, Class<T> toClass) {
+		ByteArrayInputStream rawBody = cacheBody(request);
+		return readBody(request, rawBody, toClass);
+	}
+
+	public static ByteArrayInputStream cacheBody(ServerRequest request) {
+		try {
+			byte[] bytes = StreamUtils.copyToByteArray(request.servletRequest().getInputStream());
+			ByteArrayInputStream body = new ByteArrayInputStream(bytes);
+			putAttribute(request, MvcUtils.CACHED_REQUEST_BODY_ATTR, body);
+			return body;
+		}
+		catch (IOException e) {
+			throw new UncheckedIOException(e);
+		}
+	}
+
 	public static String expand(ServerRequest request, String template) {
 		Assert.notNull(request, "request may not be null");
 		Assert.notNull(template, "template may not be null");
@@ -147,6 +173,24 @@ public abstract class MvcUtils {
 		}
 	}
 
+	@SuppressWarnings("unchecked")
+	public static <T> Optional<T> readBody(ServerRequest request, ByteArrayInputStream body, Class<T> toClass) {
+		try {
+			HttpInputMessage inputMessage = new ByteArrayInputMessage(request, body);
+			List<HttpMessageConverter<?>> httpMessageConverters = request.messageConverters();
+			for (HttpMessageConverter<?> messageConverter : httpMessageConverters) {
+				if (messageConverter.canRead(toClass, request.headers().contentType().orElse(null))) {
+					T convertedValue = (T) messageConverter.read((Class) toClass, inputMessage);
+					return Optional.of(convertedValue);
+				}
+			}
+		}
+		catch (IOException e) {
+			throw new UncheckedIOException(e);
+		}
+		return Optional.empty();
+	}
+
 	public static void setRouteId(ServerRequest request, String routeId) {
 		request.attributes().put(GATEWAY_ROUTE_ID_ATTR, routeId);
 		request.servletRequest().setAttribute(GATEWAY_ROUTE_ID_ATTR, routeId);
@@ -157,4 +201,27 @@ public abstract class MvcUtils {
 		request.servletRequest().setAttribute(GATEWAY_REQUEST_URL_ATTR, url);
 	}
 
+	private final static class ByteArrayInputMessage implements HttpInputMessage {
+
+		private final ServerRequest request;
+
+		private final ByteArrayInputStream body;
+
+		private ByteArrayInputMessage(ServerRequest request, ByteArrayInputStream body) {
+			this.request = request;
+			this.body = body;
+		}
+
+		@Override
+		public InputStream getBody() {
+			return body;
+		}
+
+		@Override
+		public HttpHeaders getHeaders() {
+			return request.headers().asHttpHeaders();
+		}
+
+	}
+
 }

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/filter/BeforeFilterFunctions.java

@@ -174,6 +174,11 @@ public abstract class BeforeFilterFunctions {
 		};
 	}
 
+	public static <T, R> Function<ServerRequest, ServerRequest> modifyRequestBody(Class<T> inClass, Class<R> outClass,
+			String newContentType, BodyFilterFunctions.RewriteFunction<T, R> rewriteFunction) {
+		return BodyFilterFunctions.modifyRequestBody(inClass, outClass, newContentType, rewriteFunction);
+	}
+
 	public static Function<ServerRequest, ServerRequest> prefixPath(String prefix) {
 		final UriTemplate uriTemplate = new UriTemplate(prefix);
 

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/filter/BodyFilterFunctions.java

@@ -17,7 +17,10 @@
 package org.springframework.cloud.gateway.server.mvc.filter;
 
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
 import java.net.InetSocketAddress;
 import java.net.URI;
 import java.security.Principal;
@@ -25,6 +28,7 @@ import java.time.Instant;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -39,17 +43,22 @@ import jakarta.servlet.http.Part;
 
 import org.springframework.cloud.gateway.server.mvc.common.MvcUtils;
 import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpOutputMessage;
+import org.springframework.http.MediaType;
 import org.springframework.http.converter.HttpMessageConverter;
 import org.springframework.http.server.PathContainer;
 import org.springframework.http.server.RequestPath;
 import org.springframework.util.MultiValueMap;
+import org.springframework.util.StringUtils;
 import org.springframework.validation.BindException;
 import org.springframework.web.bind.WebDataBinder;
 import org.springframework.web.servlet.function.ServerRequest;
 import org.springframework.web.servlet.function.ServerResponse;
 import org.springframework.web.util.UriBuilder;
 
+import static org.springframework.cloud.gateway.server.mvc.common.MvcUtils.cacheAndReadBody;
 import static org.springframework.cloud.gateway.server.mvc.common.MvcUtils.getAttribute;
 
 public abstract class BodyFilterFunctions {
@@ -61,26 +70,108 @@ public abstract class BodyFilterFunctions {
 		return request -> {
 			Object o = getAttribute(request, MvcUtils.CACHED_REQUEST_BODY_ATTR);
 			if (o instanceof ByteArrayInputStream body) {
-				ByteArrayServletInputStream inputStream = new ByteArrayServletInputStream(body);
-				HttpServletRequestWrapper wrapper = new HttpServletRequestWrapper(request.servletRequest()) {
-					@Override
-					public ServletInputStream getInputStream() {
-						return inputStream;
-					}
-				};
-
-				return new ServerRequestWrapper(request) {
-					@Override
-					public HttpServletRequest servletRequest() {
-						return wrapper;
-					}
-				};
+				return wrapRequest(request, body);
 			}
 
 			return request;
 		};
 	}
 
+	private static ServerRequestWrapper wrapRequest(ServerRequest request, byte[] body) {
+		return wrapRequest(request, new ByteArrayInputStream(body));
+	}
+
+	private static ServerRequestWrapper wrapRequest(ServerRequest request, ByteArrayInputStream body) {
+		ByteArrayServletInputStream inputStream = new ByteArrayServletInputStream(body);
+		HttpServletRequestWrapper wrapper = new HttpServletRequestWrapper(request.servletRequest()) {
+			@Override
+			public ServletInputStream getInputStream() {
+				return inputStream;
+			}
+		};
+
+		return new ServerRequestWrapper(request) {
+			@Override
+			public HttpServletRequest servletRequest() {
+				return wrapper;
+			}
+		};
+	}
+
+	@SuppressWarnings("unchecked")
+	public static <T, R> Function<ServerRequest, ServerRequest> modifyRequestBody(Class<T> inClass, Class<R> outClass,
+			String newContentType, RewriteFunction<T, R> rewriteFunction) {
+		return request -> cacheAndReadBody(request, inClass).map(body -> {
+			R convertedBody = rewriteFunction.apply(request, body);
+			// TODO: cache converted body
+
+			MediaType contentType = (StringUtils.hasText(newContentType)) ? MediaType.parseMediaType(newContentType)
+					: request.headers().contentType().orElse(null);
+
+			List<HttpMessageConverter<?>> httpMessageConverters = request.messageConverters();
+			for (HttpMessageConverter<?> messageConverter : httpMessageConverters) {
+				if (messageConverter.canWrite(outClass, contentType)) {
+					HttpHeaders headers = new HttpHeaders();
+					headers.putAll(request.headers().asHttpHeaders());
+
+					// the new content type will be computed by converter
+					// and then set in the request decorator
+					headers.remove(HttpHeaders.CONTENT_LENGTH);
+
+					// if the body is changing content types, set it here, to the
+					// bodyInserter
+					// will know about it
+					if (contentType != null) {
+						headers.setContentType(contentType);
+					}
+					try {
+						ByteArrayHttpOutputMessage outputMessage = new ByteArrayHttpOutputMessage(headers);
+						((HttpMessageConverter<R>) messageConverter).write(convertedBody, contentType, outputMessage);
+						ServerRequest modified = ServerRequest.from(request)
+								.headers(httpHeaders -> httpHeaders.putAll(headers)).build();
+						return wrapRequest(modified, outputMessage.getBytes());
+					}
+					catch (IOException e) {
+						throw new UncheckedIOException(e);
+					}
+				}
+			}
+
+			return request;
+		}).orElse(request);
+	}
+
+	private final static class ByteArrayHttpOutputMessage implements HttpOutputMessage {
+
+		private final HttpHeaders headers;
+
+		private final ByteArrayOutputStream body;
+
+		private ByteArrayHttpOutputMessage(HttpHeaders headers) {
+			this.headers = headers;
+			this.body = new ByteArrayOutputStream();
+		}
+
+		@Override
+		public OutputStream getBody() throws IOException {
+			return this.body;
+		}
+
+		@Override
+		public HttpHeaders getHeaders() {
+			return headers;
+		}
+
+		public byte[] getBytes() {
+			return this.body.toByteArray();
+		}
+
+	}
+
+	public interface RewriteFunction<T, R> extends BiFunction<ServerRequest, T, R> {
+
+	}
+
 	private static class ByteArrayServletInputStream extends ServletInputStream {
 
 		private final ByteArrayInputStream body;

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

@@ -57,4 +57,5 @@ public class GatewayDelegatingRouterFunction<T extends ServerResponse> implement
 	public String toString() {
 		return String.format("RouterFunction routeId=%s delegate=%s", routeId, delegate);
 	}
+
 }

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/predicate/GatewayRequestPredicates.java

@@ -16,9 +16,6 @@
 
 package org.springframework.cloud.gateway.server.mvc.predicate;
 
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
 import java.lang.reflect.Method;
 import java.time.ZonedDateTime;
 import java.util.Arrays;
@@ -42,14 +39,11 @@ import org.springframework.cloud.gateway.server.mvc.common.Shortcut;
 import org.springframework.cloud.gateway.server.mvc.common.WeightConfig;
 import org.springframework.core.io.Resource;
 import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpInputMessage;
 import org.springframework.http.HttpMethod;
-import org.springframework.http.converter.HttpMessageConverter;
 import org.springframework.http.server.PathContainer;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
-import org.springframework.util.StreamUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.web.cors.CorsUtils;
 import org.springframework.web.servlet.function.HandlerFunction;
@@ -63,6 +57,7 @@ import org.springframework.web.util.pattern.PathPatternParser;
 
 import static org.springframework.cloud.gateway.server.mvc.common.MvcUtils.GATEWAY_ROUTE_ID_ATTR;
 import static org.springframework.cloud.gateway.server.mvc.common.MvcUtils.WEIGHT_ATTR;
+import static org.springframework.cloud.gateway.server.mvc.common.MvcUtils.cacheAndReadBody;
 import static org.springframework.cloud.gateway.server.mvc.common.MvcUtils.getAttribute;
 import static org.springframework.cloud.gateway.server.mvc.common.MvcUtils.putAttribute;
 
@@ -165,9 +160,8 @@ public abstract class GatewayRequestPredicates {
 		return RequestPredicates.path(pattern);
 	}
 
-	@SuppressWarnings("unchecked")
 	public static <T> RequestPredicate readBody(Class<T> inClass, Predicate<T> predicate) {
-		return new ReadBodyPredicate(inClass, (Predicate<Object>) predicate);
+		return new ReadBodyPredicate<>(inClass, predicate);
 	}
 
 	/**
@@ -426,24 +420,25 @@ public abstract class GatewayRequestPredicates {
 
 	}
 
-	private static final class ReadBodyPredicate implements RequestPredicate {
+	private static final class ReadBodyPredicate<T> implements RequestPredicate {
 
-		private final Class toRead;
+		private final Class<T> toRead;
 
-		private final Predicate<Object> predicate;
+		private final Predicate<T> predicate;
 
-		<T> ReadBodyPredicate(Class toRead, Predicate<Object> predicate) {
+		ReadBodyPredicate(Class<T> toRead, Predicate<T> predicate) {
 			this.toRead = toRead;
 			this.predicate = predicate;
 		}
 
 		@Override
+		@SuppressWarnings("unchecked")
 		public boolean test(ServerRequest request) {
 			try {
 				Object cachedBody = getAttribute(request, READ_BODY_CACHE_OBJECT_KEY);
 
 				if (cachedBody != null) {
-					return predicate.test(cachedBody);
+					return predicate.test((T) cachedBody);
 				}
 			}
 			catch (ClassCastException e) {
@@ -453,34 +448,10 @@ public abstract class GatewayRequestPredicates {
 				}
 			}
 
-			try {
-				byte[] bytes = StreamUtils.copyToByteArray(request.servletRequest().getInputStream());
-				ByteArrayInputStream body = new ByteArrayInputStream(bytes);
-				putAttribute(request, MvcUtils.CACHED_REQUEST_BODY_ATTR, body);
-				HttpInputMessage inputMessage = new HttpInputMessage() {
-					@Override
-					public InputStream getBody() {
-						return body;
-					}
-
-					@Override
-					public HttpHeaders getHeaders() {
-						return request.headers().asHttpHeaders();
-					}
-				};
-				List<HttpMessageConverter<?>> httpMessageConverters = request.messageConverters();
-				for (HttpMessageConverter<?> messageConverter : httpMessageConverters) {
-					if (messageConverter.canRead(toRead, request.headers().contentType().orElse(null))) {
-						Object value = messageConverter.read(toRead, inputMessage);
-						putAttribute(request, READ_BODY_CACHE_OBJECT_KEY, value);
-						return predicate.test(value);
-					}
-				}
-			}
-			catch (IOException e) {
-				throw new RuntimeException(e);
-			}
-			return false;
+			return cacheAndReadBody(request, toRead).map(body -> {
+				putAttribute(request, READ_BODY_CACHE_OBJECT_KEY, body);
+				return predicate.test(body);
+			}).orElse(false);
 		}
 
 		@Override

spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/ServerMvcIntegrationTests.java

@@ -88,6 +88,7 @@ import static org.springframework.cloud.gateway.server.mvc.filter.BeforeFilterFu
 import static org.springframework.cloud.gateway.server.mvc.filter.BeforeFilterFunctions.adaptCachedBody;
 import static org.springframework.cloud.gateway.server.mvc.filter.BeforeFilterFunctions.fallbackHeaders;
 import static org.springframework.cloud.gateway.server.mvc.filter.BeforeFilterFunctions.mapRequestHeader;
+import static org.springframework.cloud.gateway.server.mvc.filter.BeforeFilterFunctions.modifyRequestBody;
 import static org.springframework.cloud.gateway.server.mvc.filter.BeforeFilterFunctions.preserveHost;
 import static org.springframework.cloud.gateway.server.mvc.filter.BeforeFilterFunctions.removeRequestParameter;
 import static org.springframework.cloud.gateway.server.mvc.filter.BeforeFilterFunctions.requestHeaderSize;
@@ -545,7 +546,6 @@ public class ServerMvcIntegrationTests {
 
 	@Test
 	public void readBodyWorks() {
-
 		Event messageEvent = new Event("message", "bar");
 
 		restClient.post().uri("/events").bodyValue(messageEvent).exchange().expectStatus().isOk().expectHeader()
@@ -560,6 +560,23 @@ public class ServerMvcIntegrationTests {
 
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void rewriteRequestBodyStringWorks() {
+		restClient.post().uri("/post").header("Host", "www.modifyrequestbodystring.org").bodyValue("hello").exchange()
+				.expectStatus().isOk().expectBody(Map.class)
+				.consumeWith(result -> assertThat(result.getResponseBody()).containsEntry("data", "HELLOHELLO"));
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void rewriteRequestBodyObjectWorks() {
+		restClient.post().uri("/post").header("Host", "www.modifyrequestbodyobject.org").bodyValue("hello world")
+				.exchange().expectStatus().isOk().expectBody(Map.class)
+				.consumeWith(result -> assertThat(result.getResponseBody()).containsEntry("data",
+						"{\"message\":\"HELLO WORLD\"}"));
+	}
+
 	@SpringBootConfiguration
 	@EnableAutoConfiguration
 	@LoadBalancerClient(name = "httpbin", configuration = TestLoadBalancerConfig.Httpbin.class)
@@ -1010,15 +1027,35 @@ public class ServerMvcIntegrationTests {
 
 		@Bean
 		public RouterFunction<ServerResponse> gatewayRouterFunctionsReadBodyPredicate() {
-			// @formatter:of
+			// @formatter:off
 			return route("testreadbodypredicate")
-					.POST("/events", readBody(Event.class, eventPredicate("message")), http()).before(
-							new LocalServerPortUriResolver())
-					.filter(setPath("/do/events")).before(adaptCachedBody()).build()
-					.and(route("testreadbodypredicate2")
-							.POST("/events", readBody(Event.class, eventPredicate("message.channel")), http())
-							.before(new LocalServerPortUriResolver()).filter(setPath("/do/events/channel"))
-							.before(adaptCachedBody()).build());
+					.POST("/events", readBody(Event.class, eventPredicate("message")), http())
+					.before(new LocalServerPortUriResolver())
+					.filter(setPath("/do/events"))
+					.before(adaptCachedBody())
+					.build().and(
+				route("testreadbodypredicate2")
+					.POST("/events", readBody(Event.class, eventPredicate("message.channel")), http())
+					.before(new LocalServerPortUriResolver())
+					.filter(setPath("/do/events/channel"))
+					.before(adaptCachedBody())
+					.build());
+			// @formatter:on
+		}
+
+		@Bean
+		public RouterFunction<ServerResponse> gatewayRouterFunctionsModifyRequestBody() {
+			// @formatter:off
+			return route("testmodifyrequestbodystring")
+					.POST("/post", host("**.modifyrequestbodystring.org"), http())
+					.before(new HttpbinUriResolver())
+					.before(modifyRequestBody(String.class, String.class, null, (request, s) -> s.toUpperCase() + s.toUpperCase()))
+					.build().and(
+				route("testmodifyrequestbodyobject")
+					.POST("/post", host("**.modifyrequestbodyobject.org"), http())
+					.before(new HttpbinUriResolver())
+					.before(modifyRequestBody(String.class, Hello.class, MediaType.APPLICATION_JSON_VALUE, (request, s) -> new Hello(s.toUpperCase())))
+					.build());
 			// @formatter:on
 		}
 
@@ -1038,6 +1075,10 @@ public class ServerMvcIntegrationTests {
 
 	}
 
+	protected record Hello(String message) {
+
+	}
+
 	protected record Event(String foo, String bar) {
 
 	}
@@ -1089,6 +1130,7 @@ public class ServerMvcIntegrationTests {
 	}
 
 	protected static class TestHandler implements HandlerFunction<ServerResponse> {
+
 		@Override
 		public ServerResponse handle(ServerRequest request) {
 			return ServerResponse.ok().body("Hello");
@@ -1098,6 +1140,7 @@ public class ServerMvcIntegrationTests {
 		public String toString() {
 			return "TestHandler Hello";
 		}
+
 	}
 
 }