View Javadoc
1   /*
2    * Copyright 2014 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License, version 2.0 (the
5    * "License"); you may not use this file except in compliance with the License. You may obtain a
6    * 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 distributed under the License
11   * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
12   * or implied. See the License for the specific language governing permissions and limitations under
13   * the License.
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   * A decorating HTTP2 encoder that will compress data frames according to the {@code content-encoding} header for each
61   * stream. The compression provided by this class will be applied to the data for the entire stream.
62   */
63  public class CompressorHttp2ConnectionEncoder extends DecoratingHttp2ConnectionEncoder {
64      // We cannot remove this because it'll be breaking change
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       * Create a new {@link CompressorHttp2ConnectionEncoder} instance
84       * with default implementation of {@link StandardCompressionOptions}
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      * Create a new {@link CompressorHttp2ConnectionEncoder} instance
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      * Create a new {@link CompressorHttp2ConnectionEncoder} with
131      * specified {@link StandardCompressionOptions}
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             // BrotliOptions' class initialization depends on Brotli classes being on the classpath.
141             // The Brotli.isAvailable check ensures that BrotliOptions will only get instantiated if Brotli is on
142             // the classpath.
143             // This results in the static analysis of native-image identifying the instanceof BrotliOptions check
144             // and thus BrotliOptions itself as unreachable, enabling native-image to link all classes at build time
145             // and not complain about the missing Brotli classes.
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             // The compressor may be null if no compatible encoding type was found in this stream's headers
183             return super.writeData(ctx, streamId, data, padding, endOfStream, promise);
184         }
185 
186         try {
187             // The channel will release the buffer after being written
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                 // END_STREAM is not set and the assumption is data is still forthcoming.
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; // Padding is only communicated once on the first iteration
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             // Determine if compression is required and sanitize the headers.
239             compressor = newCompressor(ctx, headers, endStream);
240 
241             // Write the headers and create the stream object.
242             ChannelFuture future = super.writeHeaders(ctx, streamId, headers, padding, endStream, promise);
243 
244             // After the stream object has been created, then attach the compressor as a property for data compression.
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             // Determine if compression is required and sanitize the headers.
267             compressor = newCompressor(ctx, headers, endOfStream);
268 
269             // Write the headers and create the stream object.
270             ChannelFuture future = super.writeHeaders(ctx, streamId, headers, streamDependency, weight, exclusive,
271                                                       padding, endOfStream, promise);
272 
273             // After the stream object has been created, then attach the compressor as a property for data compression.
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      * Returns a new {@link EmbeddedChannel} that encodes the HTTP2 message content encoded in the specified
291      * {@code contentEncoding}.
292      *
293      * @param ctx the context.
294      * @param contentEncoding the value of the {@code content-encoding} header
295      * @return a new {@link ByteToMessageDecoder} if the specified encoding is supported. {@code null} otherwise
296      * (alternatively, you can throw a {@link Http2Exception} to block unknown encoding).
297      * @throws Http2Exception If the specified encoding is not supported and warrants an exception
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         // 'identity' or unsupported
334         return null;
335     }
336 
337     /**
338      * Returns the expected content encoding of the decoded content. Returning {@code contentEncoding} is the default
339      * behavior, which is the case for most compressors.
340      *
341      * @param contentEncoding the value of the {@code content-encoding} header
342      * @return the expected content encoding of the new content.
343      * @throws Http2Exception if the {@code contentEncoding} is not supported and warrants an exception
344      */
345     protected CharSequence getTargetContentEncoding(CharSequence contentEncoding) throws Http2Exception {
346         return contentEncoding;
347     }
348 
349     /**
350      * Generate a new instance of an {@link EmbeddedChannel} capable of compressing data
351      * @param ctx the context.
352      * @param wrapper Defines what type of encoder should be used
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      * Checks if a new compressor object is needed for the stream identified by {@code streamId}. This method will
394      * modify the {@code content-encoding} header contained in {@code headers}.
395      *
396      * @param ctx the context.
397      * @param headers Object representing headers which are to be written
398      * @param endOfStream Indicates if the stream has ended
399      * @return The channel used to compress data.
400      * @throws Http2Exception if any problems occur during initialization.
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                 // The content length will be for the decompressed data. Since we will compress the data
423                 // this content-length will not be correct. Instead of queuing messages or delaying sending
424                 // header frames...just remove the content-length header
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      * Called after the super class has written the headers and created any associated stream objects.
440      * @param compressor The compressor associated with the stream identified by {@code streamId}.
441      * @param streamId The stream id for which the headers were written.
442      * @return {@code true} if ownership of {@code compressor} was transferred to the stream.
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      * Release remaining content from {@link EmbeddedChannel} and remove the compressor from the {@link Http2Stream}.
457      *
458      * @param stream The stream for which {@code compressor} is the compressor for
459      * @param compressor The compressor for {@code stream}
460      */
461     void cleanup(Http2Stream stream, EmbeddedChannel compressor) {
462         compressor.finishAndReleaseAll();
463         stream.removeProperty(propertyKey);
464     }
465 
466     /**
467      * Read the next compressed {@link ByteBuf} from the {@link EmbeddedChannel} or {@code null} if one does not exist.
468      *
469      * @param compressor The channel to read from
470      * @return The next decoded {@link ByteBuf} from the {@link EmbeddedChannel} or {@code null} if one does not exist
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 }