Repository: netty/netty (Java)

Issue:
Broken BrotliDecoder
### Expected behavior
Transparent stream compression
### Actual behavior
<details>
  <summary>io.netty.handler.codec.compression.DecompressionException: Brotli stream corrupted</summary>

```
io.netty.handler.codec.compression.DecompressionException: Brotli stream corrupted
	at io.netty.handler.codec.compression.BrotliDecoder.decode(BrotliDecoder.java:141)
	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:491)
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:430)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:267)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
	at io.netty.handler.codec.ByteToMessageDecoder.handlerRemoved(ByteToMessageDecoder.java:238)
	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:496)
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:430)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:267)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
	at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:248)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
	at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
	at eu.exante.atp.protocol.tcpserver.pipeline.MetricsHandler.channelRead(MetricsHandler.scala:15)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
	at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1273)
	at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1084)
	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:491)
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:430)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:267)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
	at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
	at eu.exante.atp.protocol.tcpserver.pipeline.MetricsHandler.channelRead(MetricsHandler.scala:15)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
	at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1294)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:911)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
	at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:131)
	at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)
	at java.lang.Thread.run(Thread.java:750)

```
</details>

### Steps to reproduce
Just run example. It work with zlib, lz4 and others codecs. But not with brotli.

### Minimal yet complete reproducer code (or URL to code)
https://github.com/fshp/brotli_netty


### Netty version
4.1.84.Final
### JVM version (e.g. `java -version`)
openjdk version "11.0.8" 2020-07-14
OpenJDK Runtime Environment AdoptOpenJDK (build 11.0.8+10)
OpenJDK 64-Bit Server VM AdoptOpenJDK (build 11.0.8+10, mixed mode)



openjdk 17.0.2 2022-01-18
OpenJDK Runtime Environment (build 17.0.2+8-86)
OpenJDK 64-Bit Server VM (build 17.0.2+8-86, mixed mode, sharing)


java version "1.8.0_162"
Java(TM) SE Runtime Environment (build 1.8.0_162-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.162-b12, mixed mode)
### OS version (e.g. `uname -a`)
macOS 12.6 - Darwin host 21.6.0 Darwin Kernel Version 21.6.0: Mon Aug 22 20:17:10 PDT 2022; root:xnu-8020.140.49~2/RELEASE_X86_64 x86_64

Linux host 5.10.0-18-amd64 #1 SMP Debian 5.10.140-1 (2022-09-02) x86_64 GNU/Linux


Hints:
@hyperxpro  can you have a look ?
Thanks for the heads up. There is a major bug in Encoder.

PR coming soon.

Patch:
diff --git a/codec/src/main/java/io/netty/handler/codec/compression/BrotliEncoder.java b/codec/src/main/java/io/netty/handler/codec/compression/BrotliEncoder.java
--- a/codec/src/main/java/io/netty/handler/codec/compression/BrotliEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/compression/BrotliEncoder.java
@@ -15,30 +15,61 @@
  */
 package io.netty.handler.codec.compression;
 
+import com.aayushatharva.brotli4j.encoder.BrotliEncoderChannel;
 import com.aayushatharva.brotli4j.encoder.Encoder;
