1
2
3
4
5
6
7
8
9
10
11
12
13
14
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
30
31
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
59
60 public BrotliDecoder() {
61 this(DEFAULT_INPUT_BUFFER_SIZE);
62 }
63
64
65
66
67
68 public BrotliDecoder(int inputBufferSize) {
69 this(inputBufferSize == 0 ? DEFAULT_INPUT_BUFFER_SIZE : inputBufferSize, DEFAULT_MAX_FORWARD_BYTES);
70 }
71
72
73
74
75
76
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
85
86
87
88
89
90
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
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
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
228 discardSomeReadBytes();
229
230 if (needsRead && !ctx.channel().config().isAutoRead()) {
231 ctx.read();
232 }
233 ctx.fireChannelReadComplete();
234 }
235 }