Support server.compression with reactive servers

This commit adds support for HTTP compression with reactive servers,
with the following exceptions:

* `server.compression.mime-types` and
`server.compression.exclude-user-agents` are not supported by Reactor
Netty at the moment

* `server.compression.min-response-size` is only supported by Reactor
Netty right now, since other implementations rely on the
`"Content-Length"` HTTP response header to measure the response size
and most reactive responses are using `"Transfer-Encoding: chunked"`.

Closes gh-10782

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/web/embedded/jetty/JettyHandlerWrappers.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2012-2018 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
+ *
+ *      http://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.boot.web.embedded.jetty;
+
+import java.io.IOException;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.eclipse.jetty.http.HttpMethod;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.handler.HandlerWrapper;
+import org.eclipse.jetty.server.handler.gzip.GzipHandler;
+
+import org.springframework.boot.web.server.Compression;
+
+/**
+ * Jetty {@code HandlerWrapper} static factory.
+ *
+ * @author Brian Clozel
+ */
+final class JettyHandlerWrappers {
+
+	private JettyHandlerWrappers() {
+	}
+
+	static HandlerWrapper createGzipHandlerWrapper(Compression compression) {
+		GzipHandler handler = new GzipHandler();
+		handler.setMinGzipSize(compression.getMinResponseSize());
+		handler.setIncludedMimeTypes(compression.getMimeTypes());
+		for (HttpMethod httpMethod : HttpMethod.values()) {
+			handler.addIncludedMethods(httpMethod.name());
+		}
+		if (compression.getExcludedUserAgents() != null) {
+			handler.setExcludedAgentPatterns(compression.getExcludedUserAgents());
+		}
+		return handler;
+	}
+
+	static HandlerWrapper createServerHeaderHandlerWrapper(String header) {
+		return new ServerHeaderHandler(header);
+	}
+
+	/**
+	 * {@link HandlerWrapper} to add a custom {@code server} header.
+	 */
+	private static class ServerHeaderHandler extends HandlerWrapper {
+
+		private static final String SERVER_HEADER = "server";
+
+		private final String value;
+
+		ServerHeaderHandler(String value) {
+			this.value = value;
+		}
+
+		@Override
+		public void handle(String target, Request baseRequest, HttpServletRequest request,
+				HttpServletResponse response) throws IOException, ServletException {
+			if (!response.getHeaderNames().contains(SERVER_HEADER)) {
+				response.setHeader(SERVER_HEADER, this.value);
+			}
+			super.handle(target, baseRequest, request, response);
+		}
+
+	}
+}

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/web/embedded/jetty/JettyReactiveWebServerFactory.java

@@ -26,9 +26,11 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.eclipse.jetty.server.AbstractConnector;
 import org.eclipse.jetty.server.ConnectionFactory;
+import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.HttpConfiguration;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.handler.HandlerWrapper;
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.servlet.ServletHolder;
 import org.eclipse.jetty.util.thread.ThreadPool;
@@ -39,6 +41,7 @@ import org.springframework.boot.web.server.WebServer;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.JettyHttpHandlerAdapter;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * {@link ReactiveWebServerFactory} that can be used to create {@link JettyWebServer}s.
@@ -140,6 +143,7 @@ public class JettyReactiveWebServerFactory extends AbstractReactiveWebServerFact
 		ServletContextHandler contextHandler = new ServletContextHandler(server, "",
 				false, false);
 		contextHandler.addServlet(servletHolder, "/");
+		server.setHandler(addHandlerWrappers(contextHandler));
 		JettyReactiveWebServerFactory.logger
 				.info("Server initialized with port: " + port);
 		if (getSsl() != null && getSsl().isEnabled()) {
@@ -168,6 +172,23 @@ public class JettyReactiveWebServerFactory extends AbstractReactiveWebServerFact
 		return connector;
 	}
 