-import com.aayushatharva.brotli4j.encoder.Encoders;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.MessageToByteEncoder;
+import io.netty.util.AttributeKey;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.internal.ObjectUtil;
 
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.WritableByteChannel;
+
 /**
- * Compress a {@link ByteBuf} with the brotli format.
- *
+ * Compress a {@link ByteBuf} with the Brotli compression.
+ * <p>
  * See <a href="https://github.com/google/brotli">brotli</a>.
  */
 @ChannelHandler.Sharable
 public final class BrotliEncoder extends MessageToByteEncoder<ByteBuf> {
 
+    private static final AttributeKey<Writer> ATTR = AttributeKey.valueOf("BrotliEncoderWriter");
+
+    /**
+     * Encoder flush method is package-private, so we have to
+     * use reflection to call that method.
+     */
+    private static final Method FLUSH_METHOD;
+
+    static {
+        Method method;
+        try {
+            method = Encoder.class.getDeclaredMethod("flush");
+            method.setAccessible(true);
+        } catch (NoSuchMethodException e) {
+            throw new IllegalStateException(e);
+        }
+        FLUSH_METHOD = method;
+    }
+
     private final Encoder.Parameters parameters;
+    private final boolean isSharable;
+    private Writer writer;
 
     /**
-     * Create a new {@link BrotliEncoder} Instance
-     * with {@link Encoder.Parameters#setQuality(int)} set to 4
-     * and {@link Encoder.Parameters#setMode(Encoder.Mode)} set to {@link Encoder.Mode#TEXT}
+     * Create a new {@link BrotliEncoder} Instance with {@link BrotliOptions#DEFAULT}
+     * and {@link #isSharable()} set to {@code true}
      */
     public BrotliEncoder() {
         this(BrotliOptions.DEFAULT);
@@ -47,19 +78,59 @@ public BrotliEncoder() {
     /**
      * Create a new {@link BrotliEncoder} Instance
      *
-     * @param parameters {@link Encoder.Parameters} Instance
+     * @param brotliOptions {@link BrotliOptions} to use and
+     *                      {@link #isSharable()} set to {@code true}
      */
-    public BrotliEncoder(Encoder.Parameters parameters) {
-        this.parameters = ObjectUtil.checkNotNull(parameters, "Parameters");
+    public BrotliEncoder(BrotliOptions brotliOptions) {
+        this(brotliOptions.parameters());
     }
 
     /**
      * Create a new {@link BrotliEncoder} Instance
+     * and {@link #isSharable()} set to {@code true}
      *
-     * @param brotliOptions {@link BrotliOptions} to use.
+     * @param parameters {@link Encoder.Parameters} to use
      */
-    public BrotliEncoder(BrotliOptions brotliOptions) {
-        this(brotliOptions.parameters());
+    public BrotliEncoder(Encoder.Parameters parameters) {
+        this(parameters, true);
+    }
+
+    /**
+     * <p>
+     * Create a new {@link BrotliEncoder} Instance and specify
+     * whether this instance will be shared with multiple pipelines or not.
+     * </p>
+     *
+     * If {@link #isSharable()} is true then on {@link #handlerAdded(ChannelHandlerContext)} call,
+     * a new {@link Writer} will create, and it will be mapped using {@link Channel#attr(AttributeKey)}
+     * so {@link BrotliEncoder} can be shared with multiple pipelines. This works fine but there on every
+     * {@link #encode(ChannelHandlerContext, ByteBuf, ByteBuf)} call, we have to get the {@link Writer} associated
+     * with the appropriate channel. And this will add a overhead. So it is recommended to set {@link #isSharable()}
+     * to {@code false} and create new {@link BrotliEncoder} instance for every pipeline.
+     *
+     * @param parameters {@link Encoder.Parameters} to use
+     * @param isSharable Set to {@code true} if this instance is shared else set to {@code false}
+     */
+    public BrotliEncoder(Encoder.Parameters parameters, boolean isSharable) {
+        this.parameters = ObjectUtil.checkNotNull(parameters, "Parameters");
+        this.isSharable = isSharable;
+    }
+
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+        Writer writer = new Writer(parameters, ctx);
+        if (isSharable) {
+            ctx.channel().attr(ATTR).set(writer);
+        } else {
+            this.writer = writer;
+        }
+        super.handlerAdded(ctx);
+    }
+
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+        finish(ctx);
+        super.handlerRemoved(ctx);
     }
 
     @Override
@@ -69,23 +140,145 @@ protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throw
 
     @Override
     protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, ByteBuf msg, boolean preferDirect) throws Exception {
-        // If ByteBuf is unreadable, then return EMPTY_BUFFER.
         if (!msg.isReadable()) {
             return Unpooled.EMPTY_BUFFER;
         }
 
-        try {
-            ByteBuf out;
+        Writer writer;
+        if (isSharable) {
+            writer = ctx.channel().attr(ATTR).get();
+        } else {
+            writer = this.writer;
+        }
+
+        // If Writer is 'null' then Writer is not open.
+        if (writer == null) {
+            return Unpooled.EMPTY_BUFFER;
+        } else {
+            writer.encode(msg, preferDirect);
+            return writer.writableBuffer;
+        }
+    }
+
+    @Override
+    public boolean isSharable() {
+        return isSharable;
+    }
+
+    /**
+     * Finish the encoding, close streams and write final {@link ByteBuf} to the channel.
+     *
+     * @param ctx {@link ChannelHandlerContext} which we want to close
+     * @throws IOException If an error occurred during closure
+     */
+    public void finish(ChannelHandlerContext ctx) throws IOException {
+        Writer writer;
+
+        if (isSharable) {
+            writer = ctx.channel().attr(ATTR).getAndSet(null);
+        } else {
+            writer = this.writer;
+        }
+
+        if (writer != null) {
+            writer.close();
+            this.writer = null;
+        }
+    }
+
+    /**
+     * {@link Writer} is the implementation of {@link WritableByteChannel} which encodes
+     * Brotli data and stores it into {@link ByteBuf}.
+     */
+    private static final class Writer implements WritableByteChannel {
+
+        private ByteBuf writableBuffer;
+        private final BrotliEncoderChannel brotliEncoderChannel;
+        private final ChannelHandlerContext ctx;
+        private boolean isClosed;
+
+        private Writer(Encoder.Parameters parameters, ChannelHandlerContext ctx) throws IOException {
+            brotliEncoderChannel = new BrotliEncoderChannel(this, parameters);
+            this.ctx = ctx;
+        }
+
+        private void encode(ByteBuf msg, boolean preferDirect) throws Exception {
+            try {
+                allocate(preferDirect);
+
+                // Compress data and flush it into Buffer.
+                //
+                // As soon as we call flush, Encoder will be triggered to write encoded
+                // data into WritableByteChannel.
+                //
+                // A race condition will not arise because one flush call to encoder will result
+                // in only 1 call at `write(ByteBuffer)`.
+                brotliEncoderChannel.write(msg.nioBuffer());
+                FLUSH_METHOD.invoke(brotliEncoderChannel);
+
+            } catch (Exception e) {
+                ReferenceCountUtil.release(msg);
+                throw e;
+            }
+        }
+
+        private void allocate(boolean preferDirect) {
             if (preferDirect) {
-                out = ctx.alloc().ioBuffer();
+                writableBuffer = ctx.alloc().ioBuffer();
             } else {
-                out = ctx.alloc().buffer();
+                writableBuffer = ctx.alloc().buffer();
+            }
+        }
+
+        @Override
+        public int write(ByteBuffer src) throws IOException {
+            if (!isOpen()) {
+                throw new ClosedChannelException();
+            }
+
+            return writableBuffer.writeBytes(src).readableBytes();
+        }
+
+        @Override
+        public boolean isOpen() {
+            return !isClosed;
+        }
+
+        @Override
+        public void close() {
+            final ChannelPromise promise = ctx.newPromise();
+
+            ctx.executor().execute(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        finish(promise);
+                    } catch (IOException ex) {
+                        promise.setFailure(new IllegalStateException("Failed to finish encoding", ex));
+                    }
+                }
+            });
+        }
+
+        public void finish(final ChannelPromise promise) throws IOException {
+            if (!isClosed) {
+                // Allocate a buffer and write last pending data.
+                allocate(true);
+
+                try {
+                    brotliEncoderChannel.close();
+                    isClosed = true;
+                } catch (Exception ex) {
+                    promise.setFailure(ex);
+
+                    // Since we have already allocated Buffer for close operation,
+                    // we will release that buffer to prevent memory leak.
+                    ReferenceCountUtil.release(writableBuffer);
+                    return;
+                }
+
+                ctx.writeAndFlush(writableBuffer, promise);
             }
-            Encoders.compress(msg, out, parameters);
-            return out;
-        } catch (Exception e) {
-            ReferenceCountUtil.release(msg);
-            throw e;
         }
     }
 }


