1
2
3
4
5
6
7
8
9
10
11
12
13
14
15 package io.netty.handler.codec.http2;
16
17 import io.netty.buffer.ByteBuf;
18 import io.netty.buffer.Unpooled;
19 import io.netty.channel.ChannelFuture;
20 import io.netty.channel.ChannelHandlerContext;
21 import io.netty.channel.ChannelPromise;
22 import io.netty.channel.embedded.EmbeddedChannel;
23 import io.netty.handler.codec.ByteToMessageDecoder;
24 import io.netty.handler.codec.compression.BrotliEncoder;
25 import io.netty.handler.codec.compression.ZlibCodecFactory;
26 import io.netty.handler.codec.compression.ZlibWrapper;
27 import io.netty.handler.codec.compression.Brotli;
28 import io.netty.handler.codec.compression.BrotliOptions;
29 import io.netty.handler.codec.compression.CompressionOptions;
30 import io.netty.handler.codec.compression.DeflateOptions;
31 import io.netty.handler.codec.compression.GzipOptions;
32 import io.netty.handler.codec.compression.StandardCompressionOptions;
33 import io.netty.handler.codec.compression.Zstd;
34 import io.netty.handler.codec.compression.ZstdEncoder;
35 import io.netty.handler.codec.compression.ZstdOptions;
36 import io.netty.handler.codec.compression.SnappyFrameEncoder;
37 import io.netty.handler.codec.compression.SnappyOptions;
38 import io.netty.util.concurrent.PromiseCombiner;
39 import io.netty.util.internal.ObjectUtil;
40
41 import java.util.ArrayList;
42 import java.util.List;
43
44 import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_ENCODING;
45 import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
46 import static io.netty.handler.codec.http.HttpHeaderValues.BR;
47 import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
48 import static io.netty.handler.codec.http.HttpHeaderValues.GZIP;
49 import static io.netty.handler.codec.http.HttpHeaderValues.IDENTITY;
50 import static io.netty.handler.codec.http.HttpHeaderValues.X_DEFLATE;
51 import static io.netty.handler.codec.http.HttpHeaderValues.X_GZIP;
52 import static io.netty.handler.codec.http.HttpHeaderValues.ZSTD;
53 import static io.netty.handler.codec.http.HttpHeaderValues.SNAPPY;
54
55
56
57
58
59 public class CompressorHttp2ConnectionEncoder extends DecoratingHttp2ConnectionEncoder {
60
61 public static final int DEFAULT_COMPRESSION_LEVEL = 6;
62 public static final int DEFAULT_WINDOW_BITS = 15;
63 public static final int DEFAULT_MEM_LEVEL = 8;
64
65 private int compressionLevel;
66 private int windowBits;
67 private int memLevel;
68 private final Http2Connection.PropertyKey propertyKey;
69
70 private final boolean supportsCompressionOptions;
71
72 private BrotliOptions brotliOptions;
73 private GzipOptions gzipCompressionOptions;
74 private DeflateOptions deflateOptions;
75 private ZstdOptions zstdOptions;
76 private SnappyOptions snappyOptions;
77
78
79
80
81
82 public CompressorHttp2ConnectionEncoder(Http2ConnectionEncoder delegate) {
83 this(delegate, defaultCompressionOptions());
84 }
85
86 private static CompressionOptions[] defaultCompressionOptions() {
87 List<CompressionOptions> compressionOptions = new ArrayList<CompressionOptions>();
88 compressionOptions.add(StandardCompressionOptions.gzip());
89 compressionOptions.add(StandardCompressionOptions.deflate());
90 compressionOptions.add(StandardCompressionOptions.snappy());
91 if (Brotli.isAvailable()) {
92 compressionOptions.add(StandardCompressionOptions.brotli());
93 }
94 if (Zstd.isAvailable()) {
95 compressionOptions.add(StandardCompressionOptions.zstd());
96 }
97 return compressionOptions.toArray(new CompressionOptions[0]);
98 }
99
100
101
102
103 @Deprecated
104 public CompressorHttp2ConnectionEncoder(Http2ConnectionEncoder delegate, int compressionLevel, int windowBits,
105 int memLevel) {
106 super(delegate);
107 this.compressionLevel = ObjectUtil.checkInRange(compressionLevel, 0, 9, "compressionLevel");
108 this.windowBits = ObjectUtil.checkInRange(windowBits, 9, 15, "windowBits");
109 this.memLevel = ObjectUtil.checkInRange(memLevel, 1, 9, "memLevel");
110
111 propertyKey = connection().newKey();
112 connection().addListener(new Http2ConnectionAdapter() {
113 @Override
114 public void onStreamRemoved(Http2Stream stream) {
115 final EmbeddedChannel compressor = stream.getProperty(propertyKey);
116 if (compressor != null) {
117 cleanup(stream, compressor);
118 }
119 }
120 });
121
122 supportsCompressionOptions = false;
123 }
124
125
126
127
128
129 public CompressorHttp2ConnectionEncoder(Http2ConnectionEncoder delegate,
130 CompressionOptions... compressionOptionsArgs) {
131 super(delegate);
132 ObjectUtil.checkNotNull(compressionOptionsArgs, "CompressionOptions");
133 ObjectUtil.deepCheckNotNull("CompressionOptions", compressionOptionsArgs);
134
135 for (CompressionOptions compressionOptions : compressionOptionsArgs) {
136
137
138
139
140
141
142 if (Brotli.isAvailable() && compressionOptions instanceof BrotliOptions) {
143 brotliOptions = (BrotliOptions) compressionOptions;
144 } else if (compressionOptions instanceof GzipOptions) {
145 gzipCompressionOptions = (GzipOptions) compressionOptions;
146 } else if (compressionOptions instanceof DeflateOptions) {
147 deflateOptions = (DeflateOptions) compressionOptions;
148 } else if (compressionOptions instanceof ZstdOptions) {
149 zstdOptions = (ZstdOptions) compressionOptions;
150 } else if (compressionOptions instanceof SnappyOptions) {
151 snappyOptions = (SnappyOptions) compressionOptions;
152 } else {
153 throw new IllegalArgumentException("Unsupported " + CompressionOptions.class.getSimpleName() +
154 ": " + compressionOptions);
155 }
156 }
157
158 supportsCompressionOptions = true;
159
160 propertyKey = connection().newKey();
161 connection().addListener(new Http2ConnectionAdapter() {
162 @Override
163 public void onStreamRemoved(Http2Stream stream) {
164 final EmbeddedChannel compressor = stream.getProperty(propertyKey);
165 if (compressor != null) {
166 cleanup(stream, compressor);
167 }
168 }
169 });
170 }
171
172 @Override
173 public ChannelFuture writeData(final ChannelHandlerContext ctx, final int streamId, ByteBuf data, int padding,
174 final boolean endOfStream, ChannelPromise promise) {
175 final Http2Stream stream = connection().stream(streamId);
176 final EmbeddedChannel channel = stream == null ? null : (EmbeddedChannel) stream.getProperty(propertyKey);
177 if (channel == null) {
178
179 return super.writeData(ctx, streamId, data, padding, endOfStream, promise);
180 }
181
182 try {
183
184 channel.writeOutbound(data);
185 ByteBuf buf = nextReadableBuf(channel);
186 if (buf == null) {
187 if (endOfStream) {
188 if (channel.finish()) {
189 buf = nextReadableBuf(channel);
190 }
191 return super.writeData(ctx, streamId, buf == null ? Unpooled.EMPTY_BUFFER : buf, padding,
192 true, promise);
193 }
194
195 promise.setSuccess();
196 return promise;
197 }
198
199 PromiseCombiner combiner = new PromiseCombiner(ctx.executor());
200 for (;;) {
201 ByteBuf nextBuf = nextReadableBuf(channel);
202 boolean compressedEndOfStream = nextBuf == null && endOfStream;
203 if (compressedEndOfStream && channel.finish()) {
204 nextBuf = nextReadableBuf(channel);
205 compressedEndOfStream = nextBuf == null;
206 }
207
208 ChannelPromise bufPromise = ctx.newPromise();
209 combiner.add(bufPromise);
210 super.writeData(ctx, streamId, buf, padding, compressedEndOfStream, bufPromise);
211 if (nextBuf == null) {
212 break;
213 }
214
215 padding = 0;
216 buf = nextBuf;
217 }
218 combiner.finish(promise);
219 } catch (Throwable cause) {
220 promise.tryFailure(cause);
221 } finally {
222 if (endOfStream) {
223 cleanup(stream, channel);
224 }
225 }
226 return promise;
227 }
228
229 @Override
230 public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
231 boolean endStream, ChannelPromise promise) {
232 EmbeddedChannel compressor = null;
233 try {
234
235 compressor = newCompressor(ctx, headers, endStream);
236
237
238 ChannelFuture future = super.writeHeaders(ctx, streamId, headers, padding, endStream, promise);
239
240
241 if (bindCompressorToStream(compressor, streamId)) {
242 compressor = null;
243 }
244
245 return future;
246 } catch (Throwable e) {
247 promise.tryFailure(e);
248 } finally {
249 if (compressor != null) {
250 compressor.finishAndReleaseAll();
251 }
252 }
253 return promise;
254 }
255
256 @Override
257 public ChannelFuture writeHeaders(final ChannelHandlerContext ctx, final int streamId, final Http2Headers headers,
258 final int streamDependency, final short weight, final boolean exclusive, final int padding,
259 final boolean endOfStream, final ChannelPromise promise) {
260 EmbeddedChannel compressor = null;
261 try {
262
263 compressor = newCompressor(ctx, headers, endOfStream);
264
265
266 ChannelFuture future = super.writeHeaders(ctx, streamId, headers, streamDependency, weight, exclusive,
267 padding, endOfStream, promise);
268
269
270 if (bindCompressorToStream(compressor, streamId)) {
271 compressor = null;
272 }
273
274 return future;
275 } catch (Throwable e) {
276 promise.tryFailure(e);
277 } finally {
278 if (compressor != null) {
279 compressor.finishAndReleaseAll();
280 }
281 }
282 return promise;
283 }
284
285
286
287
288
289
290
291
292
293
294
295 protected EmbeddedChannel newContentCompressor(ChannelHandlerContext ctx, CharSequence contentEncoding)
296 throws Http2Exception {
297 if (GZIP.contentEqualsIgnoreCase(contentEncoding) || X_GZIP.contentEqualsIgnoreCase(contentEncoding)) {
298 return newCompressionChannel(ctx, ZlibWrapper.GZIP);
299 }
300 if (DEFLATE.contentEqualsIgnoreCase(contentEncoding) || X_DEFLATE.contentEqualsIgnoreCase(contentEncoding)) {
301 return newCompressionChannel(ctx, ZlibWrapper.ZLIB);
302 }
303 if (Brotli.isAvailable() && brotliOptions != null && BR.contentEqualsIgnoreCase(contentEncoding)) {
304 return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
305 ctx.channel().config(), new BrotliEncoder(brotliOptions.parameters()));
306 }
307 if (zstdOptions != null && ZSTD.contentEqualsIgnoreCase(contentEncoding)) {
308 return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
309 ctx.channel().config(), new ZstdEncoder(zstdOptions.compressionLevel(),
310 zstdOptions.blockSize(), zstdOptions.maxEncodeSize()));
311 }
312 if (snappyOptions != null && SNAPPY.contentEqualsIgnoreCase(contentEncoding)) {
313 return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
314 ctx.channel().config(), new SnappyFrameEncoder());
315 }
316
317 return null;
318 }
319
320
321
322
323
324
325
326
327
328 protected CharSequence getTargetContentEncoding(CharSequence contentEncoding) throws Http2Exception {
329 return contentEncoding;
330 }
331
332
333
334
335
336
337 private EmbeddedChannel newCompressionChannel(final ChannelHandlerContext ctx, ZlibWrapper wrapper) {
338 if (supportsCompressionOptions) {
339 if (wrapper == ZlibWrapper.GZIP && gzipCompressionOptions != null) {
340 return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
341 ctx.channel().config(), ZlibCodecFactory.newZlibEncoder(wrapper,
342 gzipCompressionOptions.compressionLevel(), gzipCompressionOptions.windowBits(),
343 gzipCompressionOptions.memLevel()));
344 } else if (wrapper == ZlibWrapper.ZLIB && deflateOptions != null) {
345 return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
346 ctx.channel().config(), ZlibCodecFactory.newZlibEncoder(wrapper,
347 deflateOptions.compressionLevel(), deflateOptions.windowBits(),
348 deflateOptions.memLevel()));
349 } else {
350 throw new IllegalArgumentException("Unsupported ZlibWrapper: " + wrapper);
351 }
352 } else {
353 return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
354 ctx.channel().config(), ZlibCodecFactory.newZlibEncoder(wrapper, compressionLevel, windowBits,
355 memLevel));
356 }
357 }
358
359
360
361
362
363
364
365
366
367
368
369 private EmbeddedChannel newCompressor(ChannelHandlerContext ctx, Http2Headers headers, boolean endOfStream)
370 throws Http2Exception {
371 if (endOfStream) {
372 return null;
373 }
374
375 CharSequence encoding = headers.get(CONTENT_ENCODING);
376 if (encoding == null) {
377 encoding = IDENTITY;
378 }
379 EmbeddedChannel compressor = newContentCompressor(ctx, encoding);
380 try {
381 if (compressor != null) {
382 CharSequence targetContentEncoding = getTargetContentEncoding(encoding);
383 if (IDENTITY.contentEqualsIgnoreCase(targetContentEncoding)) {
384 headers.remove(CONTENT_ENCODING);
385 } else {
386 headers.set(CONTENT_ENCODING, targetContentEncoding);
387 }
388
389
390
391
392 headers.remove(CONTENT_LENGTH);
393 }
394
395 EmbeddedChannel result = compressor;
396 compressor = null;
397 return result;
398 } finally {
399 if (compressor != null) {
400 compressor.finishAndReleaseAll();
401 }
402 }
403 }
404
405
406
407
408
409
410
411 private boolean bindCompressorToStream(EmbeddedChannel compressor, int streamId) {
412 if (compressor != null) {
413 Http2Stream stream = connection().stream(streamId);
414 if (stream != null) {
415 stream.setProperty(propertyKey, compressor);
416 return true;
417 }
418 }
419 return false;
420 }
421
422
423
424
425
426
427
428 void cleanup(Http2Stream stream, EmbeddedChannel compressor) {
429 compressor.finishAndReleaseAll();
430 stream.removeProperty(propertyKey);
431 }
432
433
434
435
436
437
438
439 private static ByteBuf nextReadableBuf(EmbeddedChannel compressor) {
440 for (;;) {
441 final ByteBuf buf = compressor.readOutbound();
442 if (buf == null) {
443 return null;
444 }
445 if (!buf.isReadable()) {
446 buf.release();
447 continue;
448 }
449 return buf;
450 }
451 }
452 }