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