View Javadoc
1   /*
2    * Copyright 2021 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    *   https://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  
17  package io.netty.handler.codec.compression;
18  
19  import com.aayushatharva.brotli4j.decoder.DecoderJNI;
20  import io.netty.buffer.ByteBuf;
21  import io.netty.channel.ChannelHandlerContext;
22  import io.netty.handler.codec.ByteToMessageDecoder;
23  import io.netty.util.internal.ObjectUtil;
24  
25  import java.nio.ByteBuffer;
26  import java.util.List;
27  
28  /**
29   * Decompresses a {@link ByteBuf} encoded with the brotli format.
30   * <p>
31   * See <a href="https://github.com/google/brotli">brotli</a>.
32   */
33  public final class BrotliDecoder extends ByteToMessageDecoder {
34  
35      private static final int DEFAULT_MAX_FORWARD_BYTES = CompressionUtil.DEFAULT_MAX_FORWARD_BYTES;
36      private static final int DEFAULT_INPUT_BUFFER_SIZE = 8 * 1024;
37  
38      private enum State {
39          DONE, NEEDS_MORE_INPUT, ERROR
40      }
41  
42      static {
43          try {
44              Brotli.ensureAvailability();
45          } catch (Throwable throwable) {
46              throw new ExceptionInInitializerError(throwable);
47          }
48      }
49  
50      private final int inputBufferSize;
51      private final int outputBufferSize;
52      private DecoderJNI.Wrapper decoder;
53      private boolean destroyed;
54      private boolean needsRead;
55      private ByteBuf accumBuffer;
56  
57      /**
58       * Creates a new BrotliDecoder with a default 8kB input buffer
59       */
60      public BrotliDecoder() {
61          this(DEFAULT_INPUT_BUFFER_SIZE);
62      }
63  
64      /**
65       * Creates a new BrotliDecoder
66       * @param inputBufferSize desired size of the input buffer in bytes
67       */
68      public BrotliDecoder(int inputBufferSize) {
69          this(inputBufferSize == 0 ? DEFAULT_INPUT_BUFFER_SIZE : inputBufferSize, DEFAULT_MAX_FORWARD_BYTES);
70      }
71  
72      /**
73       * Creates a new BrotliDecoder
74       * @param inputBufferSize desired size of the input buffer in bytes
75       * @param outputBufferSize desired max size of the output buffer in bytes
76       *                         (produce multiple output buffers if exceeded)
77       */
78      public BrotliDecoder(int inputBufferSize, int outputBufferSize) {
79          this.inputBufferSize = ObjectUtil.checkPositive(inputBufferSize, "inputBufferSize");
80          this.outputBufferSize = ObjectUtil.checkPositive(outputBufferSize, "outputBufferSize");
81      }
82  
83      /**
84       * Creates a new {@link BrotliDecoder} that use the {@code maxAllocation}
85       * semantics: the supplied value bounds the size of the emitted decompressed chunks.
86       * The input buffer size stays at the decoder's default.
87       *
88       * @param maxAllocation maximum size, in bytes, of each decompressed output
89       *                      buffer forwarded downstream; if {@code 0}, the
90       *                      decoder's default output cap is used.
91       */
92      public static BrotliDecoder newDecoderWithMaxAllocation(int maxAllocation) {
93          ObjectUtil.checkPositiveOrZero(maxAllocation, "maxAllocation");
94          return maxAllocation > 0 ?
95                  new BrotliDecoder(DEFAULT_INPUT_BUFFER_SIZE, maxAllocation) :
96                  new BrotliDecoder();
97      }
98  
99      private void forwardOutput(ChannelHandlerContext ctx) {
100         ByteBuffer nativeBuffer = decoder.pull(outputBufferSize);
101         // nativeBuffer actually wraps brotli's internal buffer so we need to copy its content
102         int remaining = nativeBuffer.remaining();
103         if (accumBuffer == null) {
104             accumBuffer = ctx.alloc().buffer(remaining);
105         }
106         accumBuffer.writeBytes(nativeBuffer);
107         needsRead = false;
108         if (accumBuffer.readableBytes() >= outputBufferSize) {
109             ctx.fireChannelRead(accumBuffer);
110             accumBuffer = null;
111         }
112     }
113 
114     private void flushAccumBuffer(ChannelHandlerContext ctx) {
115         if (accumBuffer != null && accumBuffer.isReadable()) {
116             ctx.fireChannelRead(accumBuffer);
117         } else if (accumBuffer != null) {
118             accumBuffer.release();
119         }
120         accumBuffer = null;
121     }
122 
123     private State decompress(ChannelHandlerContext ctx, ByteBuf input) {
124         for (;;) {
125             switch (decoder.getStatus()) {
126                 case DONE:
127                     return State.DONE;
128 
129                 case OK:
130                     decoder.push(0);
131                     break;
132 
133                 case NEEDS_MORE_INPUT:
134                     while (decoder.hasOutput()) {
135                         forwardOutput(ctx);
136                     }
137 
138                     if (!input.isReadable()) {
139                         return State.NEEDS_MORE_INPUT;
140                     }
141 
142                     ByteBuffer decoderInputBuffer = decoder.getInputBuffer();
143                     decoderInputBuffer.clear();
144                     int readBytes = readBytes(input, decoderInputBuffer);
145                     decoder.push(readBytes);
146                     break;
147 
148                 case NEEDS_MORE_OUTPUT:
149                     forwardOutput(ctx);
150                     break;
151 
152                 default:
153                     return State.ERROR;
154             }
155         }
156     }
157 
158     private static int readBytes(ByteBuf in, ByteBuffer dest) {
159         int limit = Math.min(in.readableBytes(), dest.remaining());
160         ByteBuffer slice = dest.slice();
161         slice.limit(limit);
162         in.readBytes(slice);
163         dest.position(dest.position() + limit);
164         return limit;
165     }
166 
167     @Override
168     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
169         decoder = new DecoderJNI.Wrapper(inputBufferSize);
170     }
171 
172     @Override
173     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
174         needsRead = true;
175         if (destroyed) {
176             // Skip data received after finished.
177             in.skipBytes(in.readableBytes());
178             return;
179         }
180 
181         if (!in.isReadable()) {
182             return;
183         }
184 
185         try {
186             State state = decompress(ctx, in);
187             if (state == State.DONE) {
188                 destroy();
189             } else if (state == State.ERROR) {
190                 throw new DecompressionException("Brotli stream corrupted");
191             }
192         } catch (Exception e) {
193             destroy();
194             throw e;
195         } finally {
196             flushAccumBuffer(ctx);
197         }
198     }
199 
200     private void destroy() {
201         if (!destroyed) {
202             destroyed = true;
203             decoder.destroy();
204         }
205     }
206 
207     @Override
208     protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
209         try {
210             destroy();
211         } finally {
212             super.handlerRemoved0(ctx);
213         }
214     }
215 
216     @Override
217     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
218         try {
219             destroy();
220         } finally {
221             super.channelInactive(ctx);
222         }
223     }
224 
225     @Override
226     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
227         // Discard bytes of the cumulation buffer if needed.
228         discardSomeReadBytes();
229 
230         if (needsRead && !ctx.channel().config().isAutoRead()) {
231             ctx.read();
232         }
233         ctx.fireChannelReadComplete();
234     }
235 }