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.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   * A decorating HTTP2 encoder that will compress data frames according to the {@code content-encoding} header for each
57   * stream. The compression provided by this class will be applied to the data for the entire stream.
58   */
59  public class CompressorHttp2ConnectionEncoder extends DecoratingHttp2ConnectionEncoder {
60      // We cannot remove this because it'll be breaking change
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       * Create a new {@link CompressorHttp2ConnectionEncoder} instance
80       * with default implementation of {@link StandardCompressionOptions}
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      * Create a new {@link CompressorHttp2ConnectionEncoder} instance
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      * Create a new {@link CompressorHttp2ConnectionEncoder} with
127      * specified {@link StandardCompressionOptions}
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             // BrotliOptions' class initialization depends on Brotli classes being on the classpath.
137             // The Brotli.isAvailable check ensures that BrotliOptions will only get instantiated if Brotli is on
138             // the classpath.
139             // This results in the static analysis of native-image identifying the instanceof BrotliOptions check
140             // and thus BrotliOptions itself as unreachable, enabling native-image to link all classes at build time
141             // and not complain about the missing Brotli classes.
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             // The compressor may be null if no compatible encoding type was found in this stream's headers
179             return super.writeData(ctx, streamId, data, padding, endOfStream, promise);
180         }
181 
182         try {
183             // The channel will release the buffer after being written
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                 // END_STREAM is not set and the assumption is data is still forthcoming.
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; // Padding is only communicated once on the first iteration
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             // Determine if compression is required and sanitize the headers.
235             compressor = newCompressor(ctx, headers, endStream);
236 
237             // Write the headers and create the stream object.
238             ChannelFuture future = super.writeHeaders(ctx, streamId, headers, padding, endStream, promise);
239 
240             // After the stream object has been created, then attach the compressor as a property for data compression.
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             // Determine if compression is required and sanitize the headers.
263             compressor = newCompressor(ctx, headers, endOfStream);
264 
265             // Write the headers and create the stream object.
266             ChannelFuture future = super.writeHeaders(ctx, streamId, headers, streamDependency, weight, exclusive,
267                                                       padding, endOfStream, promise);
268 
269             // After the stream object has been created, then attach the compressor as a property for data compression.
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      * Returns a new {@link EmbeddedChannel} that encodes the HTTP2 message content encoded in the specified
287      * {@code contentEncoding}.
288      *
289      * @param ctx the context.
290      * @param contentEncoding the value of the {@code content-encoding} header
291      * @return a new {@link ByteToMessageDecoder} if the specified encoding is supported. {@code null} otherwise
292      * (alternatively, you can throw a {@link Http2Exception} to block unknown encoding).
293      * @throws Http2Exception If the specified encoding is not supported and warrants an exception
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         // 'identity' or unsupported
317         return null;
318     }
319 
320     /**
321      * Returns the expected content encoding of the decoded content. Returning {@code contentEncoding} is the default
322      * behavior, which is the case for most compressors.
323      *
324      * @param contentEncoding the value of the {@code content-encoding} header
325      * @return the expected content encoding of the new content.
326      * @throws Http2Exception if the {@code contentEncoding} is not supported and warrants an exception
327      */
328     protected CharSequence getTargetContentEncoding(CharSequence contentEncoding) throws Http2Exception {
329         return contentEncoding;
330     }
331 
332     /**
333      * Generate a new instance of an {@link EmbeddedChannel} capable of compressing data
334      * @param ctx the context.
335      * @param wrapper Defines what type of encoder should be used
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      * Checks if a new compressor object is needed for the stream identified by {@code streamId}. This method will
361      * modify the {@code content-encoding} header contained in {@code headers}.
362      *
363      * @param ctx the context.
364      * @param headers Object representing headers which are to be written
365      * @param endOfStream Indicates if the stream has ended
366      * @return The channel used to compress data.
367      * @throws Http2Exception if any problems occur during initialization.
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                 // The content length will be for the decompressed data. Since we will compress the data
390                 // this content-length will not be correct. Instead of queuing messages or delaying sending
391                 // header frames...just remove the content-length header
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      * Called after the super class has written the headers and created any associated stream objects.
407      * @param compressor The compressor associated with the stream identified by {@code streamId}.
408      * @param streamId The stream id for which the headers were written.
409      * @return {@code true} if ownership of {@code compressor} was transferred to the stream.
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      * Release remaining content from {@link EmbeddedChannel} and remove the compressor from the {@link Http2Stream}.
424      *
425      * @param stream The stream for which {@code compressor} is the compressor for
426      * @param compressor The compressor for {@code stream}
427      */
428     void cleanup(Http2Stream stream, EmbeddedChannel compressor) {
429         compressor.finishAndReleaseAll();
430         stream.removeProperty(propertyKey);
431     }
432 
433     /**
434      * Read the next compressed {@link ByteBuf} from the {@link EmbeddedChannel} or {@code null} if one does not exist.
435      *
436      * @param compressor The channel to read from
437      * @return The next decoded {@link ByteBuf} from the {@link EmbeddedChannel} or {@code null} if one does not exist
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 }