Test patch:
diff --git a/codec/src/test/java/io/netty/handler/codec/compression/BrotliEncoderTest.java b/codec/src/test/java/io/netty/handler/codec/compression/BrotliEncoderTest.java
--- a/codec/src/test/java/io/netty/handler/codec/compression/BrotliEncoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/compression/BrotliEncoderTest.java
@@ -15,9 +15,6 @@
  */
 package io.netty.handler.codec.compression;
 
-import com.aayushatharva.brotli4j.decoder.Decoder;
-import com.aayushatharva.brotli4j.decoder.DecoderJNI;
-import com.aayushatharva.brotli4j.decoder.DirectDecompress;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
@@ -26,6 +23,9 @@
 
 public class BrotliEncoderTest extends AbstractEncoderTest {
 
+    private EmbeddedChannel ENCODER_CHANNEL;
+    private EmbeddedChannel DECODER_CHANNEL;
+
     @BeforeAll
     static void setUp() {
         try {
@@ -37,22 +37,34 @@ static void setUp() {
 
     @Override
     public EmbeddedChannel createChannel() {
-        return new EmbeddedChannel(new BrotliEncoder());
+        // Setup Encoder and Decoder
+        ENCODER_CHANNEL = new EmbeddedChannel(new BrotliEncoder());
+        DECODER_CHANNEL = new EmbeddedChannel(new BrotliDecoder());
+
+        // Return the main channel (Encoder)
+        return ENCODER_CHANNEL;
+    }
+
+    @Override
+    public void destroyChannel() {
+        ENCODER_CHANNEL.finishAndReleaseAll();
+        DECODER_CHANNEL.finishAndReleaseAll();
     }
 
     @Override
-    protected ByteBuf decompress(ByteBuf compressed, int originalLength) throws Exception {
-        byte[] compressedArray = new byte[compressed.readableBytes()];
-        compressed.readBytes(compressedArray);
-        compressed.release();
+    protected ByteBuf decompress(ByteBuf compressed, int originalLength) {
+        DECODER_CHANNEL.writeInbound(compressed);
+
+        ByteBuf aggregatedBuffer = Unpooled.buffer();
+        ByteBuf decompressed = DECODER_CHANNEL.readInbound();
+        while (decompressed != null) {
+            aggregatedBuffer.writeBytes(decompressed);
 
-        DirectDecompress decompress = Decoder.decompress(compressedArray);
-        if (decompress.getResultStatus() == DecoderJNI.Status.ERROR) {
-            throw new DecompressionException("Brotli stream corrupted");
+            decompressed.release();
+            decompressed = DECODER_CHANNEL.readInbound();
         }
 
-        byte[] decompressed = decompress.getDecompressedData();
-        return Unpooled.wrappedBuffer(decompressed);
+        return aggregatedBuffer;
     }
 
     @Override