View Javadoc
1   /*
2    * Copyright 2015 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  
16  package io.netty.handler.codec.http2;
17  
18  import io.netty.buffer.ByteBuf;
19  import io.netty.buffer.ByteBufUtil;
20  import io.netty.channel.ChannelFuture;
21  import io.netty.channel.ChannelHandlerContext;
22  import io.netty.channel.ChannelPromise;
23  import io.netty.util.ReferenceCountUtil;
24  import io.netty.util.internal.UnstableApi;
25  
26  import java.util.ArrayDeque;
27  import java.util.Iterator;
28  import java.util.Map;
29  import java.util.Queue;
30  import java.util.TreeMap;
31  
32  import static io.netty.handler.codec.http2.Http2CodecUtil.SMALLEST_MAX_CONCURRENT_STREAMS;
33  import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
34  import static io.netty.handler.codec.http2.Http2Exception.connectionError;
35  
36  /**
37   * Implementation of a {@link Http2ConnectionEncoder} that dispatches all method call to another
38   * {@link Http2ConnectionEncoder}, until {@code SETTINGS_MAX_CONCURRENT_STREAMS} is reached.
39   * <p/>
40   * <p>When this limit is hit, instead of rejecting any new streams this implementation buffers newly
41   * created streams and their corresponding frames. Once an active stream gets closed or the maximum
42   * number of concurrent streams is increased, this encoder will automatically try to empty its
43   * buffer and create as many new streams as possible.
44   * <p/>
45   * <p>
46   * If a {@code GOAWAY} frame is received from the remote endpoint, all buffered writes for streams
47   * with an ID less than the specified {@code lastStreamId} will immediately fail with a
48   * {@link Http2GoAwayException}.
49   * <p/>
50   * <p>
51   * If the channel/encoder gets closed, all new and buffered writes will immediately fail with a
52   * {@link Http2ChannelClosedException}.
53   * </p>
54   * <p>This implementation makes the buffering mostly transparent and is expected to be used as a
55   * drop-in decorator of {@link DefaultHttp2ConnectionEncoder}.
56   * </p>
57   */
58  @UnstableApi
59  public class StreamBufferingEncoder extends DecoratingHttp2ConnectionEncoder {
60  
61      /**
62       * Thrown if buffered streams are terminated due to this encoder being closed.
63       */
64      public static final class Http2ChannelClosedException extends Http2Exception {
65          private static final long serialVersionUID = 4768543442094476971L;
66  
67          public Http2ChannelClosedException() {
68              super(Http2Error.REFUSED_STREAM, "Connection closed");
69          }
70      }
71  
72      /**
73       * Thrown by {@link StreamBufferingEncoder} if buffered streams are terminated due to
74       * receipt of a {@code GOAWAY}.
75       */
76      public static final class Http2GoAwayException extends Http2Exception {
77          private static final long serialVersionUID = 1326785622777291198L;
78          private final int lastStreamId;
79          private final long errorCode;
80          private final byte[] debugData;
81  
82          public Http2GoAwayException(int lastStreamId, long errorCode, byte[] debugData) {
83              super(Http2Error.STREAM_CLOSED);
84              this.lastStreamId = lastStreamId;
85              this.errorCode = errorCode;
86              this.debugData = debugData;
87          }
88  
89          public int lastStreamId() {
90              return lastStreamId;
91          }
92  
93          public long errorCode() {
94              return errorCode;
95          }
96  
97          public byte[] debugData() {
98              return debugData;
99          }
100     }
101 
102     /**
103      * Buffer for any streams and corresponding frames that could not be created due to the maximum
104      * concurrent stream limit being hit.
105      */
106     private final TreeMap<Integer, PendingStream> pendingStreams = new TreeMap<Integer, PendingStream>();
107     private int maxConcurrentStreams;
108     private boolean closed;
109 
110     public StreamBufferingEncoder(Http2ConnectionEncoder delegate) {
111         this(delegate, SMALLEST_MAX_CONCURRENT_STREAMS);
112     }
113 
114     public StreamBufferingEncoder(Http2ConnectionEncoder delegate, int initialMaxConcurrentStreams) {
115         super(delegate);
116         this.maxConcurrentStreams = initialMaxConcurrentStreams;
117         connection().addListener(new Http2ConnectionAdapter() {
118 
119             @Override
120             public void onGoAwayReceived(int lastStreamId, long errorCode, ByteBuf debugData) {
121                 cancelGoAwayStreams(lastStreamId, errorCode, debugData);
122             }
123 
124             @Override
125             public void onStreamClosed(Http2Stream stream) {
126                 tryCreatePendingStreams();
127             }
128         });
129     }
130 
131     /**
132      * Indicates the number of streams that are currently buffered, awaiting creation.
133      */
134     public int numBufferedStreams() {
135         return pendingStreams.size();
136     }
137 
138     @Override
139     public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
140                                       int padding, boolean endStream, ChannelPromise promise) {
141         return writeHeaders(ctx, streamId, headers, 0, Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT,
142                 false, padding, endStream, promise);
143     }
144 
145     @Override
146     public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
147                                       int streamDependency, short weight, boolean exclusive,
148                                       int padding, boolean endOfStream, ChannelPromise promise) {
149         if (closed) {
150             return promise.setFailure(new Http2ChannelClosedException());
151         }
152         if (isExistingStream(streamId) || connection().goAwayReceived()) {
153             return super.writeHeaders(ctx, streamId, headers, streamDependency, weight,
154                     exclusive, padding, endOfStream, promise);
155         }
156         if (canCreateStream()) {
157             return super.writeHeaders(ctx, streamId, headers, streamDependency, weight,
158                     exclusive, padding, endOfStream, promise);
159         }
160         PendingStream pendingStream = pendingStreams.get(streamId);
161         if (pendingStream == null) {
162             pendingStream = new PendingStream(ctx, streamId);
163             pendingStreams.put(streamId, pendingStream);
164         }
165         pendingStream.frames.add(new HeadersFrame(headers, streamDependency, weight, exclusive,
166                 padding, endOfStream, promise));
167         return promise;
168     }
169 
170     @Override
171     public ChannelFuture writeRstStream(ChannelHandlerContext ctx, int streamId, long errorCode,
172                                         ChannelPromise promise) {
173         if (isExistingStream(streamId)) {
174             return super.writeRstStream(ctx, streamId, errorCode, promise);
175         }
176         // Since the delegate doesn't know about any buffered streams we have to handle cancellation
177         // of the promises and releasing of the ByteBufs here.
178         PendingStream stream = pendingStreams.remove(streamId);
179         if (stream != null) {
180             // Sending a RST_STREAM to a buffered stream will succeed the promise of all frames
181             // associated with the stream, as sending a RST_STREAM means that someone "doesn't care"
182             // about the stream anymore and thus there is not point in failing the promises and invoking
183             // error handling routines.
184             stream.close(null);
185             promise.setSuccess();
186         } else {
187             promise.setFailure(connectionError(PROTOCOL_ERROR, "Stream does not exist %d", streamId));
188         }
189         return promise;
190     }
191 
192     @Override
193     public ChannelFuture writeData(ChannelHandlerContext ctx, int streamId, ByteBuf data,
194                                    int padding, boolean endOfStream, ChannelPromise promise) {
195         if (isExistingStream(streamId)) {
196             return super.writeData(ctx, streamId, data, padding, endOfStream, promise);
197         }
198         PendingStream pendingStream = pendingStreams.get(streamId);
199         if (pendingStream != null) {
200             pendingStream.frames.add(new DataFrame(data, padding, endOfStream, promise));
201         } else {
202             ReferenceCountUtil.safeRelease(data);
203             promise.setFailure(connectionError(PROTOCOL_ERROR, "Stream does not exist %d", streamId));
204         }
205         return promise;
206     }
207 
208     @Override
209     public void remoteSettings(Http2Settings settings) throws Http2Exception {
210         // Need to let the delegate decoder handle the settings first, so that it sees the
211         // new setting before we attempt to create any new streams.
212         super.remoteSettings(settings);
213 
214         // Get the updated value for SETTINGS_MAX_CONCURRENT_STREAMS.
215         maxConcurrentStreams = connection().local().maxActiveStreams();
216 
217         // Try to create new streams up to the new threshold.
218         tryCreatePendingStreams();
219     }
220 
221     @Override
222     public void close() {
223         try {
224             if (!closed) {
225                 closed = true;
226 
227                 // Fail all buffered streams.
228                 Http2ChannelClosedException e = new Http2ChannelClosedException();
229                 while (!pendingStreams.isEmpty()) {
230                     PendingStream stream = pendingStreams.pollFirstEntry().getValue();
231                     stream.close(e);
232                 }
233             }
234         } finally {
235             super.close();
236         }
237     }
238 
239     private void tryCreatePendingStreams() {
240         while (!pendingStreams.isEmpty() && canCreateStream()) {
241             Map.Entry<Integer, PendingStream> entry = pendingStreams.pollFirstEntry();
242             PendingStream pendingStream = entry.getValue();
243             try {
244                 pendingStream.sendFrames();
245             } catch (Throwable t) {
246                 pendingStream.close(t);
247             }
248         }
249     }
250 
251     private void cancelGoAwayStreams(int lastStreamId, long errorCode, ByteBuf debugData) {
252         Iterator<PendingStream> iter = pendingStreams.values().iterator();
253         Exception e = new Http2GoAwayException(lastStreamId, errorCode, ByteBufUtil.getBytes(debugData));
254         while (iter.hasNext()) {
255             PendingStream stream = iter.next();
256             if (stream.streamId > lastStreamId) {
257                 iter.remove();
258                 stream.close(e);
259             }
260         }
261     }
262 
263     /**
264      * Determines whether or not we're allowed to create a new stream right now.
265      */
266     private boolean canCreateStream() {
267         return connection().local().numActiveStreams() < maxConcurrentStreams;
268     }
269 
270     private boolean isExistingStream(int streamId) {
271         return streamId <= connection().local().lastStreamCreated();
272     }
273 
274     private static final class PendingStream {
275         final ChannelHandlerContext ctx;
276         final int streamId;
277         final Queue<Frame> frames = new ArrayDeque<Frame>(2);
278 
279         PendingStream(ChannelHandlerContext ctx, int streamId) {
280             this.ctx = ctx;
281             this.streamId = streamId;
282         }
283 
284         void sendFrames() {
285             for (Frame frame : frames) {
286                 frame.send(ctx, streamId);
287             }
288         }
289 
290         void close(Throwable t) {
291             for (Frame frame : frames) {
292                 frame.release(t);
293             }
294         }
295     }
296 
297     private abstract static class Frame {
298         final ChannelPromise promise;
299 
300         Frame(ChannelPromise promise) {
301             this.promise = promise;
302         }
303 
304         /**
305          * Release any resources (features, buffers, ...) associated with the frame.
306          */
307         void release(Throwable t) {
308             if (t == null) {
309                 promise.setSuccess();
310             } else {
311                 promise.setFailure(t);
312             }
313         }
314 
315         abstract void send(ChannelHandlerContext ctx, int streamId);
316     }
317 
318     private final class HeadersFrame extends Frame {
319         final Http2Headers headers;
320         final int streamDependency;
321         final short weight;
322         final boolean exclusive;
323         final int padding;
324         final boolean endOfStream;
325 
326         HeadersFrame(Http2Headers headers, int streamDependency, short weight, boolean exclusive,
327                      int padding, boolean endOfStream, ChannelPromise promise) {
328             super(promise);
329             this.headers = headers;
330             this.streamDependency = streamDependency;
331             this.weight = weight;
332             this.exclusive = exclusive;
333             this.padding = padding;
334             this.endOfStream = endOfStream;
335         }
336 
337         @Override
338         void send(ChannelHandlerContext ctx, int streamId) {
339             writeHeaders(ctx, streamId, headers, streamDependency, weight, exclusive, padding, endOfStream, promise);
340         }
341     }
342 
343     private final class DataFrame extends Frame {
344         final ByteBuf data;
345         final int padding;
346         final boolean endOfStream;
347 
348         DataFrame(ByteBuf data, int padding, boolean endOfStream, ChannelPromise promise) {
349             super(promise);
350             this.data = data;
351             this.padding = padding;
352             this.endOfStream = endOfStream;
353         }
354 
355         @Override
356         void release(Throwable t) {
357             super.release(t);
358             ReferenceCountUtil.safeRelease(data);
359         }
360 
361         @Override
362         void send(ChannelHandlerContext ctx, int streamId) {
363             writeData(ctx, streamId, data, padding, endOfStream, promise);
364         }
365     }
366 }