+	private Handler addHandlerWrappers(Handler handler) {
+		if (getCompression() != null && getCompression().getEnabled()) {
+			handler = applyWrapper(handler,
+					JettyHandlerWrappers.createGzipHandlerWrapper(getCompression()));
+		}
+		if (StringUtils.hasText(getServerHeader())) {
+			handler = applyWrapper(handler,
+					JettyHandlerWrappers.createServerHeaderHandlerWrapper(getServerHeader()));
+		}
+		return handler;
+	}
+
+	private Handler applyWrapper(Handler handler, HandlerWrapper wrapper) {
+		wrapper.setHandler(handler);
+		return wrapper;
+	}
+
 	private void customizeSsl(Server server, int port) {
 		new SslServerCustomizer(port, getSsl(), getSslStoreProvider(), getHttp2())
 				.customize(server);

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/web/embedded/jetty/JettyServletWebServerFactory.java

@@ -31,23 +31,16 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 
-import javax.servlet.ServletException;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
-
-import org.eclipse.jetty.http.HttpMethod;
 import org.eclipse.jetty.http.MimeTypes;
 import org.eclipse.jetty.server.AbstractConnector;
 import org.eclipse.jetty.server.ConnectionFactory;
 import org.eclipse.jetty.server.Connector;
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.HttpConfiguration;
-import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.server.handler.ErrorHandler;
 import org.eclipse.jetty.server.handler.HandlerWrapper;
-import org.eclipse.jetty.server.handler.gzip.GzipHandler;
 import org.eclipse.jetty.server.session.DefaultSessionCache;
 import org.eclipse.jetty.server.session.FileSessionDataStore;
 import org.eclipse.jetty.server.session.SessionHandler;
@@ -62,7 +55,6 @@ import org.eclipse.jetty.webapp.AbstractConfiguration;
 import org.eclipse.jetty.webapp.Configuration;
 import org.eclipse.jetty.webapp.WebAppContext;
 
-import org.springframework.boot.web.server.Compression;
 import org.springframework.boot.web.server.ErrorPage;
 import org.springframework.boot.web.server.MimeMappings;
 import org.springframework.boot.web.server.WebServer;
@@ -185,10 +177,12 @@ public class JettyServletWebServerFactory extends AbstractServletWebServerFactor
 
 	private Handler addHandlerWrappers(Handler handler) {
 		if (getCompression() != null && getCompression().getEnabled()) {
-			handler = applyWrapper(handler, createGzipHandler());
+			handler = applyWrapper(handler,
+					JettyHandlerWrappers.createGzipHandlerWrapper(getCompression()));
 		}
 		if (StringUtils.hasText(getServerHeader())) {
-			handler = applyWrapper(handler, new ServerHeaderHandler(getServerHeader()));
+			handler = applyWrapper(handler,
+					JettyHandlerWrappers.createServerHeaderHandlerWrapper(getServerHeader()));
 		}
 		return handler;
 	}
@@ -198,20 +192,6 @@ public class JettyServletWebServerFactory extends AbstractServletWebServerFactor
 		return wrapper;
 	}
 
-	private HandlerWrapper createGzipHandler() {
-		GzipHandler handler = new GzipHandler();
-		Compression compression = getCompression();
-		handler.setMinGzipSize(compression.getMinResponseSize());
-		handler.setIncludedMimeTypes(compression.getMimeTypes());
-		for (HttpMethod httpMethod : HttpMethod.values()) {
-			handler.addIncludedMethods(httpMethod.name());
-		}
-		if (compression.getExcludedUserAgents() != null) {
-			handler.setExcludedAgentPatterns(compression.getExcludedUserAgents());
-		}
-		return handler;
-	}
-
 	private void customizeSsl(Server server, int port) {
 		new SslServerCustomizer(port, getSsl(), getSslStoreProvider(), getHttp2())
 				.customize(server);
@@ -548,29 +528,6 @@ public class JettyServletWebServerFactory extends AbstractServletWebServerFactor
 		}
 	}
 
