Introduce BrotliDecoder (#10960)

Motivation:

Netty lacks client side support for decompressing Brotli compressed response bodies.

Modification:

* Introduce optional dependency to brotli4j by @hyperxpro. It will be up to the user to provide the brotli4j libraries for the target platform in the classpath. brotli4j is currently available for Linux, OSX and Windows, all for x86 only.
* Introduce BrotliDecoder in codec module
* Plug it onto `HttpContentDecompressor` for HTTP/1 and `DelegatingDecompressorFrameListener` for HTTP/2
* Add test in `HttpContentDecoderTest`
* Add `BrotliDecoderTest` that doesn't extend `AbstractDecoderTest` that looks flaky

Result:

Netty now support decompressing Brotli compressed response bodies.

codec-http/pom.xml

@@ -67,6 +67,30 @@
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
     </dependency>
+    <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.aayushatharva.brotli4j</groupId>
+      <artifactId>brotli4j</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.aayushatharva.brotli4j</groupId>
+      <artifactId>native-linux-x86_64</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.aayushatharva.brotli4j</groupId>
+      <artifactId>native-osx-x86_64</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.aayushatharva.brotli4j</groupId>
+      <artifactId>native-windows-x86_64</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>
 

codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecompressor.java

@@ -15,12 +15,15 @@
  */
 package io.netty.handler.codec.http;
 
+import static io.netty.handler.codec.http.HttpHeaderValues.BR;
 import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
 import static io.netty.handler.codec.http.HttpHeaderValues.GZIP;
 import static io.netty.handler.codec.http.HttpHeaderValues.X_DEFLATE;
 import static io.netty.handler.codec.http.HttpHeaderValues.X_GZIP;
 
 import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.compression.Brotli;
+import io.netty.handler.codec.compression.BrotliDecoder;
 import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.compression.ZlibWrapper;
 
@@ -64,6 +67,10 @@ public class HttpContentDecompressor extends HttpContentDecoder {
             return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
                     ctx.channel().config(), ZlibCodecFactory.newZlibDecoder(wrapper));
         }
+        if (Brotli.isAvailable() && BR.contentEqualsIgnoreCase(contentEncoding)) {
+            return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
+              ctx.channel().config(), new BrotliDecoder());
+        }
 
         // 'identity' or unsupported
         return null;

codec-http/src/main/java/io/netty/handler/codec/http/HttpHeaderValues.java

