Updates Netty*Filter classes to use DefaultDataBuffer.

This avoids class cast exceptions when using MockServerHttpResponse.

Fixes gh-1491

It may also work for Tomcat, Jetty and Undertow.

See gh-145

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

@@ -20,6 +20,8 @@
 import java.time.Duration;
 import java.util.List;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelOption;
 import io.netty.handler.codec.http.DefaultHttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
@@ -37,6 +39,8 @@
 import org.springframework.cloud.gateway.route.Route;
 import org.springframework.cloud.gateway.support.TimeoutException;
 import org.springframework.core.Ordered;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBuffer;
 import org.springframework.core.io.buffer.NettyDataBuffer;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
@@ -140,9 +144,7 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 										+ connection.channel().id().asShortText()
 										+ ", inbound: " + exchange.getLogPrefix()));
 					}
-					return nettyOutbound.send(request.getBody()
-							.map(dataBuffer -> ((NettyDataBuffer) dataBuffer)
-									.getNativeBuffer()));
+					return nettyOutbound.send(request.getBody().map(this::getByteBuf));
 				}).responseConnection((res, connection) -> {
 
 					// Defer committing the response until all route filters have run
@@ -203,6 +205,20 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 		return responseFlux.then(chain.filter(exchange));
 	}
 
+	protected ByteBuf getByteBuf(DataBuffer dataBuffer) {
+		if (dataBuffer instanceof NettyDataBuffer) {
+			NettyDataBuffer buffer = (NettyDataBuffer) dataBuffer;
+			return buffer.getNativeBuffer();
+		}
+		// MockServerHttpResponse creates these
+		else if (dataBuffer instanceof DefaultDataBuffer) {
+			DefaultDataBuffer buffer = (DefaultDataBuffer) dataBuffer;
+			return Unpooled.wrappedBuffer(buffer.getNativeBuffer());
+		}
+		throw new IllegalArgumentException(
+				"Unable to handle DataBuffer of type " + dataBuffer.getClass());
+	}
+
 	private void setResponseStatus(HttpClientResponse clientResponse,
 			ServerHttpResponse response) {
 		HttpStatus status = HttpStatus.resolve(clientResponse.status().code());

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

@@ -18,14 +18,16 @@
 
 import java.util.List;
 
+import io.netty.buffer.ByteBuf;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.Connection;
 
 import org.springframework.core.Ordered;
-import org.springframework.core.io.buffer.NettyDataBuffer;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.http.MediaType;
 import org.springframework.http.server.reactive.ServerHttpResponse;
@@ -77,16 +79,12 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 					}
 					ServerHttpResponse response = exchange.getResponse();
 
-					// TODO: what if it's not netty
-					NettyDataBufferFactory factory = (NettyDataBufferFactory) response
-							.bufferFactory();
-
 					// TODO: needed?
-					final Flux<NettyDataBuffer> body = connection
+					final Flux<DataBuffer> body = connection
 							.inbound()
 							.receive()
 							.retain()
-							.map(factory::wrap);
+							.map(byteBuf -> wrap(byteBuf, response));
 
 					MediaType contentType = null;
 					try {
@@ -104,6 +102,22 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 		// @formatter:on
 	}
 
+	protected DataBuffer wrap(ByteBuf byteBuf, ServerHttpResponse response) {
+		if (response.bufferFactory() instanceof NettyDataBufferFactory) {
+			NettyDataBufferFactory factory = (NettyDataBufferFactory) response
+					.bufferFactory();
+			return factory.wrap(byteBuf);
+		}
+		// MockServerHttpResponse creates these
+		else if (response.bufferFactory() instanceof DefaultDataBufferFactory) {
+			DefaultDataBufferFactory factory = (DefaultDataBufferFactory) response
+					.bufferFactory();
+			return factory.wrap(byteBuf.nioBuffer());
+		}
+		throw new IllegalArgumentException(
+				"Unkown DataBufferFactory type " + response.bufferFactory().getClass());
+	}
+
 	private void cleanup(ServerWebExchange exchange) {
 		Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);
 		if (connection != null) {

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

@@ -0,0 +1,67 @@
+/*
+ * 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 org.hamcrest.Matchers;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+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.route.RouteLocator;
+import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder;
+import org.springframework.cloud.gateway.test.PermitAllSecurityConfiguration;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.web.reactive.server.WebTestClient;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+public class NettyRoutingFilterTests {
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Test
+	public void mockServerWorks() {
+		WebTestClient client = WebTestClient.bindToApplicationContext(this.context)
+				.build();
+		client.get().uri("/mockexample").exchange().expectStatus()
+				.value(Matchers.lessThan(500));
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	@Import(PermitAllSecurityConfiguration.class)
+	public static class TestConfig {
+
+		@Bean
+		public RouteLocator routes(RouteLocatorBuilder builder) {
+			return builder.routes()
+					.route(p -> p.path("/mockexample")
+							.filters(f -> f.prefixPath("/httpbin"))
+							.uri("http://example.com"))
+					.build();
+		}
+
+	}
+
+}