-	/**
-	 * {@link HandlerWrapper} to add a custom {@code server} header.
-	 */
-	private static class ServerHeaderHandler extends HandlerWrapper {
-
-		private static final String SERVER_HEADER = "server";
-
-		private final String value;
-
-		ServerHeaderHandler(String value) {
-			this.value = value;
-		}
-
-		@Override
-		public void handle(String target, Request baseRequest, HttpServletRequest request,
-				HttpServletResponse response) throws IOException, ServletException {
-			if (!response.getHeaderNames().contains(SERVER_HEADER)) {
-				response.setHeader(SERVER_HEADER, this.value);
-			}
-			super.handle(target, baseRequest, request, response);
-		}
-
-	}
 
 	private static final class LoaderHidingResource extends Resource {
 

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/web/embedded/netty/NettyReactiveWebServerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2017 the original author or authors.
+ * Copyright 2012-2018 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.
@@ -94,6 +94,9 @@ public class NettyReactiveWebServerFactory extends AbstractReactiveWebServerFact
 						getSsl(), getSslStoreProvider());
 				sslServerCustomizer.customize(options);
 			}
+			if (getCompression() != null && getCompression().getEnabled()) {
+				options.compression(getCompression().getMinResponseSize());
+			}
 			applyCustomizers(options);
 		}).build();
 	}

spring-boot-project/spring-boot/src/test/java/org/springframework/boot/web/reactive/server/AbstractReactiveWebServerFactoryTests.java

@@ -18,12 +18,19 @@ package org.springframework.boot.web.reactive.server;
 
 import java.io.File;
 import java.io.FileInputStream;
+import java.nio.charset.StandardCharsets;
 import java.security.KeyStore;
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.function.Consumer;
 
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLException;
 
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpResponse;
 import io.netty.handler.ssl.SslProvider;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import org.assertj.core.api.Assumptions;
@@ -33,14 +40,21 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.rules.TemporaryFolder;
 import reactor.core.publisher.Mono;
+import reactor.ipc.netty.NettyPipeline;
+import reactor.ipc.netty.http.client.HttpClientOptions;
 import reactor.test.StepVerifier;
 
 import org.springframework.boot.testsupport.rule.OutputCapture;
+import org.springframework.boot.web.embedded.netty.NettyReactiveWebServerFactory;
 import org.springframework.boot.web.embedded.undertow.UndertowReactiveWebServerFactory;
+import org.springframework.boot.web.server.Compression;
 import org.springframework.boot.web.server.Ssl;
 import org.springframework.boot.web.server.WebServer;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ServerHttpRequest;
@@ -232,14 +246,95 @@ public abstract class AbstractReactiveWebServerFactoryTests {
 	}
 
 	protected WebClient.Builder getWebClient() {
+		return getWebClient(options -> {
+		});
+	}
+
+	protected WebClient.Builder getWebClient(Consumer<? super HttpClientOptions.Builder> clientOptions) {
 		return WebClient.builder()
+				.clientConnector(new ReactorClientHttpConnector(clientOptions))
 				.baseUrl("http://localhost:" + this.webServer.getPort());
 	}
 