@@ -111,6 +111,10 @@ public final class HttpHeaderValues {
      * {@code "gzip"}
      */
     public static final AsciiString GZIP = AsciiString.cached("gzip");
+    /**
+     * {@code "br"}
+     */
+    public static final AsciiString BR = AsciiString.cached("br");
     /**
      * {@code "gzip,deflate"}
      */

codec-http/src/test/java/io/netty/handler/codec/http/HttpContentDecoderTest.java

@@ -23,14 +23,17 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.CodecException;
 import io.netty.handler.codec.DecoderException;
+import io.netty.handler.codec.compression.Brotli;
 import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.compression.ZlibDecoder;
 import io.netty.handler.codec.compression.ZlibEncoder;
 import io.netty.handler.codec.compression.ZlibWrapper;
 import io.netty.util.CharsetUtil;
 import io.netty.util.ReferenceCountUtil;
+import org.apache.commons.io.IOUtils;
 import org.junit.Test;
 
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Queue;
@@ -48,6 +51,29 @@ public class HttpContentDecoderTest {
             31, -117, 8, 8, 12, 3, -74, 84, 0, 3, 50, 0, -53, 72, -51, -55, -55,
             -41, 81, 40, -49, 47, -54, 73, 1, 0, 58, 114, -85, -1, 12, 0, 0, 0
     };
+    private static final String SAMPLE_STRING;
+    private static final byte[] SAMPLE_BZ_BYTES;
+
+    static {
+        InputStream uncompressed = HttpContentDecoderTest.class.getClassLoader()
+          .getResourceAsStream("sample.json");
+        try {
+            SAMPLE_STRING = IOUtils.toString(uncompressed, CharsetUtil.UTF_8);
+        } catch (Throwable e) {
+            throw new ExceptionInInitializerError(e);
+        } finally {
+            IOUtils.closeQuietly(uncompressed, null);
+        }
+        InputStream compressed = HttpContentDecoderTest.class.getClassLoader()
+          .getResourceAsStream("sample.json.br");
+        try {
+            SAMPLE_BZ_BYTES = IOUtils.toByteArray(compressed);
+        } catch (Throwable e) {
+            throw new ExceptionInInitializerError(e);
+        } finally {
+            IOUtils.closeQuietly(compressed, null);
+        }
+    }
 
     @Test
     public void testBinaryDecompression() throws Exception {
@@ -115,13 +141,13 @@ public class HttpContentDecoderTest {
         assertThat(ob1, is(instanceOf(DefaultHttpResponse.class)));
 
         Object ob2 = channel.readInbound();
-        assertThat(ob1, is(instanceOf(DefaultHttpResponse.class)));
+        assertThat(ob2, is(instanceOf(HttpContent.class)));
         HttpContent content = (HttpContent) ob2;
         assertEquals(HELLO_WORLD, content.content().toString(CharsetUtil.US_ASCII));
         content.release();
 
         Object ob3 = channel.readInbound();
-        assertThat(ob1, is(instanceOf(DefaultHttpResponse.class)));
+        assertThat(ob3, is(instanceOf(LastHttpContent.class)));
         LastHttpContent lastContent = (LastHttpContent) ob3;
         assertNotNull(lastContent.decoderResult());
         assertTrue(lastContent.decoderResult().isSuccess());
@@ -159,6 +185,81 @@ public class HttpContentDecoderTest {
         assertFalse(channel.finish()); // assert that no messages are left in channel
     }
 
+    @Test
+    public void testResponseBrotliDecompression() throws Throwable {
+        Brotli.ensureAvailability();
+        HttpResponseDecoder decoder = new HttpResponseDecoder();
+        HttpContentDecoder decompressor = new HttpContentDecompressor();
+        HttpObjectAggregator aggregator = new HttpObjectAggregator(Integer.MAX_VALUE);
+        EmbeddedChannel channel = new EmbeddedChannel(decoder, decompressor, aggregator);
+
+        String headers = "HTTP/1.1 200 OK\r\n" +
+          "Content-Length: " + SAMPLE_BZ_BYTES.length + "\r\n" +
+          "Content-Encoding: br\r\n" +
+          "\r\n";
+        ByteBuf buf = Unpooled.wrappedBuffer(headers.getBytes(CharsetUtil.US_ASCII), SAMPLE_BZ_BYTES);
+        assertTrue(channel.writeInbound(buf));
+
+        Object o = channel.readInbound();
+        assertThat(o, is(instanceOf(FullHttpResponse.class)));
+        FullHttpResponse resp = (FullHttpResponse) o;
+        assertNull("Content-Encoding header should be removed", resp.headers().get(HttpHeaderNames.CONTENT_ENCODING));
+        assertEquals("Content-Length header should match uncompressed string's length",
+          SAMPLE_STRING.length(),
+          resp.headers().getInt(HttpHeaderNames.CONTENT_LENGTH).intValue());
+        assertEquals("Response body should match uncompressed string",
+          SAMPLE_STRING,
+          resp.content().toString(CharsetUtil.UTF_8));
+        resp.release();
+
+        assertHasInboundMessages(channel, false);
+        assertHasOutboundMessages(channel, false);
+        assertFalse(channel.finish()); // assert that no messages are left in channel
+    }
+
+    @Test
+    public void testResponseChunksBrotliDecompression() throws Throwable {
+        Brotli.ensureAvailability();
+        HttpResponseDecoder decoder = new HttpResponseDecoder();
+        HttpContentDecoder decompressor = new HttpContentDecompressor();
+        HttpObjectAggregator aggregator = new HttpObjectAggregator(Integer.MAX_VALUE);
+        EmbeddedChannel channel = new EmbeddedChannel(decoder, decompressor, aggregator);
+
+        String headers = "HTTP/1.1 200 OK\r\n" +
+          "Content-Length: " + SAMPLE_BZ_BYTES.length + "\r\n" +
+          "Content-Encoding: br\r\n" +
+          "\r\n";
+
+        assertFalse(channel.writeInbound(Unpooled.wrappedBuffer(headers.getBytes(CharsetUtil.US_ASCII))));
+
+        int offset = 0;
+        while (offset < SAMPLE_BZ_BYTES.length) {
+            int len = Math.min(1500, SAMPLE_BZ_BYTES.length - offset);
+            boolean available = channel.writeInbound(Unpooled.wrappedBuffer(SAMPLE_BZ_BYTES, offset, len));
+            offset += 1500;
+            if (offset < SAMPLE_BZ_BYTES.length) {
+                assertFalse(available);
+            } else {
+                assertTrue(available);
+            }
+        }
+
+        Object o = channel.readInbound();
+        assertThat(o, is(instanceOf(FullHttpResponse.class)));
+        FullHttpResponse resp = (FullHttpResponse) o;
+        assertEquals("Content-Length header should match uncompressed string's length",
+          SAMPLE_STRING.length(),
+          resp.headers().getInt(HttpHeaderNames.CONTENT_LENGTH).intValue());
+        assertEquals("Response body should match uncompressed string",
+          SAMPLE_STRING,
+          resp.content().toString(CharsetUtil.UTF_8));
+        resp.release();
+
+        assertHasInboundMessages(channel, false);
+        assertHasOutboundMessages(channel, false);
+        assertFalse(channel.finish()); // assert that no messages are left in channel
+    }
+
     @Test
     public void testExpectContinueResponse1() {
         // request with header "Expect: 100-continue" must be replied with one "100 Continue" response

codec-http2/src/main/java/io/netty/handler/codec/http2/DelegatingDecompressorFrameListener.java

@@ -19,12 +19,15 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.compression.Brotli;
+import io.netty.handler.codec.compression.BrotliDecoder;
 import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.compression.ZlibWrapper;
 import io.netty.util.internal.UnstableApi;
 
 import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_ENCODING;
 import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
+import static io.netty.handler.codec.http.HttpHeaderValues.BR;
 import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
 import static io.netty.handler.codec.http.HttpHeaderValues.GZIP;
 import static io.netty.handler.codec.http.HttpHeaderValues.IDENTITY;
@@ -175,6 +178,10 @@ public class DelegatingDecompressorFrameListener extends Http2FrameListenerDecor
             return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
                     ctx.channel().config(), ZlibCodecFactory.newZlibDecoder(wrapper));
         }
+        if (Brotli.isAvailable() && BR.contentEqualsIgnoreCase(contentEncoding)) {
+            return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
+              ctx.channel().config(), new BrotliDecoder());
+        }
         // 'identity' or unsupported
         return null;
     }

