Adds ability to cache body without converting to bytes (#1095)

Creates new Cache filter that is activated with retry filter is used.

Other filters can opt into caching if needed.

Uses netty factory and retainedSlice() for retry.

Only cache if retry filter is enabled

Fixes gh-982
Fixes gh-1064

pom.xml

@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-build</artifactId>
-		<version>2.1.4.BUILD-SNAPSHOT</version>
+		<version>2.1.6.BUILD-SNAPSHOT</version>
 		<relativePath/>
 	</parent>
 	<scm>

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

@@ -45,11 +45,13 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.cloud.gateway.actuate.GatewayControllerEndpoint;
 import org.springframework.cloud.gateway.filter.AdaptCachedBodyGlobalFilter;
+import org.springframework.cloud.gateway.filter.AlwaysRetainBodyGlobalFilter;
 import org.springframework.cloud.gateway.filter.ForwardPathFilter;
 import org.springframework.cloud.gateway.filter.ForwardRoutingFilter;
 import org.springframework.cloud.gateway.filter.GlobalFilter;
 import org.springframework.cloud.gateway.filter.NettyRoutingFilter;
 import org.springframework.cloud.gateway.filter.NettyWriteResponseFilter;
+import org.springframework.cloud.gateway.filter.RemoveCachedBodyFilter;
 import org.springframework.cloud.gateway.filter.RouteToRequestUrlFilter;
 import org.springframework.cloud.gateway.filter.WebsocketRoutingFilter;
 import org.springframework.cloud.gateway.filter.WeightCalculatorWebFilter;
@@ -261,6 +263,16 @@ public class GatewayAutoConfiguration {
 		return new AdaptCachedBodyGlobalFilter();
 	}
 
+	@Bean
+	public AlwaysRetainBodyGlobalFilter alwaysRetainBodyGlobalFilter() {
+		return new AlwaysRetainBodyGlobalFilter();
+	}
+
+	@Bean
+	public RemoveCachedBodyFilter removeCachedBodyFilter() {
+		return new RemoveCachedBodyFilter();
+	}
+
 	@Bean
 	public RouteToRequestUrlFilter routeToRequestUrlFilter() {
 		return new RouteToRequestUrlFilter();
@@ -588,6 +600,9 @@ public class GatewayAutoConfiguration {
 				});
 			}
 