+	@Test
+	public void compressionOfResponseToGetRequest() throws Exception {
+		WebClient client = prepareCompressionTest();
+		ResponseEntity<Void> response = client.get()
+				.exchange().flatMap(res -> res.toEntity(Void.class)).block();
+		assertResponseIsCompressed(response);
+	}
+
+	@Test
+	public void compressionOfResponseToPostRequest() throws Exception {
+		WebClient client = prepareCompressionTest();
+		ResponseEntity<Void> response = client.post()
+				.exchange().flatMap(res -> res.toEntity(Void.class)).block();
+		assertResponseIsCompressed(response);
+	}
+
+	@Test
+	public void noCompressionForSmallResponse() throws Exception {
+		Assumptions.assumeThat(getFactory()).isInstanceOf(NettyReactiveWebServerFactory.class);
+		Compression compression = new Compression();
+		compression.setEnabled(true);
+		compression.setMinResponseSize(3001);
+		WebClient client = prepareCompressionTest(compression);
+		ResponseEntity<Void> response = client.get()
+				.exchange().flatMap(res -> res.toEntity(Void.class)).block();
+		assertResponseIsNotCompressed(response);
+	}
+
+	@Test
+	public void noCompressionForMimeType() throws Exception {
+		Assumptions.assumeThat(getFactory()).isNotInstanceOf(NettyReactiveWebServerFactory.class);
+		Compression compression = new Compression();
+		compression.setMimeTypes(new String[] {"application/json"});
+		WebClient client = prepareCompressionTest(compression);
+		ResponseEntity<Void> response = client.get()
+				.exchange().flatMap(res -> res.toEntity(Void.class)).block();
+		assertResponseIsNotCompressed(response);
+	}
+
+	@Test
+	public void noCompressionForUserAgent() throws Exception {
+		Assumptions.assumeThat(getFactory()).isNotInstanceOf(NettyReactiveWebServerFactory.class);
+		Compression compression = new Compression();
+		compression.setEnabled(true);
+		compression.setExcludedUserAgents(new String[] { "testUserAgent" });
+		WebClient client = prepareCompressionTest(compression);
+		ResponseEntity<Void> response = client.get().header("User-Agent", "testUserAgent")
+				.exchange().flatMap(res -> res.toEntity(Void.class)).block();
+		assertResponseIsNotCompressed(response);
+	}
+
+	protected WebClient prepareCompressionTest() {
+		Compression compression = new Compression();
+		compression.setEnabled(true);
+		return prepareCompressionTest(compression);
+
+	}
+	protected WebClient prepareCompressionTest(Compression compression) {
+		AbstractReactiveWebServerFactory factory = getFactory();
+		factory.setCompression(compression);
+		this.webServer = factory.getWebServer(new CharsHandler(3000, MediaType.TEXT_PLAIN));
+		this.webServer.start();
+		return getWebClient(options -> options.compression(true).afterChannelInit(channel -> {
+			channel.pipeline().addBefore(NettyPipeline.HttpDecompressor,
+					"CompressionTest", new CompressionDetectionHandler());
+		})).build();
+	}
+
+	protected void assertResponseIsCompressed(ResponseEntity<Void> response) {
+		assertThat(response.getHeaders().getFirst("X-Test-Compressed")).isEqualTo("true");
+	}
+
+	protected void assertResponseIsNotCompressed(ResponseEntity<Void> response) {
+		assertThat(response.getHeaders().keySet()).doesNotContain("X-Test-Compressed");
+	}
+
 	protected static class EchoHandler implements HttpHandler {
 
 		public EchoHandler() {
-
 		}
 
 		@Override
@@ -250,4 +345,43 @@ public abstract class AbstractReactiveWebServerFactoryTests {
 
 	}
 
+	protected static class CompressionDetectionHandler extends ChannelInboundHandlerAdapter {
+
+		@Override
+		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+			if (msg instanceof HttpResponse) {
+				HttpResponse response = (HttpResponse) msg;
+				boolean compressed = response.headers()
+						.contains(HttpHeaderNames.CONTENT_ENCODING, "gzip", true);
+				if (compressed) {
+					response.headers().set("X-Test-Compressed", "true");
+				}
+			}
+			ctx.fireChannelRead(msg);
+		}
+	}
+
+	protected static class CharsHandler implements HttpHandler {
+
+		private static final DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+
+		private final DataBuffer bytes;
+
+		private final MediaType mediaType;
+
+		public CharsHandler(int contentSize, MediaType mediaType) {
+			char[] chars = new char[contentSize];
+			Arrays.fill(chars, 'F');
+			this.bytes = factory.wrap(new String(chars).getBytes(StandardCharsets.UTF_8));
+			this.mediaType = mediaType;
+		}
+
+		@Override
+		public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
+			response.setStatusCode(HttpStatus.OK);
+			response.getHeaders().setContentType(this.mediaType);
+			return response.writeWith(Mono.just(this.bytes));
+		}
+	}
+
 }