codec/pom.xml

@@ -78,6 +78,26 @@
       <artifactId>lzma-java</artifactId>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>com.aayushatharva.brotli4j</groupId>
+      <artifactId>brotli4j</artifactId>
+      <optional>true</optional>
+    </dependency>
+    <dependency>
+      <groupId>com.aayushatharva.brotli4j</groupId>
+      <artifactId>native-linux-x86_64</artifactId>
+      <optional>true</optional>
+    </dependency>
+    <dependency>
+      <groupId>com.aayushatharva.brotli4j</groupId>
+      <artifactId>native-osx-x86_64</artifactId>
+      <optional>true</optional>
+    </dependency>
+    <dependency>
+      <groupId>com.aayushatharva.brotli4j</groupId>
+      <artifactId>native-windows-x86_64</artifactId>
+      <optional>true</optional>
+    </dependency>
 
     <dependency>
       <groupId>org.mockito</groupId>

codec/src/main/java/io/netty/handler/codec/compression/Brotli.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.netty.handler.codec.compression;
+
+import com.aayushatharva.brotli4j.Brotli4jLoader;
+import io.netty.util.internal.PlatformDependent;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+
+public final class Brotli {
+
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(Brotli.class);
+    private static final ClassNotFoundException CNFE;
+
+    static {
+        ClassNotFoundException cnfe = null;
+
+        try {
+            Class.forName("com.aayushatharva.brotli4j.Brotli4jLoader", false,
+                PlatformDependent.getClassLoader(Brotli.class));
+        } catch (ClassNotFoundException t) {
+            cnfe = t;
+            logger.debug(
+                "brotli4j not in the classpath; Brotli support will be unavailable.");
+        }
+
+        CNFE = cnfe;
+
+        // If in the classpath, try to load the native library and initialize brotli4j.
+        if (cnfe == null) {
+            Throwable cause = Brotli4jLoader.getUnavailabilityCause();
+            if (cause != null) {
+                logger.debug("Failed to load brotli4j; Brotli support will be unavailable.", cause);
+            }
+        }
+    }
+
+    /**
+     *
+     * @return true when brotli4j is in the classpath
+     * and native library is available on this platform and could be loaded
+     */
+    public static boolean isAvailable() {
+        return CNFE == null && Brotli4jLoader.isAvailable();
+    }
+
+    /**
+     * Throws when brotli support is missing from the classpath or is unavailable on this platform
+     * @throws Throwable a ClassNotFoundException if brotli4j is missing
+     * or a UnsatisfiedLinkError if brotli4j native lib can't be loaded
+     */
+    public static void ensureAvailability() throws Throwable {
+        if (CNFE != null) {
+            throw CNFE;
+        }
+        Brotli4jLoader.ensureAvailability();
+    }
+
+    private Brotli() {
+    }
+}

