* added support for streaming response modification

Fixes spring-cloudgh-2275

Author: usrivastava92

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

@@ -21,7 +21,6 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import org.jetbrains.annotations.NotNull;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -54,19 +53,17 @@
 /**
  * GatewayFilter that modifies the response body.
  */
-public class ModifyResponseBodyGatewayFilterFactory extends
-		AbstractGatewayFilterFactory<ModifyResponseBodyGatewayFilterFactory.Config> {
+public class ModifyResponseBodyGatewayFilterFactory
+		extends AbstractGatewayFilterFactory<ModifyResponseBodyGatewayFilterFactory.Config> {
 
 	private final Map<String, MessageBodyDecoder> messageBodyDecoders;
 
 	private final Map<String, MessageBodyEncoder> messageBodyEncoders;
 
 	private final List<HttpMessageReader<?>> messageReaders;
 
-	public ModifyResponseBodyGatewayFilterFactory(
-			List<HttpMessageReader<?>> messageReaders,
-			Set<MessageBodyDecoder> messageBodyDecoders,
-			Set<MessageBodyEncoder> messageBodyEncoders) {
+	public ModifyResponseBodyGatewayFilterFactory(List<HttpMessageReader<?>> messageReaders,
+			Set<MessageBodyDecoder> messageBodyDecoders, Set<MessageBodyEncoder> messageBodyEncoders) {
 		super(Config.class);
 		this.messageReaders = messageReaders;
 		this.messageBodyDecoders = messageBodyDecoders.stream()
@@ -77,8 +74,7 @@ public ModifyResponseBodyGatewayFilterFactory(
 
 	@Override
 	public GatewayFilter apply(Config config) {
-		ModifyResponseGatewayFilter gatewayFilter = new ModifyResponseGatewayFilter(
-				config);
+		ModifyResponseGatewayFilter gatewayFilter = new ModifyResponseGatewayFilter(config);
 		gatewayFilter.setFactory(this);
 		return gatewayFilter;
 	}
@@ -226,8 +222,7 @@ public ModifyResponseGatewayFilter(Config config) {
 
 		@Override
 		public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
-			return chain.filter(exchange.mutate()
-					.response(new ModifiedServerHttpResponse(exchange, config)).build());
+			return chain.filter(exchange.mutate().response(new ModifiedServerHttpResponse(exchange, config)).build());
 		}
 
 		@Override
@@ -237,12 +232,11 @@ public int getOrder() {
 
 		@Override
 		public String toString() {
-			Object obj = (this.gatewayFilterFactory != null) ? this.gatewayFilterFactory
-					: this;
-			return filterToStringCreator(obj)
-					.append("New content type", config.getNewContentType())
-					.append("In class", config.getInClass())
-					.append("Out class", config.getOutClass()).toString();
+			Object obj = (this.gatewayFilterFactory != null) ? this.gatewayFilterFactory : this;
+			return filterToStringCreator(obj).append("New content type", config.getNewContentType())
+				.append("In class", config.getInClass())
+				.append("Out class", config.getOutClass())
+				.toString();
 		}
 
 		public void setFactory(GatewayFilterFactory<Config> gatewayFilterFactory) {
@@ -348,7 +342,7 @@ public Mono<Void> writeAndFlushWith(
 					}));
 		}
 
-		private @NotNull HttpHeaders prepareHttpHeaders() {
+		private HttpHeaders prepareHttpHeaders() {
 			String originalResponseContentType = exchange
 					.getAttribute(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR);
 			HttpHeaders httpHeaders = new HttpHeaders();
@@ -360,25 +354,20 @@ public Mono<Void> writeAndFlushWith(
 			return httpHeaders;
 		}
 
-		private ClientResponse prepareClientResponse(Publisher<? extends DataBuffer> body,
-				HttpHeaders httpHeaders) {
+		private ClientResponse prepareClientResponse(Publisher<? extends DataBuffer> body, HttpHeaders httpHeaders) {
 			ClientResponse.Builder builder;
-			builder = ClientResponse.create(exchange.getResponse().getStatusCode(),
-					messageReaders);
-			return builder.headers(headers -> headers.putAll(httpHeaders))
-					.body(Flux.from(body)).build();
+			builder = ClientResponse.create(exchange.getResponse().getStatusCode(), messageReaders);
+			return builder.headers(headers -> headers.putAll(httpHeaders)).body(Flux.from(body)).build();
 		}
 
-		private <T> Mono<T> extractBody(ServerWebExchange exchange,
-				ClientResponse clientResponse, Class<T> inClass) {
+		private <T> Mono<T> extractBody(ServerWebExchange exchange, ClientResponse clientResponse, Class<T> inClass) {
 			// if inClass is byte[] then just return body, otherwise check if
 			// decoding required
 			if (byte[].class.isAssignableFrom(inClass)) {
 				return clientResponse.bodyToMono(inClass);
 			}
 
-			List<String> encodingHeaders = exchange.getResponse().getHeaders()
-					.getOrEmpty(HttpHeaders.CONTENT_ENCODING);
+			List<String> encodingHeaders = exchange.getResponse().getHeaders().getOrEmpty(HttpHeaders.CONTENT_ENCODING);
 			for (String encoding : encodingHeaders) {
 				MessageBodyDecoder decoder = messageBodyDecoders.get(encoding);
 				if (decoder != null) {
@@ -394,15 +383,14 @@ private <T> Mono<T> extractBody(ServerWebExchange exchange,
 			return clientResponse.bodyToMono(inClass);
 		}
 
-		private Mono<DataBuffer> writeBody(ServerHttpResponse httpResponse,
-				CachedBodyOutputMessage message, Class<?> outClass) {
+		private Mono<DataBuffer> writeBody(ServerHttpResponse httpResponse, CachedBodyOutputMessage message,
+				Class<?> outClass) {
 			Mono<DataBuffer> response = DataBufferUtils.join(message.getBody());
 			if (byte[].class.isAssignableFrom(outClass)) {
 				return response;
 			}
 
-			List<String> encodingHeaders = httpResponse.getHeaders()
-					.getOrEmpty(HttpHeaders.CONTENT_ENCODING);
+			List<String> encodingHeaders = httpResponse.getHeaders().getOrEmpty(HttpHeaders.CONTENT_ENCODING);
 			for (String encoding : encodingHeaders) {
 				MessageBodyEncoder encoder = messageBodyEncoders.get(encoding);
 				if (encoder != null) {