Compression support in PipeliningServerHandler

gradle/libs.versions.toml

@@ -3,6 +3,7 @@ asm = "9.6"
 awaitility = "4.2.0"
 bcpkix = "1.70"
 blaze = "1.6.10"
+brotli4j = "1.13.0"
 caffeine = "2.9.3"
 compile-testing = "0.21.0"
 
@@ -165,6 +166,8 @@ bcpkix = { module = "org.bouncycastle:bcpkix-jdk15on", version.ref = "bcpkix" }
 
 blaze-persistence-core = { module = "com.blazebit:blaze-persistence-core-impl", version.ref = "blaze" }
 
+brotli4j = { module = "com.aayushatharva.brotli4j:brotli4j", version.ref = "brotli4j" }
+
 caffeine = { module = "com.github.ben-manes.caffeine:caffeine", version.ref = "caffeine" }
 
 compile-testing = { module = "com.google.testing.compile:compile-testing", version.ref = "compile-testing" }

http-server-netty/build.gradle

@@ -32,6 +32,7 @@ dependencies {
     compileOnly libs.managed.kotlin.stdlib
     compileOnly libs.managed.netty.transport.native.unix.common
     compileOnly libs.managed.netty.incubator.codec.http3
+    compileOnly libs.brotli4j
 
     testImplementation libs.jmh.core
     testAnnotationProcessor libs.jmh.generator.annprocess

http-server-netty/src/main/java/io/micronaut/http/server/netty/HttpCompressionStrategy.java

@@ -15,6 +15,7 @@
  */
 package io.micronaut.http.server.netty;
 
+import io.netty.handler.codec.compression.StandardCompressionOptions;
 import io.netty.handler.codec.http.HttpResponse;
 
 /**
@@ -36,6 +37,6 @@ public interface HttpCompressionStrategy {
      * @return The compression level (0-9)
      */
     default int getCompressionLevel() {
-        return 6;
+        return StandardCompressionOptions.gzip().compressionLevel();
     }
 }

http-server-netty/src/main/java/io/micronaut/http/server/netty/HttpPipelineBuilder.java

@@ -597,9 +597,6 @@ private void insertMicronautHandlers() {
                 channel.attr(CERTIFICATE_SUPPLIER_ATTRIBUTE.get()).set(sslHandler.findPeerCert());
             }
 