codec/src/main/java/io/netty/handler/codec/compression/BrotliDecoder.java

@@ -0,0 +1,173 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.netty.handler.codec.compression;
+
+import com.aayushatharva.brotli4j.decoder.DecoderJNI;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.util.internal.ObjectUtil;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+/**
+ * Uncompresses a {@link ByteBuf} encoded with the brotli format.
+ *
+ * See <a href="https://github.com/google/brotli">brotli</a>.
+ */
+public final class BrotliDecoder extends ByteToMessageDecoder {
+
+    private enum State {
+        DONE, NEEDS_MORE_INPUT, ERROR
+    }
+
+    static {
+        try {
+            Brotli.ensureAvailability();
+        } catch (Throwable throwable) {
+            throw new ExceptionInInitializerError(throwable);
+        }
+    }
+
+    private final int inputBufferSize;
+    private DecoderJNI.Wrapper decoder;
+    private boolean destroyed;
+
+    /**
+     * Creates a new BrotliDecoder with a default 8kB input buffer
+     */
+    public BrotliDecoder() {
+        this(8 * 1024);
+    }
+
+    /**
+     * Creates a new BrotliDecoder
+     * @param inputBufferSize desired size of the input buffer in bytes
+     */
+    public BrotliDecoder(int inputBufferSize) {
+        this.inputBufferSize = ObjectUtil.checkPositive(inputBufferSize, "inputBufferSize");
+    }
+
+    private ByteBuf pull(ByteBufAllocator alloc) {
+        ByteBuffer nativeBuffer = decoder.pull();
+        // nativeBuffer actually wraps brotli's internal buffer so we need to copy its content
+        ByteBuf copy = alloc.buffer(nativeBuffer.remaining());
+        copy.writeBytes(nativeBuffer);
+        return copy;
+    }
+
+    private State decompress(ChannelHandlerContext ctx, ByteBuf input, ByteBufAllocator alloc) {
+        for (;;) {
+            switch (decoder.getStatus()) {
+                case DONE:
+                    return State.DONE;
+
+                case OK:
+                    decoder.push(0);
+                    break;
+
+                case NEEDS_MORE_INPUT:
+                    if (decoder.hasOutput()) {
+                        ctx.fireChannelRead(pull(alloc));
+                    }
+
+                    if (!input.isReadable()) {
+                        return State.NEEDS_MORE_INPUT;
+                    }
+
+                    ByteBuffer decoderInputBuffer = decoder.getInputBuffer();
+                    decoderInputBuffer.clear();
+                    int readBytes = readBytes(input, decoderInputBuffer);
+                    decoder.push(readBytes);
+                    break;
+
+                case NEEDS_MORE_OUTPUT:
+                    ctx.fireChannelRead(pull(alloc));
+                    break;
+
+                default:
+                    return State.ERROR;
+            }
+        }
+    }
+
+    private static int readBytes(ByteBuf in, ByteBuffer dest) {
+        int limit = Math.min(in.readableBytes(), dest.remaining());
+        ByteBuffer slice = dest.slice();
+        slice.limit(limit);
+        in.readBytes(slice);
+        dest.position(dest.position() + limit);
+        return limit;
+    }
+
+    @Override
+    public void handlerAdded0(ChannelHandlerContext ctx) throws Exception {
+        decoder = new DecoderJNI.Wrapper(inputBufferSize);
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
+        if (destroyed) {
+            // Skip data received after finished.
+            in.skipBytes(in.readableBytes());
+            return;
+        }
+
+        if (!in.isReadable()) {
+            return;
+        }
+
+        try {
+            State state = decompress(ctx, in, ctx.alloc());
+            if (state == State.DONE) {
+                destroy();
+            } else if (state == State.ERROR) {
+                throw new DecompressionException("Brotli stream corrupted");
+            }
+        } catch (Exception e) {
+            destroy();
+            throw e;
+        }
+    }
+
+    private void destroy() {
+        if (!destroyed) {
+            destroyed = true;
+            decoder.destroy();
+        }
+    }
+
+    @Override
+    protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
+        try {
+            destroy();
+        } finally {
+            super.handlerRemoved0(ctx);
+        }
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        try {
+            destroy();
+        } finally {
+            super.channelInactive(ctx);
+        }
+    }
+}

