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