Merge branch '4.1.x' into 4.2.x

Author: ryanjbaxter

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

@@ -23,6 +23,7 @@
 import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
 import reactor.netty.Connection;
 
 import org.springframework.core.Ordered;
@@ -98,8 +99,12 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 					return (isStreamingMediaType(contentType)
 							? response.writeAndFlushWith(body.map(Flux::just))
 							: response.writeWith(body));
-				})).doOnCancel(() -> cleanup(exchange))
-				.doOnError(throwable -> cleanup(exchange));
+				}))
+				.doFinally(signalType -> {
+					if (signalType == SignalType.CANCEL || signalType == SignalType.ON_ERROR) {
+						cleanup(exchange);
+					}
+				});
 		// @formatter:on
 	}
 
@@ -116,12 +121,12 @@ else if (bufferFactory instanceof DefaultDataBufferFactory) {
 			byteBuf.release();
 			return buffer;
 		}
-		throw new IllegalArgumentException("Unkown DataBufferFactory type " + bufferFactory.getClass());
+		throw new IllegalArgumentException("Unknown DataBufferFactory type " + bufferFactory.getClass());
 	}
 
 	private void cleanup(ServerWebExchange exchange) {
 		Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);
-		if (connection != null && connection.channel().isActive()) {
+		if (connection != null) {
 			connection.dispose();
 		}
 	}