codec/src/test/java/io/netty/handler/codec/compression/BrotliDecoderTest.java

@@ -0,0 +1,161 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.netty.handler.codec.compression;
+
+import com.aayushatharva.brotli4j.encoder.BrotliOutputStream;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.experimental.theories.DataPoints;
+import org.junit.experimental.theories.FromDataPoints;
+import org.junit.experimental.theories.Theories;
+import org.junit.experimental.theories.Theory;
+import org.junit.runner.RunWith;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Random;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Theories.class)
+public class BrotliDecoderTest {
+
+    private static final Random RANDOM;
+    private static final byte[] BYTES_SMALL = new byte[256];
+    private static final byte[] BYTES_LARGE = new byte[256 * 1024];
+    private static final ByteBuf WRAPPED_BYTES_SMALL;
+    private static final ByteBuf WRAPPED_BYTES_LARGE;
+    private static final byte[] COMPRESSED_BYTES_SMALL;
+    private static final byte[] COMPRESSED_BYTES_LARGE;
+
+    static {
+        try {
+            Brotli.ensureAvailability();
+            RANDOM = new Random();
+            fillArrayWithCompressibleData(BYTES_SMALL);
+            fillArrayWithCompressibleData(BYTES_LARGE);
+            WRAPPED_BYTES_SMALL = Unpooled.wrappedBuffer(BYTES_SMALL);
+            WRAPPED_BYTES_LARGE = Unpooled.wrappedBuffer(BYTES_LARGE);
+            COMPRESSED_BYTES_SMALL = compress(BYTES_SMALL);
+            COMPRESSED_BYTES_LARGE = compress(BYTES_LARGE);
+        } catch (Throwable throwable) {
+            throw new ExceptionInInitializerError(throwable);
+        }
+    }
+
+    private static void fillArrayWithCompressibleData(byte[] array) {
+        for (int i = 0; i < array.length; i++) {
+            array[i] = i % 4 != 0 ? 0 : (byte) RANDOM.nextInt();
+        }
+    }
+
+    private static byte[] compress(byte[] data) throws IOException {
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        BrotliOutputStream brotliOs = new BrotliOutputStream(os);
+        brotliOs.write(data);
+        brotliOs.close();
+        return os.toByteArray();
+    }
+
+    private EmbeddedChannel channel;
+
+    @Before
+    public void initChannel() {
+        channel = new EmbeddedChannel(new BrotliDecoder());
+    }
+
+    @After
+    public void destroyChannel() {
+        if (channel != null) {
+            channel.finishAndReleaseAll();
+            channel = null;
+        }
+    }
+
+    @DataPoints("smallData")
+    public static ByteBuf[] smallData() {
+        ByteBuf heap = Unpooled.wrappedBuffer(COMPRESSED_BYTES_SMALL);
+        ByteBuf direct = Unpooled.directBuffer(COMPRESSED_BYTES_SMALL.length);
+        direct.writeBytes(COMPRESSED_BYTES_SMALL);
+        return new ByteBuf[]{heap, direct};
+    }
+
+    @DataPoints("largeData")
+    public static ByteBuf[] largeData() {
+        ByteBuf heap = Unpooled.wrappedBuffer(COMPRESSED_BYTES_LARGE);
+        ByteBuf direct = Unpooled.directBuffer(COMPRESSED_BYTES_LARGE.length);
+        direct.writeBytes(COMPRESSED_BYTES_LARGE);
+        return new ByteBuf[]{heap, direct};
+    }
+
+    @Theory
+    public void testDecompressionOfSmallChunkOfData(@FromDataPoints("smallData") ByteBuf data) {
+        testDecompression(WRAPPED_BYTES_SMALL, data);
+    }
+
+    @Theory
+    public void testDecompressionOfLargeChunkOfData(@FromDataPoints("largeData") ByteBuf data) {
+        testDecompression(WRAPPED_BYTES_LARGE, data);
+    }
+
+    @Theory
+    public void testDecompressionOfBatchedFlowOfData(@FromDataPoints("largeData") ByteBuf data) {
+        testDecompressionOfBatchedFlow(WRAPPED_BYTES_LARGE, data);
+    }
+
+    private void testDecompression(final ByteBuf expected, final ByteBuf data) {
+        assertTrue(channel.writeInbound(data));
+
+        ByteBuf decompressed = readDecompressed(channel);
+        assertEquals(expected, decompressed);
+
+        decompressed.release();
+    }
+
+  private void testDecompressionOfBatchedFlow(final ByteBuf expected, final ByteBuf data) {
+        final int compressedLength = data.readableBytes();
+        int written = 0, length = RANDOM.nextInt(100);
+        while (written + length < compressedLength) {
+            ByteBuf compressedBuf = data.retainedSlice(written, length);
+            channel.writeInbound(compressedBuf);
+            written += length;
+            length = RANDOM.nextInt(100);
+        }
+        ByteBuf compressedBuf = data.slice(written, compressedLength - written);
+        assertTrue(channel.writeInbound(compressedBuf.retain()));
+
+        ByteBuf decompressedBuf = readDecompressed(channel);
+        assertEquals(expected, decompressedBuf);
+
+        decompressedBuf.release();
+        data.release();
+    }
+
+  private static ByteBuf readDecompressed(final EmbeddedChannel channel) {
+        CompositeByteBuf decompressed = Unpooled.compositeBuffer();
+        ByteBuf msg;
+        while ((msg = channel.readInbound()) != null) {
+            decompressed.addComponent(true, msg);
+        }
+        return decompressed;
+    }
+}

