1 /*
2 * Copyright 2014 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 package io.netty.handler.codec.compression;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.channel.ChannelHandlerContext;
20 import io.netty.handler.codec.ByteToMessageDecoder;
21 import io.netty.util.internal.ObjectUtil;
22 import net.jpountz.lz4.LZ4Exception;
23 import net.jpountz.lz4.LZ4Factory;
24 import net.jpountz.lz4.LZ4SafeDecompressor;
25
26 import java.nio.ByteBuffer;
27 import java.util.List;
28 import java.util.zip.Checksum;
29
30 import static io.netty.handler.codec.compression.Lz4Constants.BLOCK_TYPE_COMPRESSED;
31 import static io.netty.handler.codec.compression.Lz4Constants.BLOCK_TYPE_NON_COMPRESSED;
32 import static io.netty.handler.codec.compression.Lz4Constants.COMPRESSION_LEVEL_BASE;
33 import static io.netty.handler.codec.compression.Lz4Constants.DEFAULT_SEED;
34 import static io.netty.handler.codec.compression.Lz4Constants.HEADER_LENGTH;
35 import static io.netty.handler.codec.compression.Lz4Constants.MAGIC_NUMBER;
36 import static io.netty.handler.codec.compression.Lz4Constants.MAX_BLOCK_SIZE;
37
38 /**
39 * Uncompresses a {@link ByteBuf} encoded with the LZ4 format.
40 *
41 * See original <a href="https://github.com/Cyan4973/lz4">LZ4 Github project</a>
42 * and <a href="https://fastcompression.blogspot.ru/2011/05/lz4-explained.html">LZ4 block format</a>
43 * for full description.
44 *
45 * Since the original LZ4 block format does not contains size of compressed block and size of original data
46 * this encoder uses format like <a href="https://github.com/idelpivnitskiy/lz4-java">LZ4 Java</a> library
47 * written by Adrien Grand and approved by Yann Collet (author of original LZ4 library).
48 *
49 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
50 * * Magic * Token * Compressed * Decompressed * Checksum * + * LZ4 compressed *
51 * * * * length * length * * * block *
52 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
53 */
54 public class Lz4FrameDecoder extends ByteToMessageDecoder {
55 private final int maxDecompressedLength;
56 /**
57 * Current state of stream.
58 */
59 private enum State {
60 INIT_BLOCK,
61 DECOMPRESS_DATA,
62 FINISHED,
63 CORRUPTED
64 }
65
66 private State currentState = State.INIT_BLOCK;
67
68 /**
69 * Underlying decompressor in use.
70 */
71 private LZ4SafeDecompressor decompressor;
72
73 /**
74 * Underlying checksum calculator in use.
75 */
76 private ByteBufChecksum checksum;
77
78 /**
79 * Type of current block.
80 */
81 private int blockType;
82
83 /**
84 * Compressed length of current incoming block.
85 */
86 private int compressedLength;
87
88 /**
89 * Decompressed length of current incoming block.
90 */
91 private int decompressedLength;
92
93 /**
94 * Checksum value of current incoming block.
95 */
96 private int currentChecksum;
97
98 /**
99 * Creates the fastest LZ4 decoder.
100 *
101 * Note that by default, validation of the checksum header in each chunk is
102 * DISABLED for performance improvements. If performance is less of an issue,
103 * or if you would prefer the safety that checksum validation brings, please
104 * use the {@link #Lz4FrameDecoder(boolean)} constructor with the argument
105 * set to {@code true}.
106 */
107 public Lz4FrameDecoder() {
108 this(false);
109 }
110
111 /**
112 * Creates a LZ4 decoder with fastest decoder instance available on your machine.
113 *
114 * @param validateChecksums if {@code true}, the checksum field will be validated against the actual
115 * uncompressed data, and if the checksums do not match, a suitable
116 * {@link DecompressionException} will be thrown
117 */
118 public Lz4FrameDecoder(boolean validateChecksums) {
119 this(LZ4Factory.fastestInstance(), validateChecksums);
120 }
121
122 /**
123 * Creates a LZ4 decoder with fastest decoder instance available on your machine.
124 *
125 * @param validateChecksums if {@code true}, the checksum field will be validated against the actual
126 * uncompressed data, and if the checksums do not match, a suitable
127 * {@link DecompressionException} will be thrown
128 * @param maxDecompressedLength
129 * maximum length of the decompressed block. If {@code 0} is given it uses {@code 32MB}
130 * by default.
131 */
132 public Lz4FrameDecoder(boolean validateChecksums, int maxDecompressedLength) {
133 this(LZ4Factory.fastestInstance(), validateChecksums ? new Lz4XXHash32(DEFAULT_SEED) : null,
134 maxDecompressedLength);
135 }
136
137 /**
138 * Creates a new LZ4 decoder with customizable implementation.
139 *
140 * @param factory user customizable {@link LZ4Factory} instance
141 * which may be JNI bindings to the original C implementation, a pure Java implementation
142 * or a Java implementation that uses the {@link sun.misc.Unsafe}
143 * @param validateChecksums if {@code true}, the checksum field will be validated against the actual
144 * uncompressed data, and if the checksums do not match, a suitable
145 * {@link DecompressionException} will be thrown. In this case encoder will use
146 * xxhash hashing for Java, based on Yann Collet's work available at
147 * <a href="https://github.com/Cyan4973/xxHash">Github</a>.
148 */
149 public Lz4FrameDecoder(LZ4Factory factory, boolean validateChecksums) {
150 this(factory, validateChecksums ? new Lz4XXHash32(DEFAULT_SEED) : null);
151 }
152
153 /**
154 * Creates a new customizable LZ4 decoder.
155 *
156 * @param factory user customizable {@link LZ4Factory} instance
157 * which may be JNI bindings to the original C implementation, a pure Java implementation
158 * or a Java implementation that uses the {@link sun.misc.Unsafe}
159 * @param checksum the {@link Checksum} instance to use to check data for integrity.
160 * You may set {@code null} if you do not want to validate checksum of each block
161 */
162 public Lz4FrameDecoder(LZ4Factory factory, Checksum checksum) {
163 this(factory, checksum, MAX_BLOCK_SIZE);
164 }
165
166 /**
167 * Creates a new customizable LZ4 decoder.
168 *
169 * @param factory user customizable {@link LZ4Factory} instance
170 * which may be JNI bindings to the original C implementation, a pure Java implementation
171 * or a Java implementation that uses the {@link sun.misc.Unsafe}
172 * @param checksum the {@link Checksum} instance to use to check data for integrity.
173 * You may set {@code null} if you do not want to validate checksum of each block
174 * @param maxDecompressedLength
175 * maximum length of the decompressed block. If {@code 0} is given it uses {@code 32MB} by default.
176 */
177 public Lz4FrameDecoder(LZ4Factory factory, Checksum checksum, int maxDecompressedLength) {
178 decompressor = ObjectUtil.checkNotNull(factory, "factory").safeDecompressor();
179 this.checksum = checksum == null ? null : ByteBufChecksum.wrapChecksum(checksum);
180 this.maxDecompressedLength = maxDecompressedLength == 0 ? MAX_BLOCK_SIZE :
181 ObjectUtil.checkInRange(maxDecompressedLength, 0, MAX_BLOCK_SIZE, "maxDecompressedLength");
182 }
183
184 @Override
185 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
186 try {
187 switch (currentState) {
188 case INIT_BLOCK:
189 if (in.readableBytes() < HEADER_LENGTH) {
190 break;
191 }
192 final long magic = in.readLong();
193 if (magic != MAGIC_NUMBER) {
194 throw new DecompressionException("unexpected block identifier");
195 }
196
197 final int token = in.readByte();
198 final int compressionLevel = (token & 0x0F) + COMPRESSION_LEVEL_BASE;
199 int blockType = token & 0xF0;
200
201 int compressedLength = Integer.reverseBytes(in.readInt());
202 if (compressedLength < 0 || compressedLength > MAX_BLOCK_SIZE) {
203 throw new DecompressionException(String.format(
204 "invalid compressedLength: %d (expected: 0-%d)",
205 compressedLength, MAX_BLOCK_SIZE));
206 }
207
208 int decompressedLength = Integer.reverseBytes(in.readInt());
209 if (decompressedLength > maxDecompressedLength) {
210 throw new DecompressionException(String.format(
211 "decompressedLength too large: %d (expected: 0-%d)",
212 decompressedLength, maxDecompressedLength));
213 }
214
215 final int maxLocalDecompressedLength = 1 << compressionLevel;
216 if (decompressedLength < 0 || decompressedLength > maxLocalDecompressedLength) {
217 throw new DecompressionException(String.format(
218 "invalid decompressedLength: %d (expected: 0-%d)",
219 decompressedLength, maxLocalDecompressedLength));
220 }
221 if (decompressedLength == 0 && compressedLength != 0
222 || decompressedLength != 0 && compressedLength == 0
223 || blockType == BLOCK_TYPE_NON_COMPRESSED && decompressedLength != compressedLength) {
224 throw new DecompressionException(String.format(
225 "stream corrupted: compressedLength(%d) and decompressedLength(%d) mismatch",
226 compressedLength, decompressedLength));
227 }
228
229 int currentChecksum = Integer.reverseBytes(in.readInt());
230 if (decompressedLength == 0 && compressedLength == 0) {
231 if (currentChecksum != 0) {
232 throw new DecompressionException("stream corrupted: checksum error");
233 }
234 currentState = State.FINISHED;
235 decompressor = null;
236 checksum = null;
237 break;
238 }
239
240 this.blockType = blockType;
241 this.compressedLength = compressedLength;
242 this.decompressedLength = decompressedLength;
243 this.currentChecksum = currentChecksum;
244
245 currentState = State.DECOMPRESS_DATA;
246 // fall through
247 case DECOMPRESS_DATA:
248 blockType = this.blockType;
249 compressedLength = this.compressedLength;
250 decompressedLength = this.decompressedLength;
251 currentChecksum = this.currentChecksum;
252
253 if (in.readableBytes() < compressedLength) {
254 break;
255 }
256
257 final ByteBufChecksum checksum = this.checksum;
258 ByteBuf uncompressed = null;
259
260 try {
261 switch (blockType) {
262 case BLOCK_TYPE_NON_COMPRESSED:
263 // Just pass through, we not update the readerIndex yet as we do this outside of the
264 // switch statement.
265 uncompressed = in.retainedSlice(in.readerIndex(), decompressedLength);
266 break;
267 case BLOCK_TYPE_COMPRESSED:
268 uncompressed = ctx.alloc().buffer(decompressedLength, decompressedLength);
269
270 ByteBuffer source = CompressionUtil.safeNioBuffer(
271 in, in.readerIndex(), compressedLength);
272 ByteBuffer destination = uncompressed.internalNioBuffer(
273 uncompressed.writerIndex(), decompressedLength);
274 int actualDecompressedLength = decompressor.decompress(
275 source, source.position(), compressedLength,
276 destination, destination.position(), decompressedLength);
277 if (actualDecompressedLength != decompressedLength) {
278 throw new DecompressionException(String.format(
279 "stream corrupted: decompressedLength(%d) and " +
280 "actualDecompressedLength(%d) mismatch",
281 decompressedLength, actualDecompressedLength));
282 }
283 // Update the writerIndex now to reflect what we decompressed.
284 uncompressed.writerIndex(uncompressed.writerIndex() + decompressedLength);
285 break;
286 default:
287 throw new DecompressionException(String.format(
288 "unexpected blockType: %d (expected: %d or %d)",
289 blockType, BLOCK_TYPE_NON_COMPRESSED, BLOCK_TYPE_COMPRESSED));
290 }
291 // Skip inbound bytes after we processed them.
292 in.skipBytes(compressedLength);
293
294 if (checksum != null) {
295 CompressionUtil.checkChecksum(checksum, uncompressed, currentChecksum);
296 }
297 out.add(uncompressed);
298 uncompressed = null;
299 currentState = State.INIT_BLOCK;
300 } catch (LZ4Exception e) {
301 throw new DecompressionException(e);
302 } finally {
303 if (uncompressed != null) {
304 uncompressed.release();
305 }
306 }
307 break;
308 case FINISHED:
309 case CORRUPTED:
310 in.skipBytes(in.readableBytes());
311 break;
312 default:
313 throw new IllegalStateException();
314 }
315 } catch (Exception e) {
316 currentState = State.CORRUPTED;
317 throw e;
318 }
319 }
320
321 /**
322 * Returns {@code true} if and only if the end of the compressed stream
323 * has been reached.
324 */
325 public boolean isClosed() {
326 return currentState == State.FINISHED;
327 }
328 }