+			//TODO: add configuration to turn on wiretap
+			//httpClient = httpClient.wiretap(true);
+
 			return httpClient;
 		}
 

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/event/EnableBodyCachingEvent.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2013-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.event;
+
+import org.springframework.context.ApplicationEvent;
+
+public class EnableBodyCachingEvent extends ApplicationEvent {
+
+	private final String routeId;
+
+	public EnableBodyCachingEvent(Object source, String routeId) {
+		super(source);
+		this.routeId = routeId;
+	}
+
+	public String getRouteId() {
+		return this.routeId;
+	}
+
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/AlwaysRetainBodyGlobalFilter.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2013-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.filter;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.event.EnableBodyCachingEvent;
+import org.springframework.cloud.gateway.route.Route;
+import org.springframework.context.ApplicationListener;
+import org.springframework.core.Ordered;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.NettyDataBuffer;
+import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
+import org.springframework.web.server.ServerWebExchange;
+
+import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR;
+
+public class AlwaysRetainBodyGlobalFilter
+		implements GlobalFilter, Ordered, ApplicationListener<EnableBodyCachingEvent> {
+
+	private static final Log log = LogFactory.getLog(AlwaysRetainBodyGlobalFilter.class);
+
+	private ConcurrentMap<String, Boolean> routesToCache = new ConcurrentHashMap<>();
+
+	/**
+	 * Request body cache key.
+	 */
+	public static final String ALWAYS_CACHE_REQUEST_BODY_KEY = "alwaysCacheRequestBody";
+
+	@Override
+	public void onApplicationEvent(EnableBodyCachingEvent event) {
+		this.routesToCache.putIfAbsent(event.getRouteId(), true);
+	}
+
+	@Override
+	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
+		Object body = exchange.getAttributeOrDefault(ALWAYS_CACHE_REQUEST_BODY_KEY, null);
+		Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
+
+		if (body != null || !this.routesToCache.containsKey(route.getId())) {
+			return chain.filter(exchange);
+		}
+
+		return DataBufferUtils.join(exchange.getRequest().getBody())
+				.flatMap(dataBuffer -> {
+					if (dataBuffer.readableByteCount() > 0) {
+						if (log.isTraceEnabled()) {
+							log.trace("retaining body in exchange attribute");
+						}
+						exchange.getAttributes().put(ALWAYS_CACHE_REQUEST_BODY_KEY,
+								dataBuffer);
+					}
+
+					ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(
+							exchange.getRequest()) {
+						@Override
+						public Flux<DataBuffer> getBody() {
+							return Mono.<DataBuffer>fromSupplier(() -> {
+								if (exchange.getAttributeOrDefault(
+										ALWAYS_CACHE_REQUEST_BODY_KEY, null) == null) {
+									// probably == downstream closed
+									return null;
+								}
+								// TODO: deal with Netty
+								NettyDataBuffer pdb = (NettyDataBuffer) dataBuffer;
+								return pdb.factory()
+										.wrap(pdb.getNativeBuffer().retainedSlice());
+							}).flux();
+						}
+					};
+					return chain.filter(exchange.mutate().request(decorator).build());
+				}).switchIfEmpty(chain.filter(exchange));
+	}
+
+	@Override
+	public int getOrder() {
+		return -10;
+	}
+
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/RemoveCachedBodyFilter.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2013-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.filter;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.Ordered;
+import org.springframework.core.io.buffer.PooledDataBuffer;
+import org.springframework.web.server.ServerWebExchange;
+
+import static org.springframework.cloud.gateway.filter.AlwaysRetainBodyGlobalFilter.ALWAYS_CACHE_REQUEST_BODY_KEY;
+
+public class RemoveCachedBodyFilter implements GlobalFilter, Ordered {
+
+	private static final Log log = LogFactory.getLog(RemoveCachedBodyFilter.class);
+
+	@Override
+	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
+		return chain.filter(exchange).doFinally(s -> {
+			PooledDataBuffer b = (PooledDataBuffer) exchange.getAttributes()
+					.remove(ALWAYS_CACHE_REQUEST_BODY_KEY);
+			if (b != null && b.isAllocated()) {
+				if (log.isTraceEnabled()) {
+					log.trace("releasing cached body in exchange attribute");
+				}
+				b.release();
+			}
+		});
+	}
+
+	@Override
+	public int getOrder() {
+		return HIGHEST_PRECEDENCE;
+	}
+
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/AbstractGatewayFilterFactory.java

@@ -17,6 +17,8 @@
 package org.springframework.cloud.gateway.filter.factory;
 
 import org.springframework.cloud.gateway.support.AbstractConfigurable;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 
 /**
  * This class is BETA and may be subject to change in a future release.
@@ -24,7 +26,9 @@ import org.springframework.cloud.gateway.support.AbstractConfigurable;
  * @param <C> {@link AbstractConfigurable} subtype
  */
 public abstract class AbstractGatewayFilterFactory<C> extends AbstractConfigurable<C>
-		implements GatewayFilterFactory<C> {
+		implements GatewayFilterFactory<C>, ApplicationEventPublisherAware {
+
+	private ApplicationEventPublisher publisher;
 
 	@SuppressWarnings("unchecked")
 	public AbstractGatewayFilterFactory() {
@@ -35,6 +39,15 @@ public abstract class AbstractGatewayFilterFactory<C> extends AbstractConfigurab
 		super(configClass);
 	}
 
+	protected ApplicationEventPublisher getPublisher() {
+		return this.publisher;
+	}
+
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
+		this.publisher = publisher;
+	}
+
 	public static class NameConfig {
 
 		private String name;

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/GatewayFilterFactory.java

@@ -20,6 +20,7 @@ import java.util.function.Consumer;
 
 import org.springframework.cloud.gateway.filter.GatewayFilter;
 import org.springframework.cloud.gateway.support.Configurable;
+import org.springframework.cloud.gateway.support.HasRouteId;
 import org.springframework.cloud.gateway.support.NameUtils;
 import org.springframework.cloud.gateway.support.ShortcutConfigurable;
 import org.springframework.http.server.reactive.ServerHttpRequest;
@@ -41,6 +42,12 @@ public interface GatewayFilterFactory<C> extends ShortcutConfigurable, Configura
 	String VALUE_KEY = "value";
 
 	// useful for javadsl
+	default GatewayFilter apply(String routeId, Consumer<C> consumer) {
+		C config = newConfig();
+		consumer.accept(config);
+		return apply(routeId, config);
+	}
+
 	default GatewayFilter apply(Consumer<C> consumer) {
 		C config = newConfig();
 		consumer.accept(config);
@@ -58,6 +65,14 @@ public interface GatewayFilterFactory<C> extends ShortcutConfigurable, Configura
 
 	GatewayFilter apply(C config);
 
+	default GatewayFilter apply(String routeId, C config) {
+		if (config instanceof HasRouteId) {
+			HasRouteId hasRouteId = (HasRouteId) config;
+			hasRouteId.setRouteId(routeId);
+		}
+		return apply(config);
+	}
+
 	default String name() {
 		// TODO: deal with proxys
 		return NameUtils.normalizeFilterFactoryName(getClass());

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactory.java

@@ -86,6 +86,8 @@ public class HystrixGatewayFilterFactory
 		return singletonList(NAME_KEY);
 	}
 
+	@Override
+	// TODO: make Config implement HasRouteId and remove this method.
 	public GatewayFilter apply(String routeId, Consumer<Config> consumer) {
 		Config config = newConfig();
 		consumer.accept(config);

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/RequestSizeGatewayFilterFactory.java

@@ -66,9 +66,11 @@ public class RequestSizeGatewayFilterFactory extends
 				Long currentRequestSize = Long.valueOf(contentLength);
 				if (currentRequestSize > requestSizeConfig.getMaxSize()) {
 					exchange.getResponse().setStatusCode(HttpStatus.PAYLOAD_TOO_LARGE);
-					exchange.getResponse().getHeaders().add("errorMessage",
-							getErrorMessage(currentRequestSize,
-									requestSizeConfig.getMaxSize()));
+					if (!exchange.getResponse().isCommitted()) {
+						exchange.getResponse().getHeaders().add("errorMessage",
+								getErrorMessage(currentRequestSize,
+										requestSizeConfig.getMaxSize()));
+					}
 					return exchange.getResponse().setComplete();
 				}
 			}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactory.java

@@ -33,7 +33,9 @@ import reactor.retry.RepeatContext;
 import reactor.retry.Retry;
 import reactor.retry.RetryContext;
 
+import org.springframework.cloud.gateway.event.EnableBodyCachingEvent;
 import org.springframework.cloud.gateway.filter.GatewayFilter;
+import org.springframework.cloud.gateway.support.HasRouteId;
 import org.springframework.cloud.gateway.support.TimeoutException;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
@@ -130,7 +132,7 @@ public class RetryGatewayFilterFactory
 					.retryMax(retryConfig.getRetries());
 		}
 
-		return apply(statusCodeRepeat, exceptionRetry);
+		return apply(retryConfig.getRouteId(), statusCodeRepeat, exceptionRetry);
 	}
 
 	public boolean exceedsMaxIterations(ServerWebExchange exchange,
@@ -145,7 +147,7 @@ public class RetryGatewayFilterFactory
 	}
 
 	public void reset(ServerWebExchange exchange) {
-		// TODO: what else to do to reset SWE?
+		// TODO: what else to do to reset exchange?
 		Set<String> addedHeaders = exchange.getAttributeOrDefault(
 				CLIENT_RESPONSE_HEADER_NAMES, Collections.emptySet());
 		addedHeaders
@@ -153,8 +155,18 @@ public class RetryGatewayFilterFactory
 		exchange.getAttributes().remove(GATEWAY_ALREADY_ROUTED_ATTR);
 	}
 
+	@Deprecated
 	public GatewayFilter apply(Repeat<ServerWebExchange> repeat,
 			Retry<ServerWebExchange> retry) {
+		return apply(null, repeat, retry);
+	}
+
+	public GatewayFilter apply(String routeId, Repeat<ServerWebExchange> repeat,
+			Retry<ServerWebExchange> retry) {
+		if (routeId != null && getPublisher() != null) {
+			// send an event to enable caching
+			getPublisher().publishEvent(new EnableBodyCachingEvent(this, routeId));
+		}
 		return (exchange, chain) -> {
 			trace("Entering retry-filter");
 
@@ -193,7 +205,9 @@ public class RetryGatewayFilterFactory
 	}
 
 	@SuppressWarnings("unchecked")
-	public static class RetryConfig {
+	public static class RetryConfig implements HasRouteId {
+
+		private String routeId;
 
 		private int retries = 3;
 
@@ -219,6 +233,16 @@ public class RetryGatewayFilterFactory
 			Assert.notEmpty(this.methods, "methods may not be empty");
 		}
 
+		@Override
+		public void setRouteId(String routeId) {
+			this.routeId = routeId;
+		}
+
+		@Override
+		public String getRouteId() {
+			return this.routeId;
+		}
+
 		public int getRetries() {
 			return retries;
 		}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/RouteDefinitionRouteLocator.java

@@ -147,7 +147,7 @@ public class RouteDefinitionRouteLocator
 				.replaceFilters(gatewayFilters).build();
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "Duplicates", "unchecked" })
 	private List<GatewayFilter> loadGatewayFilters(String id,
 			List<FilterDefinition> filterDefinitions) {
 		List<GatewayFilter> filters = filterDefinitions.stream().map(definition -> {
@@ -173,7 +173,7 @@ public class RouteDefinitionRouteLocator
 					factory.shortcutFieldPrefix(), definition.getName(), validator,
 					conversionService);
 
-			GatewayFilter gatewayFilter = factory.apply(configuration);
+			GatewayFilter gatewayFilter = factory.apply(id, configuration);
 			if (this.publisher != null) {
 				this.publisher.publishEvent(new FilterArgsEvent(this, id, properties));
 			}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java

@@ -474,8 +474,9 @@ public class GatewayFilterSpec extends UriSpec {
 	 * @return a {@link GatewayFilterSpec} that can be used to apply additional filters
 	 */
 	public GatewayFilterSpec retry(int retries) {
-		return filter(getBean(RetryGatewayFilterFactory.class)
-				.apply(retryConfig -> retryConfig.setRetries(retries)));
+		return filter(
+				getBean(RetryGatewayFilterFactory.class).apply(this.routeBuilder.getId(),
+						retryConfig -> retryConfig.setRetries(retries)));
 	}
 
 	/**
@@ -487,7 +488,8 @@ public class GatewayFilterSpec extends UriSpec {
 	 */
 	public GatewayFilterSpec retry(
 			Consumer<RetryGatewayFilterFactory.RetryConfig> retryConsumer) {
-		return filter(getBean(RetryGatewayFilterFactory.class).apply(retryConsumer));
+		return filter(getBean(RetryGatewayFilterFactory.class)
+				.apply(this.routeBuilder.getId(), retryConsumer));
 	}
 
 	/**
@@ -498,7 +500,9 @@ public class GatewayFilterSpec extends UriSpec {
 	 */
 	public GatewayFilterSpec retry(Repeat<ServerWebExchange> repeat,
 			Retry<ServerWebExchange> retry) {
-		return filter(getBean(RetryGatewayFilterFactory.class).apply(repeat, retry));
+		RetryGatewayFilterFactory filterFactory = getBean(
+				RetryGatewayFilterFactory.class);
+		return filter(filterFactory.apply(this.routeBuilder.getId(), repeat, retry));
 	}
 
 	/**

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/support/HasRouteId.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2013-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.support;
+
+public interface HasRouteId {
+
+	void setRouteId(String routeId);
+
+	String getRouteId();
+
+}

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/config/GatewayAutoConfigurationTests.java

@@ -37,7 +37,8 @@ public class GatewayAutoConfigurationTests {
 	@Test
 	public void noHiddenHttpMethodFilter() {
 		try (ConfigurableApplicationContext ctx = SpringApplication.run(
-				NoHiddenHttpMethodFilterConfig.class, "--spring.jmx.enabled=false")) {
+				NoHiddenHttpMethodFilterConfig.class, "--spring.jmx.enabled=false",
+				"--server.port=0")) {
 			assertThat(ctx.getEnvironment()
 					.getProperty("spring.webflux.hiddenmethod.filter.enabled"))
 							.isEqualTo("false");

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RequestSizeGatewayFilterFactoryTest.java

@@ -43,14 +43,14 @@ import static org.springframework.boot.test.context.SpringBootTest.WebEnvironmen
 @DirtiesContext
 public class RequestSizeGatewayFilterFactoryTest extends BaseWebClientTests {
 
-	private static final String responseMesssage = "Request size is larger than permissible limit. Request size is 6.0 MB "
-			+ "where permissible limit is 5.0 MB";
+	private static final String responseMesssage = "Request size is larger than permissible limit. Request size is . . "
+			+ "where permissible limit is .*";
 
 	@Test
 	public void setRequestSizeFilterWorks() {
-		testClient.get().uri("/headers").header("Host", "www.setrequestsize.org")
-				.header("content-length", "6000000").exchange().expectStatus()
-				.isEqualTo(HttpStatus.PAYLOAD_TOO_LARGE).expectHeader()
+		testClient.post().uri("/post").header("Host", "www.setrequestsize.org")
+				.header("content-length", "6").syncBody("123456").exchange()
+				.expectStatus().isEqualTo(HttpStatus.PAYLOAD_TOO_LARGE).expectHeader()
 				.valueMatches("errorMessage", responseMesssage);
 	}
 
@@ -67,7 +67,7 @@ public class RequestSizeGatewayFilterFactoryTest extends BaseWebClientTests {
 			return builder.routes()
 					.route("test_request_size",
 							r -> r.order(-1).host("**.setrequestsize.org")
-									.filters(f -> f.setRequestSize(5000000L)).uri(uri))
+									.filters(f -> f.setRequestSize(5L)).uri(uri))
 					.build();
 		}
 

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java

@@ -40,6 +40,7 @@ import org.springframework.cloud.netflix.ribbon.StaticServerList;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.test.annotation.DirtiesContext;
@@ -79,11 +80,10 @@ public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTest
 	}
 
 	@Test
-	// TODO: support post
 	public void retryFilterPost() {
-		testClient.post().uri("/retry?key=post").exchange().expectStatus()
-				.is5xxServerError();
-		// .expectBody(String.class).isEqualTo("3");
+		testClient.post().uri("/retry?key=post")
+				.header(HttpHeaders.HOST, "www.retryjava.org").syncBody("Hello")
+				.exchange().expectStatus().isOk().expectBody(String.class).isEqualTo("3");
 	}
 
 	@Test
@@ -145,7 +145,8 @@ public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTest
 			return builder.routes()
 					.route("retry_java", r -> r.host("**.retryjava.org")
 							.filters(f -> f.prefixPath("/httpbin")
-									.retry(config -> config.setRetries(2)))
+									.retry(config -> config.setRetries(2)
+											.setMethods(HttpMethod.POST, HttpMethod.GET)))
 							.uri(uri))
 					.route("retry_with_loadbalancer",
 							r -> r.host("**.retrywithloadbalancer.org")

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/FormIntegrationTests.java

@@ -21,17 +21,18 @@ import java.util.Map;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
 
 import org.springframework.boot.SpringBootConfiguration;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.client.TestRestTemplate;
 import org.springframework.context.annotation.Import;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.util.LinkedMultiValueMap;
@@ -42,7 +43,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
 import static org.springframework.cloud.gateway.test.TestUtils.getMap;
 import static org.springframework.http.MediaType.APPLICATION_FORM_URLENCODED;
-import static org.springframework.web.reactive.function.BodyExtractors.toMono;
 
 @RunWith(SpringRunner.class)
 @SpringBootTest(webEnvironment = RANDOM_PORT)
@@ -73,6 +73,8 @@ public class FormIntegrationTests extends BaseWebClientTests {
 	public void multipartFormDataWorks() {
 		ClassPathResource img = new ClassPathResource("1x1.png");
 
+		TestRestTemplate rest = new TestRestTemplate();
+
 		HttpHeaders headers = new HttpHeaders();
 		headers.setContentType(MediaType.IMAGE_PNG);
 
@@ -81,17 +83,14 @@ public class FormIntegrationTests extends BaseWebClientTests {
 		MultiValueMap<String, Object> parts = new LinkedMultiValueMap<>();
 		parts.add("imgpart", entity);
 
-		Mono<Map> result = webClient.post().uri("/post")
-				.contentType(MediaType.MULTIPART_FORM_DATA)
-				.body(BodyInserters.fromMultipartData(parts)).exchange()
-				.flatMap(response -> response.body(toMono(Map.class)));
+		ResponseEntity<Map> response = rest.postForEntity(baseUri + "/post", parts,
+				Map.class);
 
-		StepVerifier.create(result).consumeNextWith(map -> {
-			Map<String, Object> files = getMap(map, "files");
-			assertThat(files).containsKey("imgpart");
-			String file = (String) files.get("imgpart");
-			assertThat(file).startsWith("data:").contains(";base64,");
-		}).expectComplete().verify(DURATION);
+		assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK);
+		Map<String, Object> files = getMap(response.getBody(), "files");
+		assertThat(files).containsKey("imgpart");
+		String file = (String) files.get("imgpart");
+		assertThat(file).startsWith("data:").contains(";base64,");
 	}
 
 	@EnableAutoConfiguration

spring-cloud-gateway-dependencies/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<artifactId>spring-cloud-dependencies-parent</artifactId>
 		<groupId>org.springframework.cloud</groupId>
-		<version>2.1.4.BUILD-SNAPSHOT</version>
+		<version>2.1.6.BUILD-SNAPSHOT</version>
 		<relativePath/>
 	</parent>