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