1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.handler.codec.compression;
17
18 import com.ning.compress.BufferRecycler;
19 import com.ning.compress.lzf.ChunkEncoder;
20 import com.ning.compress.lzf.LZFChunk;
21 import com.ning.compress.lzf.LZFEncoder;
22 import com.ning.compress.lzf.util.ChunkEncoderFactory;
23 import io.netty.buffer.ByteBuf;
24 import io.netty.channel.ChannelHandlerContext;
25 import io.netty.handler.codec.MessageToByteEncoder;
26
27 import static com.ning.compress.lzf.LZFChunk.MAX_CHUNK_LEN;
28
29
30
31
32
33
34
35 public class LzfEncoder extends MessageToByteEncoder<ByteBuf> {
36
37
38
39
40
41 private static final int MIN_BLOCK_TO_COMPRESS = 16;
42
43
44
45
46
47
48
49
50 private final int compressThreshold;
51
52
53
54
55 private final ChunkEncoder encoder;
56
57
58
59
60 private final BufferRecycler recycler;
61
62
63
64
65
66
67
68 public LzfEncoder() {
69 this(false);
70 }
71
72
73
74
75
76
77
78
79
80
81 public LzfEncoder(boolean safeInstance) {
82 this(safeInstance, MAX_CHUNK_LEN);
83 }
84
85
86
87
88
89
90
91
92
93
94
95
96 public LzfEncoder(boolean safeInstance, int totalLength) {
97 this(safeInstance, totalLength, MIN_BLOCK_TO_COMPRESS);
98 }
99
100
101
102
103
104
105
106
107
108 public LzfEncoder(int totalLength) {
109 this(false, totalLength);
110 }
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126 public LzfEncoder(boolean safeInstance, int totalLength, int compressThreshold) {
127 super(false);
128 if (totalLength < MIN_BLOCK_TO_COMPRESS || totalLength > MAX_CHUNK_LEN) {
129 throw new IllegalArgumentException("totalLength: " + totalLength +
130 " (expected: " + MIN_BLOCK_TO_COMPRESS + '-' + MAX_CHUNK_LEN + ')');
131 }
132
133 if (compressThreshold < MIN_BLOCK_TO_COMPRESS) {
134
135 throw new IllegalArgumentException("compressThreshold:" + compressThreshold +
136 " expected >=" + MIN_BLOCK_TO_COMPRESS);
137 }
138 this.compressThreshold = compressThreshold;
139
140 this.encoder = safeInstance ?
141 ChunkEncoderFactory.safeNonAllocatingInstance(totalLength)
142 : ChunkEncoderFactory.optimalNonAllocatingInstance(totalLength);
143
144 this.recycler = BufferRecycler.instance();
145 }
146
147 @Override
148 protected void encode(ChannelHandlerContext ctx, ByteBuf in, ByteBuf out) throws Exception {
149 final int length = in.readableBytes();
150 final int idx = in.readerIndex();
151 final byte[] input;
152 final int inputPtr;
153 if (in.hasArray()) {
154 input = in.array();
155 inputPtr = in.arrayOffset() + idx;
156 } else {
157 input = recycler.allocInputBuffer(length);
158 in.getBytes(idx, input, 0, length);
159 inputPtr = 0;
160 }
161
162
163 final int maxOutputLength = LZFEncoder.estimateMaxWorkspaceSize(length) + 1;
164 out.ensureWritable(maxOutputLength);
165 final byte[] output;
166 final int outputPtr;
167 if (out.hasArray()) {
168 output = out.array();
169 outputPtr = out.arrayOffset() + out.writerIndex();
170 } else {
171 output = new byte[maxOutputLength];
172 outputPtr = 0;
173 }
174
175 final int outputLength;
176 if (length >= compressThreshold) {
177
178 outputLength = encodeCompress(input, inputPtr, length, output, outputPtr);
179 } else {
180
181 outputLength = encodeNonCompress(input, inputPtr, length, output, outputPtr);
182 }
183
184 if (out.hasArray()) {
185 out.writerIndex(out.writerIndex() + outputLength);
186 } else {
187 out.writeBytes(output, 0, outputLength);
188 }
189
190 in.skipBytes(length);
191
192 if (!in.hasArray()) {
193 recycler.releaseInputBuffer(input);
194 }
195 }
196
197 private int encodeCompress(byte[] input, int inputPtr, int length, byte[] output, int outputPtr) {
198 return LZFEncoder.appendEncoded(encoder,
199 input, inputPtr, length, output, outputPtr) - outputPtr;
200 }
201
202 private static int lzfEncodeNonCompress(byte[] input, int inputPtr, int length, byte[] output, int outputPtr) {
203 int left = length;
204 int chunkLen = Math.min(LZFChunk.MAX_CHUNK_LEN, left);
205 outputPtr = LZFChunk.appendNonCompressed(input, inputPtr, chunkLen, output, outputPtr);
206 left -= chunkLen;
207 if (left < 1) {
208 return outputPtr;
209 }
210 inputPtr += chunkLen;
211 do {
212 chunkLen = Math.min(left, LZFChunk.MAX_CHUNK_LEN);
213 outputPtr = LZFChunk.appendNonCompressed(input, inputPtr, chunkLen, output, outputPtr);
214 inputPtr += chunkLen;
215 left -= chunkLen;
216 } while (left > 0);
217 return outputPtr;
218 }
219
220
221
222
223 private static int encodeNonCompress(byte[] input, int inputPtr, int length, byte[] output, int outputPtr) {
224 return lzfEncodeNonCompress(input, inputPtr, length, output, outputPtr) - outputPtr;
225 }
226
227 @Override
228 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
229 encoder.close();
230 super.handlerRemoved(ctx);
231 }
232 }