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.netty5.handler.codec.http2;
16  
17  import io.netty5.buffer.api.Buffer;
18  import io.netty5.channel.ChannelHandlerContext;
19  import io.netty5.channel.embedded.EmbeddedChannel;
20  import io.netty5.handler.codec.compression.Brotli;
21  import io.netty5.handler.codec.compression.BrotliDecompressor;
22  import io.netty5.handler.codec.compression.Decompressor;
23  import io.netty5.handler.codec.compression.ZlibDecompressor;
24  import io.netty5.handler.codec.compression.ZlibWrapper;
25  import io.netty5.util.internal.UnstableApi;
26  
27  import static io.netty5.handler.codec.http.HttpHeaderNames.CONTENT_ENCODING;
28  import static io.netty5.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
29  import static io.netty5.handler.codec.http.HttpHeaderValues.BR;
30  import static io.netty5.handler.codec.http.HttpHeaderValues.DEFLATE;
31  import static io.netty5.handler.codec.http.HttpHeaderValues.GZIP;
32  import static io.netty5.handler.codec.http.HttpHeaderValues.IDENTITY;
33  import static io.netty5.handler.codec.http.HttpHeaderValues.X_DEFLATE;
34  import static io.netty5.handler.codec.http.HttpHeaderValues.X_GZIP;
35  import static io.netty5.handler.codec.http2.Http2Error.INTERNAL_ERROR;
36  import static io.netty5.handler.codec.http2.Http2Exception.streamError;
37  import static io.netty5.util.internal.ObjectUtil.checkPositiveOrZero;
38  import static java.util.Objects.requireNonNull;
39  
40  /**
41   * An HTTP2 frame listener that will decompress data frames according to the {@code content-encoding} header for each
42   * stream. The decompression provided by this class will be applied to the data for the entire stream.
43   */
44  @UnstableApi
45  public class DelegatingDecompressorFrameListener extends Http2FrameListenerDecorator {
46  
47      private final Http2Connection connection;
48      private final boolean strict;
49      private boolean flowControllerInitialized;
50      private final Http2Connection.PropertyKey propertyKey;
51  
52      public DelegatingDecompressorFrameListener(Http2Connection connection, Http2FrameListener listener) {
53          this(connection, listener, true);
54      }
55  
56      public DelegatingDecompressorFrameListener(Http2Connection connection, Http2FrameListener listener,
57                      boolean strict) {
58          super(listener);
59          this.connection = connection;
60          this.strict = strict;
61  
62          propertyKey = connection.newKey();
63          connection.addListener(new Http2ConnectionAdapter() {
64              @Override
65              public void onStreamRemoved(Http2Stream stream) {
66                  final Http2Decompressor decompressor = decompressor(stream);
67                  if (decompressor != null) {
68                      cleanup(decompressor);
69                  }
70              }
71          });
72      }
73  
74      @Override
75      public int onDataRead(ChannelHandlerContext ctx, int streamId, Buffer data, int padding, boolean endOfStream)
76              throws Http2Exception {
77          final Http2Stream stream = connection.stream(streamId);
78          final Http2Decompressor decompressor = decompressor(stream);
79          if (decompressor == null) {
80              // The decompressor may be null if no compatible encoding type was found in this stream's headers
81              return listener.onDataRead(ctx, streamId, data, padding, endOfStream);
82          }
83  
84          final Decompressor decomp = decompressor.decompressor();
85          final int compressedBytes = data.readableBytes() + padding;
86          decompressor.incrementCompressedBytes(compressedBytes);
87          Buffer decompressed = null;
88          try {
89              Http2LocalFlowController flowController = connection.local().flowController();
90              decompressor.incrementDecompressedBytes(padding);
91  
92              for (;;) {
93                  if (decomp.isFinished()) {
94                      flowController.consumeBytes(stream, listener.onDataRead(
95                              ctx, streamId, ctx.bufferAllocator().allocate(0), padding, endOfStream));
96                      break;
97                  }
98                  int idx = data.readerOffset();
99                  decompressed = decomp.decompress(data, ctx.bufferAllocator());
100                 if (decompressed == null || idx == data.readerOffset()) {
101                     flowController.consumeBytes(stream, listener.onDataRead(
102                             ctx, streamId, ctx.bufferAllocator().allocate(0), padding, endOfStream));
103                     break;
104                 } else {
105                     // Immediately return the bytes back to the flow controller. ConsumedBytesConverter will convert
106                     // from the decompressed amount which the user knows about to the compressed amount which flow
107                     // control knows about.
108                     decompressor.incrementDecompressedBytes(decompressed.readableBytes());
109                     flowController.consumeBytes(stream,
110                             listener.onDataRead(ctx, streamId, decompressed, padding, false));
111                     decompressed.close();
112                     decompressed = null;
113                 }
114                 padding = 0; // Padding is only communicated once on the first iteration.
115             }
116             // We consume bytes each time we call the listener to ensure if multiple frames are decompressed
117             // that the bytes are accounted for immediately. Otherwise the user may see an inconsistent state of
118             // flow control.
119             return 0;
120         } catch (Http2Exception e) {
121             throw e;
122         } catch (Throwable t) {
123             throw streamError(stream.id(), INTERNAL_ERROR, t,
124                     "Decompressor error detected while delegating data read on streamId %d", stream.id());
125         } finally {
126             if (decompressed != null) {
127                 decompressed.close();
128             }
129         }
130     }
131 
132     @Override
133     public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
134                     boolean endStream) throws Http2Exception {
135         initDecompressor(ctx, streamId, headers, endStream);
136         listener.onHeadersRead(ctx, streamId, headers, padding, endStream);
137     }
138 
139     @Override
140     public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
141                     short weight, boolean exclusive, int padding, boolean endStream) throws Http2Exception {
142         initDecompressor(ctx, streamId, headers, endStream);
143         listener.onHeadersRead(ctx, streamId, headers, streamDependency, weight, exclusive, padding, endStream);
144     }
145 
146     /**
147      * Returns a new {@link EmbeddedChannel} that decodes the HTTP2 message content encoded in the specified
148      * {@code contentEncoding}.
149      *
150      * @param contentEncoding the value of the {@code content-encoding} header
151      * @return a new {@link Decompressor} if the specified encoding is supported. {@code null} otherwise
152      *         (alternatively, you can throw a {@link Http2Exception} to block unknown encoding).
153      * @throws Http2Exception If the specified encoding is not supported and warrants an exception
154      */
155     protected Decompressor newContentDecompressor(final ChannelHandlerContext ctx, CharSequence contentEncoding)
156             throws Http2Exception {
157         if (GZIP.contentEqualsIgnoreCase(contentEncoding) || X_GZIP.contentEqualsIgnoreCase(contentEncoding)) {
158             return ZlibDecompressor.newFactory(ZlibWrapper.GZIP).get();
159         }
160         if (DEFLATE.contentEqualsIgnoreCase(contentEncoding) || X_DEFLATE.contentEqualsIgnoreCase(contentEncoding)) {
161             final ZlibWrapper wrapper = strict ? ZlibWrapper.ZLIB : ZlibWrapper.ZLIB_OR_NONE;
162             // To be strict, 'deflate' means ZLIB, but some servers were not implemented correctly.
163             return ZlibDecompressor.newFactory(wrapper).get();
164         }
165         if (Brotli.isAvailable() && BR.contentEqualsIgnoreCase(contentEncoding)) {
166             return BrotliDecompressor.newFactory().get();
167         }
168         // 'identity' or unsupported
169         return null;
170     }
171 
172     /**
173      * Returns the expected content encoding of the decoded content. This getMethod returns {@code "identity"} by
174      * default, which is the case for most decompressors.
175      *
176      * @param contentEncoding the value of the {@code content-encoding} header
177      * @return the expected content encoding of the new content.
178      * @throws Http2Exception if the {@code contentEncoding} is not supported and warrants an exception
179      */
180     protected CharSequence getTargetContentEncoding(@SuppressWarnings("UnusedParameters") CharSequence contentEncoding)
181                     throws Http2Exception {
182         return IDENTITY;
183     }
184 
185     /**
186      * Checks if a new decompressor object is needed for the stream identified by {@code streamId}.
187      * This method will modify the {@code content-encoding} header contained in {@code headers}.
188      *
189      * @param ctx The context
190      * @param streamId The identifier for the headers inside {@code headers}
191      * @param headers Object representing headers which have been read
192      * @param endOfStream Indicates if the stream has ended
193      * @throws Http2Exception If the {@code content-encoding} is not supported
194      */
195     private void initDecompressor(ChannelHandlerContext ctx, int streamId, Http2Headers headers, boolean endOfStream)
196             throws Http2Exception {
197         final Http2Stream stream = connection.stream(streamId);
198         if (stream == null) {
199             return;
200         }
201 
202         Http2Decompressor decompressor = decompressor(stream);
203         if (decompressor == null && !endOfStream) {
204             // Determine the content encoding.
205             CharSequence contentEncoding = headers.get(CONTENT_ENCODING);
206             if (contentEncoding == null) {
207                 contentEncoding = IDENTITY;
208             }
209             final Decompressor decomp = newContentDecompressor(ctx, contentEncoding);
210             if (decomp != null) {
211                 decompressor = new Http2Decompressor(decomp);
212                 stream.setProperty(propertyKey, decompressor);
213                 // Decode the content and remove or replace the existing headers
214                 // so that the message looks like a decoded message.
215                 CharSequence targetContentEncoding = getTargetContentEncoding(contentEncoding);
216                 if (IDENTITY.contentEqualsIgnoreCase(targetContentEncoding)) {
217                     headers.remove(CONTENT_ENCODING);
218                 } else {
219                     headers.set(CONTENT_ENCODING, targetContentEncoding);
220                 }
221             }
222         }
223 
224         if (decompressor != null) {
225             // The content length will be for the compressed data. Since we will decompress the data
226             // this content-length will not be correct. Instead of queuing messages or delaying sending
227             // header frames just remove the content-length header.
228             headers.remove(CONTENT_LENGTH);
229 
230             // The first time that we initialize a decompressor, decorate the local flow controller to
231             // properly convert consumed bytes.
232             if (!flowControllerInitialized) {
233                 flowControllerInitialized = true;
234                 connection.local().flowController(new ConsumedBytesConverter(connection.local().flowController()));
235             }
236         }
237     }
238 
239     Http2Decompressor decompressor(Http2Stream stream) {
240         return stream == null ? null : (Http2Decompressor) stream.getProperty(propertyKey);
241     }
242 
243     /**
244      * Release remaining content from the {@link Decompressor}.
245      *
246      * @param decompressor The decompressor for {@code stream}
247      */
248     private static void cleanup(Http2Decompressor decompressor) {
249         decompressor.decompressor().close();
250     }
251 
252     /**
253      * A decorator around the local flow controller that converts consumed bytes from uncompressed to compressed.
254      */
255     private final class ConsumedBytesConverter implements Http2LocalFlowController {
256         private final Http2LocalFlowController flowController;
257 
258         ConsumedBytesConverter(Http2LocalFlowController flowController) {
259             this.flowController = requireNonNull(flowController, "flowController");
260         }
261 
262         @Override
263         public Http2LocalFlowController frameWriter(Http2FrameWriter frameWriter) {
264             return flowController.frameWriter(frameWriter);
265         }
266 
267         @Override
268         public void channelHandlerContext(ChannelHandlerContext ctx) throws Http2Exception {
269             flowController.channelHandlerContext(ctx);
270         }
271 
272         @Override
273         public void initialWindowSize(int newWindowSize) throws Http2Exception {
274             flowController.initialWindowSize(newWindowSize);
275         }
276 
277         @Override
278         public int initialWindowSize() {
279             return flowController.initialWindowSize();
280         }
281 
282         @Override
283         public int windowSize(Http2Stream stream) {
284             return flowController.windowSize(stream);
285         }
286 
287         @Override
288         public void incrementWindowSize(Http2Stream stream, int delta) throws Http2Exception {
289             flowController.incrementWindowSize(stream, delta);
290         }
291 
292         @Override
293         public void receiveFlowControlledFrame(Http2Stream stream, Buffer data, int padding,
294                                                boolean endOfStream) throws Http2Exception {
295             flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
296         }
297 
298         @Override
299         public boolean consumeBytes(Http2Stream stream, int numBytes) throws Http2Exception {
300             Http2Decompressor decompressor = decompressor(stream);
301             if (decompressor != null) {
302                 // Convert the decompressed bytes to compressed (on the wire) bytes.
303                 numBytes = decompressor.consumeBytes(stream.id(), numBytes);
304             }
305             try {
306                 return flowController.consumeBytes(stream, numBytes);
307             } catch (Http2Exception e) {
308                 throw e;
309             } catch (Throwable t) {
310                 // The stream should be closed at this point. We have already changed our state tracking the compressed
311                 // bytes, and there is no guarantee we can recover if the underlying flow controller throws.
312                 throw streamError(stream.id(), INTERNAL_ERROR, t, "Error while returning bytes to flow control window");
313             }
314         }
315 
316         @Override
317         public int unconsumedBytes(Http2Stream stream) {
318             return flowController.unconsumedBytes(stream);
319         }
320 
321         @Override
322         public int initialWindowSize(Http2Stream stream) {
323             return flowController.initialWindowSize(stream);
324         }
325     }
326 
327     /**
328      * Provides the state for stream {@code DATA} frame decompression.
329      */
330     private static final class Http2Decompressor {
331         private final Decompressor decompressor;
332         private int compressed;
333         private int decompressed;
334 
335         Http2Decompressor(Decompressor decompressor) {
336             this.decompressor = decompressor;
337         }
338 
339         /**
340          * Responsible for taking compressed bytes in and producing decompressed bytes.
341          */
342         Decompressor decompressor() {
343             return decompressor;
344         }
345 
346         /**
347          * Increment the number of bytes received prior to doing any decompression.
348          */
349         void incrementCompressedBytes(int delta) {
350             assert delta >= 0;
351             compressed += delta;
352         }
353 
354         /**
355          * Increment the number of bytes after the decompression process.
356          */
357         void incrementDecompressedBytes(int delta) {
358             assert delta >= 0;
359             decompressed += delta;
360         }
361 
362         /**
363          * Determines the ratio between {@code numBytes} and {@link Http2Decompressor#decompressed}.
364          * This ratio is used to decrement {@link Http2Decompressor#decompressed} and
365          * {@link Http2Decompressor#compressed}.
366          * @param streamId the stream ID
367          * @param decompressedBytes The number of post-decompressed bytes to return to flow control
368          * @return The number of pre-decompressed bytes that have been consumed.
369          */
370         int consumeBytes(int streamId, int decompressedBytes) throws Http2Exception {
371             checkPositiveOrZero(decompressedBytes, "decompressedBytes");
372             if (decompressed - decompressedBytes < 0) {
373                 throw streamError(streamId, INTERNAL_ERROR,
374                         "Attempting to return too many bytes for stream %d. decompressed: %d " +
375                                 "decompressedBytes: %d", streamId, decompressed, decompressedBytes);
376             }
377             double consumedRatio = decompressedBytes / (double) decompressed;
378             int consumedCompressed = Math.min(compressed, (int) Math.ceil(compressed * consumedRatio));
379             if (compressed - consumedCompressed < 0) {
380                 throw streamError(streamId, INTERNAL_ERROR,
381                         "overflow when converting decompressed bytes to compressed bytes for stream %d." +
382                                 "decompressedBytes: %d decompressed: %d compressed: %d consumedCompressed: %d",
383                         streamId, decompressedBytes, decompressed, compressed, consumedCompressed);
384             }
385             decompressed -= decompressedBytes;
386             compressed -= consumedCompressed;
387 
388             return consumedCompressed;
389         }
390     }
391 }