-            SmartHttpContentCompressor contentCompressor = new SmartHttpContentCompressor(embeddedServices.getHttpCompressionStrategy());
-            pipeline.addLast(ChannelPipelineCustomizer.HANDLER_HTTP_COMPRESSOR, contentCompressor);
-
             Optional<NettyServerWebSocketUpgradeHandler> webSocketUpgradeHandler = embeddedServices.getWebSocketUpgradeHandler(server);
             if (webSocketUpgradeHandler.isPresent()) {
                 pipeline.addLast(NettyServerWebSocketUpgradeHandler.COMPRESSION_HANDLER, new WebSocketServerCompressionHandler());
@@ -621,7 +618,9 @@ private void insertMicronautHandlers() {
             if (server.getServerConfiguration().isDualProtocol() && server.getServerConfiguration().isHttpToHttpsRedirect() && sslHandler == null) {
                 requestHandler = new HttpToHttpsRedirectHandler(routingInBoundHandler.conversionService, server.getServerConfiguration(), sslConfiguration, hostResolver);
             }
-            pipeline.addLast(ChannelPipelineCustomizer.HANDLER_MICRONAUT_INBOUND, new PipeliningServerHandler(requestHandler));
+            PipeliningServerHandler pipeliningServerHandler = new PipeliningServerHandler(requestHandler);
+            pipeliningServerHandler.setCompressionStrategy(embeddedServices.getHttpCompressionStrategy());
+            pipeline.addLast(ChannelPipelineCustomizer.HANDLER_MICRONAUT_INBOUND, pipeliningServerHandler);
         }
 
         /**

http-server-netty/src/main/java/io/micronaut/http/server/netty/handler/Compressor.java

@@ -0,0 +1,188 @@
+/*
+ * Copyright 2017-2023 original 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 io.micronaut.http.server.netty.handler;
+
+import io.micronaut.core.annotation.Nullable;
+import io.micronaut.http.server.netty.HttpCompressionStrategy;
+import io.netty.channel.ChannelHandler;
+import io.netty.handler.codec.compression.Brotli;
+import io.netty.handler.codec.compression.BrotliEncoder;
+import io.netty.handler.codec.compression.BrotliOptions;
+import io.netty.handler.codec.compression.DeflateOptions;
+import io.netty.handler.codec.compression.GzipOptions;
+import io.netty.handler.codec.compression.SnappyFrameEncoder;
+import io.netty.handler.codec.compression.SnappyOptions;
+import io.netty.handler.codec.compression.StandardCompressionOptions;
+import io.netty.handler.codec.compression.ZlibCodecFactory;
+import io.netty.handler.codec.compression.ZlibWrapper;
+import io.netty.handler.codec.compression.Zstd;
+import io.netty.handler.codec.compression.ZstdEncoder;
+import io.netty.handler.codec.compression.ZstdOptions;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpVersion;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+final class Compressor {
+    private final HttpCompressionStrategy strategy;
+    private final BrotliOptions brotliOptions;
+    private final GzipOptions gzipOptions;
+    private final DeflateOptions deflateOptions;
+    private final ZstdOptions zstdOptions;
+    private final SnappyOptions snappyOptions;
+
+    Compressor(HttpCompressionStrategy strategy) {
+        this.strategy = strategy;
+        // only use configured compression level for gzip and deflate, other algos have different semantics for the level
+        this.brotliOptions = Brotli.isAvailable() ? StandardCompressionOptions.brotli() : null;
+        GzipOptions stdGzip = StandardCompressionOptions.gzip();
+        this.gzipOptions = StandardCompressionOptions.gzip(strategy.getCompressionLevel(), stdGzip.windowBits(), stdGzip.memLevel());
+        DeflateOptions stdDeflate = StandardCompressionOptions.deflate();
+        this.deflateOptions = StandardCompressionOptions.deflate(strategy.getCompressionLevel(), stdDeflate.windowBits(), stdDeflate.memLevel());
+        this.zstdOptions = Zstd.isAvailable() ? StandardCompressionOptions.zstd() : null;
+        this.snappyOptions = StandardCompressionOptions.snappy();
+    }
+
+    @Nullable
+    ChannelHandler prepare(HttpRequest request, HttpResponse response) {
+        // from HttpContentEncoder: isPassthru
+        int code = response.status().code();
+        if (code < 200 || code == 204 || code == 304 ||
+            (request.method().equals(HttpMethod.HEAD) || (request.method().equals(HttpMethod.CONNECT) && code == 200)) ||
+            response.protocolVersion() == HttpVersion.HTTP_1_0) {
+            return null;
+        }
+        // special case for FHR to keep behavior identical to HttpContentEncoder
+        if (response instanceof FullHttpResponse fhr && !fhr.content().isReadable()) {
+            return null;
+        }
+        if (!strategy.shouldCompress(response)) {
+            return null;
+        }
+        if (response.headers().contains(HttpHeaderNames.CONTENT_ENCODING)) {
+            // already encoded
+            return null;
+        }
+        List<String> acceptEncoding = new ArrayList<>();
+        for (String s : request.headers().getAll(HttpHeaderNames.ACCEPT_ENCODING)) {
+            acceptEncoding.addAll(Arrays.asList(s.split(",")));
+        }
+        Algorithm encoding = determineEncoding(acceptEncoding);
+        if (encoding == null) {
+            return null;
+        }
+        response.headers().add(HttpHeaderNames.CONTENT_ENCODING, encoding.contentEncoding);
+        return switch (encoding) {
+            case BR -> makeBrotliEncoder();
+            case ZSTD -> new ZstdEncoder(zstdOptions.compressionLevel(), zstdOptions.blockSize(), zstdOptions.maxEncodeSize());
+            case SNAPPY -> new SnappyFrameEncoder();
+            case GZIP -> ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP, gzipOptions.compressionLevel(), gzipOptions.windowBits(), gzipOptions.memLevel());
+            case DEFLATE -> ZlibCodecFactory.newZlibEncoder(ZlibWrapper.ZLIB, deflateOptions.compressionLevel(), deflateOptions.windowBits(), deflateOptions.memLevel());
+        };
+    }
+
+    private BrotliEncoder makeBrotliEncoder() {
+        return new BrotliEncoder(brotliOptions.parameters());
+    }
+
+    @SuppressWarnings("FloatingPointEquality")
+    private Algorithm determineEncoding(List<String> acceptEncoding) {
+        // from HttpContentCompressor, slightly modified
+        float starQ = -1.0f;
+        float brQ = -1.0f;
+        float zstdQ = -1.0f;
+        float snappyQ = -1.0f;
+        float gzipQ = -1.0f;
+        float deflateQ = -1.0f;
+        for (String encoding : acceptEncoding) {
+            float q = 1.0f;
+            int equalsPos = encoding.indexOf('=');
+            if (equalsPos != -1) {
+                try {
+                    q = Float.parseFloat(encoding.substring(equalsPos + 1));
+                } catch (NumberFormatException e) {
+                    // Ignore encoding
+                    q = 0.0f;
+                }
+            }
+            if (encoding.contains("*")) {
+                starQ = q;
+            } else if (encoding.contains("br") && q > brQ) {
+                brQ = q;
+            } else if (encoding.contains("zstd") && q > zstdQ) {
+                zstdQ = q;
+            } else if (encoding.contains("snappy") && q > snappyQ) {
+                snappyQ = q;
+            } else if (encoding.contains("gzip") && q > gzipQ) {
+                gzipQ = q;
+            } else if (encoding.contains("deflate") && q > deflateQ) {
+                deflateQ = q;
+            }
+        }
+        if (brQ > 0.0f || zstdQ > 0.0f || snappyQ > 0.0f || gzipQ > 0.0f || deflateQ > 0.0f) {
+            if (brQ != -1.0f && brQ >= zstdQ && this.brotliOptions != null) {
+                return Algorithm.BR;
+            } else if (zstdQ != -1.0f && zstdQ >= snappyQ && this.zstdOptions != null) {
+                return Algorithm.ZSTD;
+            } else if (snappyQ != -1.0f && snappyQ >= gzipQ && this.snappyOptions != null) {
+                return Algorithm.SNAPPY;
+            } else if (gzipQ != -1.0f && gzipQ >= deflateQ && this.gzipOptions != null) {
+                return Algorithm.GZIP;
+            } else if (deflateQ != -1.0f && this.deflateOptions != null) {
+                return Algorithm.DEFLATE;
+            }
+        }
+        if (starQ > 0.0f) {
+            if (brQ == -1.0f && this.brotliOptions != null) {
+                return Algorithm.BR;
+            }
+            if (zstdQ == -1.0f && this.zstdOptions != null) {
+                return Algorithm.ZSTD;
+            }
+            if (snappyQ == -1.0f && this.snappyOptions != null) {
+                return Algorithm.SNAPPY;
+            }
+            if (gzipQ == -1.0f && this.gzipOptions != null) {
+                return Algorithm.GZIP;
+            }
+            if (deflateQ == -1.0f && this.deflateOptions != null) {
+                return Algorithm.DEFLATE;
+            }
+        }
+        return null;
+    }
+
+    private enum Algorithm {
+        BR(HttpHeaderValues.BR),
+        ZSTD(HttpHeaderValues.ZSTD),
+        SNAPPY(HttpHeaderValues.SNAPPY),
+        GZIP(HttpHeaderValues.GZIP),
+        DEFLATE(HttpHeaderValues.DEFLATE);
+
+        final CharSequence contentEncoding;
+
+        Algorithm(CharSequence contentEncoding) {
+            this.contentEncoding = contentEncoding;
+        }
+    }
+}