pom.xml

@@ -346,6 +346,7 @@
     <skipAutobahnTestsuite>false</skipAutobahnTestsuite>
     <skipHttp2Testsuite>false</skipHttp2Testsuite>
     <graalvm.version>19.2.1</graalvm.version>
+    <brotli4j.version>1.4.2</brotli4j.version>
     <skipJapicmp>true</skipJapicmp>
     <!-- By default skip native testsuite as it requires a custom environment with graalvm installed -->
     <skipNativeImageTestsuite>true</skipNativeImageTestsuite>
@@ -525,6 +526,26 @@
         <artifactId>lzma-java</artifactId>
         <version>1.3</version>
       </dependency>
+      <dependency>
+        <groupId>com.aayushatharva.brotli4j</groupId>
+        <artifactId>brotli4j</artifactId>
+        <version>${brotli4j.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.aayushatharva.brotli4j</groupId>
+        <artifactId>native-linux-x86_64</artifactId>
+        <version>${brotli4j.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.aayushatharva.brotli4j</groupId>
+        <artifactId>native-osx-x86_64</artifactId>
+        <version>${brotli4j.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.aayushatharva.brotli4j</groupId>
+        <artifactId>native-windows-x86_64</artifactId>
+        <version>${brotli4j.version}</version>
+      </dependency>
 
       <!-- Java concurrency tools for the JVM -->
       <dependency>
@@ -678,6 +699,14 @@
         <scope>test</scope>
       </dependency>
 
+      <!-- Test dependency for Brotli compression codec -->
+      <dependency>
+        <groupId>commons-io</groupId>
+        <artifactId>commons-io</artifactId>
+        <version>2.8.0</version>
+        <scope>test</scope>
+      </dependency>
+
       <!-- Test dependency used by http/2 hpack -->
       <dependency>
         <groupId>com.google.code.gson</groupId>