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