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  
16  package io.netty.handler.codec.http2;
17  
18  import io.netty.buffer.ByteBuf;
19  import io.netty.channel.ChannelFuture;
20  import io.netty.channel.ChannelPromise;
21  import io.netty.handler.codec.http2.Http2Stream.State;
22  import io.netty.util.collection.IntObjectHashMap;
23  import io.netty.util.collection.IntObjectMap;
24  import io.netty.util.collection.IntObjectMap.PrimitiveEntry;
25  import io.netty.util.concurrent.Future;
26  import io.netty.util.concurrent.Promise;
27  import io.netty.util.concurrent.PromiseNotifier;
28  import io.netty.util.internal.EmptyArrays;
29  import io.netty.util.internal.UnstableApi;
30  import io.netty.util.internal.logging.InternalLogger;
31  import io.netty.util.internal.logging.InternalLoggerFactory;
32  
33  import java.util.ArrayDeque;
34  import java.util.ArrayList;
35  import java.util.Arrays;
36  import java.util.Iterator;
37  import java.util.LinkedHashSet;
38  import java.util.List;
39  import java.util.Queue;
40  import java.util.Set;
41  
42  import static io.netty.handler.codec.http2.Http2CodecUtil.CONNECTION_STREAM_ID;
43  import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_MAX_RESERVED_STREAMS;
44  import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
45  import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
46  import static io.netty.handler.codec.http2.Http2Error.REFUSED_STREAM;
47  import static io.netty.handler.codec.http2.Http2Exception.closedStreamError;
48  import static io.netty.handler.codec.http2.Http2Exception.connectionError;
49  import static io.netty.handler.codec.http2.Http2Exception.streamError;
50  import static io.netty.handler.codec.http2.Http2Stream.State.CLOSED;
51  import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
52  import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_REMOTE;
53  import static io.netty.handler.codec.http2.Http2Stream.State.IDLE;
54  import static io.netty.handler.codec.http2.Http2Stream.State.OPEN;
55  import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_LOCAL;
56  import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_REMOTE;
57  import static io.netty.util.internal.ObjectUtil.checkNotNull;
58  import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
59  import static java.lang.Integer.MAX_VALUE;
60  
61  /**
62   * Simple implementation of {@link Http2Connection}.
63   */
64  @UnstableApi
65  public class DefaultHttp2Connection implements Http2Connection {
66      private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultHttp2Connection.class);
67      // Fields accessed by inner classes
68      final IntObjectMap<Http2Stream> streamMap = new IntObjectHashMap<Http2Stream>();
69      final PropertyKeyRegistry propertyKeyRegistry = new PropertyKeyRegistry();
70      final ConnectionStream connectionStream = new ConnectionStream();
71      final DefaultEndpoint<Http2LocalFlowController> localEndpoint;
72      final DefaultEndpoint<Http2RemoteFlowController> remoteEndpoint;
73  
74      /**
75       * We chose a {@link List} over a {@link Set} to avoid allocating an {@link Iterator} objects when iterating over
76       * the listeners.
77       * <p>
78       * Initial size of 4 because the default configuration currently has 3 listeners
79       * (local/remote flow controller and {@link StreamByteDistributor}) and we leave room for 1 extra.
80       * We could be more aggressive but the ArrayList resize will double the size if we are too small.
81       */
82      final List<Listener> listeners = new ArrayList<Listener>(4);
83      final ActiveStreams activeStreams;
84      Promise<Void> closePromise;
85  
86      /**
87       * Creates a new connection with the given settings.
88       * @param server whether or not this end-point is the server-side of the HTTP/2 connection.
89       */
90      public DefaultHttp2Connection(boolean server) {
91          this(server, DEFAULT_MAX_RESERVED_STREAMS);
92      }
93  
94      /**
95       * Creates a new connection with the given settings.
96       * @param server whether or not this end-point is the server-side of the HTTP/2 connection.
97       * @param maxReservedStreams The maximum amount of streams which can exist in the reserved state for each endpoint.
98       */
99      public DefaultHttp2Connection(boolean server, int maxReservedStreams) {
100         activeStreams = new ActiveStreams(listeners);
101         // Reserved streams are excluded from the SETTINGS_MAX_CONCURRENT_STREAMS limit according to [1] and the RFC
102         // doesn't define a way to communicate the limit on reserved streams. We rely upon the peer to send RST_STREAM
103         // in response to any locally enforced limits being exceeded [2].
104         // [1] https://tools.ietf.org/html/rfc7540#section-5.1.2
105         // [2] https://tools.ietf.org/html/rfc7540#section-8.2.2
106         localEndpoint = new DefaultEndpoint<Http2LocalFlowController>(server, server ? MAX_VALUE : maxReservedStreams);
107         remoteEndpoint = new DefaultEndpoint<Http2RemoteFlowController>(!server, maxReservedStreams);
108 
109         // Add the connection stream to the map.
110         streamMap.put(connectionStream.id(), connectionStream);
111     }
112 
113     /**
114      * Determine if {@link #close(Promise)} has been called and no more streams are allowed to be created.
115      */
116     final boolean isClosed() {
117         return closePromise != null;
118     }
119 
120     @Override
121     public Future<Void> close(final Promise<Void> promise) {
122         checkNotNull(promise, "promise");
123         // Since we allow this method to be called multiple times, we must make sure that all the promises are notified
124         // when all streams are removed and the close operation completes.
125         if (closePromise != null) {
126             if (closePromise == promise) {
127                 // Do nothing
128             } else if (promise instanceof ChannelPromise && ((ChannelFuture) closePromise).isVoid()) {
129                 closePromise = promise;
130             } else {
131                 PromiseNotifier.cascade(closePromise, promise);
132             }
133         } else {
134             closePromise = promise;
135         }
136         if (isStreamMapEmpty()) {
137             promise.trySuccess(null);
138             return promise;
139         }
140 
141         Iterator<PrimitiveEntry<Http2Stream>> itr = streamMap.entries().iterator();
142         // We must take care while iterating the streamMap as to not modify while iterating in case there are other code
143         // paths iterating over the active streams.
144         if (activeStreams.allowModifications()) {
145             activeStreams.incrementPendingIterations();
146             try {
147                 while (itr.hasNext()) {
148                     DefaultStream stream = (DefaultStream) itr.next().value();
149                     if (stream.id() != CONNECTION_STREAM_ID) {
150                         // If modifications of the activeStream map is allowed, then a stream close operation will also
151                         // modify the streamMap. Pass the iterator in so that remove will be called to prevent
152                         // concurrent modification exceptions.
153                         stream.close(itr);
154                     }
155                 }
156             } finally {
157                 activeStreams.decrementPendingIterations();
158             }
159         } else {
160             while (itr.hasNext()) {
161                 Http2Stream stream = itr.next().value();
162                 if (stream.id() != CONNECTION_STREAM_ID) {
163                     // We are not allowed to make modifications, so the close calls will be executed after this
164                     // iteration completes.
165                     stream.close();
166                 }
167             }
168         }
169         return closePromise;
170     }
171 
172     @Override
173     public void addListener(Listener listener) {
174         listeners.add(listener);
175     }
176 
177     @Override
178     public void removeListener(Listener listener) {
179         listeners.remove(listener);
180     }
181 
182     @Override
183     public boolean isServer() {
184         return localEndpoint.isServer();
185     }
186 
187     @Override
188     public Http2Stream connectionStream() {
189         return connectionStream;
190     }
191 
192     @Override
193     public Http2Stream stream(int streamId) {
194         return streamMap.get(streamId);
195     }
196 
197     @Override
198     public boolean streamMayHaveExisted(int streamId) {
199         return remoteEndpoint.mayHaveCreatedStream(streamId) || localEndpoint.mayHaveCreatedStream(streamId);
200     }
201 
202     @Override
203     public int numActiveStreams() {
204         return activeStreams.size();
205     }
206 
207     @Override
208     public Http2Stream forEachActiveStream(Http2StreamVisitor visitor) throws Http2Exception {
209         return activeStreams.forEachActiveStream(visitor);
210     }
211 
212     @Override
213     public Endpoint<Http2LocalFlowController> local() {
214         return localEndpoint;
215     }
216 
217     @Override
218     public Endpoint<Http2RemoteFlowController> remote() {
219         return remoteEndpoint;
220     }
221 
222     @Override
223     public boolean goAwayReceived() {
224         return localEndpoint.lastStreamKnownByPeer >= 0;
225     }
226 
227     @Override
228     public void goAwayReceived(final int lastKnownStream, long errorCode, ByteBuf debugData) throws Http2Exception {
229         if (localEndpoint.lastStreamKnownByPeer() >= 0 && localEndpoint.lastStreamKnownByPeer() < lastKnownStream) {
230             throw connectionError(PROTOCOL_ERROR, "lastStreamId MUST NOT increase. Current value: %d new value: %d",
231                     localEndpoint.lastStreamKnownByPeer(), lastKnownStream);
232         }
233 
234         localEndpoint.lastStreamKnownByPeer(lastKnownStream);
235         for (int i = 0; i < listeners.size(); ++i) {
236             try {
237                 listeners.get(i).onGoAwayReceived(lastKnownStream, errorCode, debugData);
238             } catch (Throwable cause) {
239                 logger.error("Caught Throwable from listener onGoAwayReceived.", cause);
240             }
241         }
242 
243         closeStreamsGreaterThanLastKnownStreamId(lastKnownStream, localEndpoint);
244     }
245 
246     @Override
247     public boolean goAwaySent() {
248         return remoteEndpoint.lastStreamKnownByPeer >= 0;
249     }
250 
251     @Override
252     public boolean goAwaySent(final int lastKnownStream, long errorCode, ByteBuf debugData) throws Http2Exception {
253         if (remoteEndpoint.lastStreamKnownByPeer() >= 0) {
254             // Protect against re-entrancy. Could happen if writing the frame fails, and error handling
255             // treating this is a connection handler and doing a graceful shutdown...
256             if (lastKnownStream == remoteEndpoint.lastStreamKnownByPeer()) {
257                 return false;
258             }
259             if (lastKnownStream > remoteEndpoint.lastStreamKnownByPeer()) {
260                 throw connectionError(PROTOCOL_ERROR, "Last stream identifier must not increase between " +
261                                 "sending multiple GOAWAY frames (was '%d', is '%d').",
262                         remoteEndpoint.lastStreamKnownByPeer(), lastKnownStream);
263             }
264         }
265 
266         remoteEndpoint.lastStreamKnownByPeer(lastKnownStream);
267         for (int i = 0; i < listeners.size(); ++i) {
268             try {
269                 listeners.get(i).onGoAwaySent(lastKnownStream, errorCode, debugData);
270             } catch (Throwable cause) {
271                 logger.error("Caught Throwable from listener onGoAwaySent.", cause);
272             }
273         }
274 
275         closeStreamsGreaterThanLastKnownStreamId(lastKnownStream, remoteEndpoint);
276         return true;
277     }
278 
279     private void closeStreamsGreaterThanLastKnownStreamId(final int lastKnownStream,
280                                                           final DefaultEndpoint<?> endpoint) throws Http2Exception {
281         forEachActiveStream(new Http2StreamVisitor() {
282             @Override
283             public boolean visit(Http2Stream stream) {
284                 if (stream.id() > lastKnownStream && endpoint.isValidStreamId(stream.id())) {
285                     stream.close();
286                 }
287                 return true;
288             }
289         });
290     }
291 
292     /**
293      * Determine if {@link #streamMap} only contains the connection stream.
294      */
295     private boolean isStreamMapEmpty() {
296         return streamMap.size() == 1;
297     }
298 
299     /**
300      * Remove a stream from the {@link #streamMap}.
301      * @param stream the stream to remove.
302      * @param itr an iterator that may be pointing to the stream during iteration and {@link Iterator#remove()} will be
303      * used if non-{@code null}.
304      */
305     void removeStream(DefaultStream stream, Iterator<?> itr) {
306         final boolean removed;
307         if (itr == null) {
308             removed = streamMap.remove(stream.id()) != null;
309         } else {
310             itr.remove();
311             removed = true;
312         }
313 
314         if (removed) {
315             for (int i = 0; i < listeners.size(); i++) {
316                 try {
317                     listeners.get(i).onStreamRemoved(stream);
318                 } catch (Throwable cause) {
319                     logger.error("Caught Throwable from listener onStreamRemoved.", cause);
320                 }
321             }
322 
323             if (closePromise != null && isStreamMapEmpty()) {
324                 closePromise.trySuccess(null);
325             }
326         }
327     }
328 
329     static State activeState(int streamId, State initialState, boolean isLocal, boolean halfClosed)
330             throws Http2Exception {
331         switch (initialState) {
332         case IDLE:
333             return halfClosed ? isLocal ? HALF_CLOSED_LOCAL : HALF_CLOSED_REMOTE : OPEN;
334         case RESERVED_LOCAL:
335             return HALF_CLOSED_REMOTE;
336         case RESERVED_REMOTE:
337             return HALF_CLOSED_LOCAL;
338         default:
339             throw streamError(streamId, PROTOCOL_ERROR, "Attempting to open a stream in an invalid state: "
340                     + initialState);
341         }
342     }
343 
344     void notifyHalfClosed(Http2Stream stream) {
345         for (int i = 0; i < listeners.size(); i++) {
346             try {
347                 listeners.get(i).onStreamHalfClosed(stream);
348             } catch (Throwable cause) {
349                 logger.error("Caught Throwable from listener onStreamHalfClosed.", cause);
350             }
351         }
352     }
353 
354     void notifyClosed(Http2Stream stream) {
355         for (int i = 0; i < listeners.size(); i++) {
356             try {
357                 listeners.get(i).onStreamClosed(stream);
358             } catch (Throwable cause) {
359                 logger.error("Caught Throwable from listener onStreamClosed.", cause);
360             }
361         }
362     }
363 
364     @Override
365     public PropertyKey newKey() {
366         return propertyKeyRegistry.newKey();
367     }
368 
369     /**
370      * Verifies that the key is valid and returns it as the internal {@link DefaultPropertyKey} type.
371      *
372      * @throws NullPointerException if the key is {@code null}.
373      * @throws ClassCastException if the key is not of type {@link DefaultPropertyKey}.
374      * @throws IllegalArgumentException if the key was not created by this connection.
375      */
376     final DefaultPropertyKey verifyKey(PropertyKey key) {
377         return checkNotNull((DefaultPropertyKey) key, "key").verifyConnection(this);
378     }
379 
380     /**
381      * Simple stream implementation. Streams can be compared to each other by priority.
382      */
383     private class DefaultStream implements Http2Stream {
384         private static final byte META_STATE_SENT_RST = 1;
385         private static final byte META_STATE_SENT_HEADERS = 1 << 1;
386         private static final byte META_STATE_SENT_TRAILERS = 1 << 2;
387         private static final byte META_STATE_SENT_PUSHPROMISE = 1 << 3;
388         private static final byte META_STATE_RECV_HEADERS = 1 << 4;
389         private static final byte META_STATE_RECV_TRAILERS = 1 << 5;
390         private final int id;
391         private final PropertyMap properties = new PropertyMap();
392         private State state;
393         private byte metaState;
394 
395         DefaultStream(int id, State state) {
396             this.id = id;
397             this.state = state;
398         }
399 
400         @Override
401         public final int id() {
402             return id;
403         }
404 
405         @Override
406         public final State state() {
407             return state;
408         }
409 
410         @Override
411         public boolean isResetSent() {
412             return (metaState & META_STATE_SENT_RST) != 0;
413         }
414 
415         @Override
416         public Http2Stream resetSent() {
417             metaState |= META_STATE_SENT_RST;
418             return this;
419         }
420 
421         @Override
422         public Http2Stream headersSent(boolean isInformational) {
423             if (!isInformational) {
424                 metaState |= isHeadersSent() ? META_STATE_SENT_TRAILERS : META_STATE_SENT_HEADERS;
425             }
426             return this;
427         }
428 
429         @Override
430         public boolean isHeadersSent() {
431             return (metaState & META_STATE_SENT_HEADERS) != 0;
432         }
433 
434         @Override
435         public boolean isTrailersSent() {
436             return (metaState & META_STATE_SENT_TRAILERS) != 0;
437         }
438 
439         @Override
440         public Http2Stream headersReceived(boolean isInformational) {
441             if (!isInformational) {
442                 metaState |= isHeadersReceived() ? META_STATE_RECV_TRAILERS : META_STATE_RECV_HEADERS;
443             }
444             return this;
445         }
446 
447         @Override
448         public boolean isHeadersReceived() {
449             return (metaState & META_STATE_RECV_HEADERS) != 0;
450         }
451 
452         @Override
453         public boolean isTrailersReceived() {
454             return (metaState & META_STATE_RECV_TRAILERS) != 0;
455         }
456 
457         @Override
458         public Http2Stream pushPromiseSent() {
459             metaState |= META_STATE_SENT_PUSHPROMISE;
460             return this;
461         }
462 
463         @Override
464         public boolean isPushPromiseSent() {
465             return (metaState & META_STATE_SENT_PUSHPROMISE) != 0;
466         }
467 
468         @Override
469         public final <V> V setProperty(PropertyKey key, V value) {
470             return properties.add(verifyKey(key), value);
471         }
472 
473         @Override
474         public final <V> V getProperty(PropertyKey key) {
475             return properties.get(verifyKey(key));
476         }
477 
478         @Override
479         public final <V> V removeProperty(PropertyKey key) {
480             return properties.remove(verifyKey(key));
481         }
482 
483         @Override
484         public Http2Stream open(boolean halfClosed) throws Http2Exception {
485             state = activeState(id, state, isLocal(), halfClosed);
486             final DefaultEndpoint<? extends Http2FlowController> endpoint = createdBy();
487             if (!endpoint.canOpenStream()) {
488                 throw connectionError(PROTOCOL_ERROR, "Maximum active streams violated for this endpoint: " +
489                         endpoint.maxActiveStreams());
490             }
491 
492             activate();
493             return this;
494         }
495 
496         void activate() {
497             // If the stream is opened in a half-closed state, the headers must have either
498             // been sent if this is a local stream, or received if it is a remote stream.
499             if (state == HALF_CLOSED_LOCAL) {
500                 headersSent(/*isInformational*/ false);
501             } else if (state == HALF_CLOSED_REMOTE) {
502                 headersReceived(/*isInformational*/ false);
503             }
504             activeStreams.activate(this);
505         }
506 
507         Http2Stream close(Iterator<?> itr) {
508             if (state == CLOSED) {
509                 return this;
510             }
511 
512             state = CLOSED;
513 
514             --createdBy().numStreams;
515             activeStreams.deactivate(this, itr);
516             return this;
517         }
518 
519         @Override
520         public Http2Stream close() {
521             return close(null);
522         }
523 
524         @Override
525         public Http2Stream closeLocalSide() {
526             switch (state) {
527             case OPEN:
528                 state = HALF_CLOSED_LOCAL;
529                 notifyHalfClosed(this);
530                 break;
531             case HALF_CLOSED_LOCAL:
532                 break;
533             default:
534                 close();
535                 break;
536             }
537             return this;
538         }
539 
540         @Override
541         public Http2Stream closeRemoteSide() {
542             switch (state) {
543             case OPEN:
544                 state = HALF_CLOSED_REMOTE;
545                 notifyHalfClosed(this);
546                 break;
547             case HALF_CLOSED_REMOTE:
548                 break;
549             default:
550                 close();
551                 break;
552             }
553             return this;
554         }
555 
556         DefaultEndpoint<? extends Http2FlowController> createdBy() {
557             return localEndpoint.isValidStreamId(id) ? localEndpoint : remoteEndpoint;
558         }
559 
560         final boolean isLocal() {
561             return localEndpoint.isValidStreamId(id);
562         }
563 
564         /**
565          * Provides the lazy initialization for the {@link DefaultStream} data map.
566          */
567         private class PropertyMap {
568             Object[] values = EmptyArrays.EMPTY_OBJECTS;
569 
570             <V> V add(DefaultPropertyKey key, V value) {
571                 resizeIfNecessary(key.index);
572                 @SuppressWarnings("unchecked")
573                 V prevValue = (V) values[key.index];
574                 values[key.index] = value;
575                 return prevValue;
576             }
577 
578             @SuppressWarnings("unchecked")
579             <V> V get(DefaultPropertyKey key) {
580                 if (key.index >= values.length) {
581                     return null;
582                 }
583                 return (V) values[key.index];
584             }
585 
586             @SuppressWarnings("unchecked")
587             <V> V remove(DefaultPropertyKey key) {
588                 V prevValue = null;
589                 if (key.index < values.length) {
590                     prevValue = (V) values[key.index];
591                     values[key.index] = null;
592                 }
593                 return prevValue;
594             }
595 
596             void resizeIfNecessary(int index) {
597                 if (index >= values.length) {
598                     values = Arrays.copyOf(values, propertyKeyRegistry.size());
599                 }
600             }
601         }
602     }
603 
604     /**
605      * Stream class representing the connection, itself.
606      */
607     private final class ConnectionStream extends DefaultStream {
608         ConnectionStream() {
609             super(CONNECTION_STREAM_ID, IDLE);
610         }
611 
612         @Override
613         public boolean isResetSent() {
614             return false;
615         }
616 
617         @Override
618         DefaultEndpoint<? extends Http2FlowController> createdBy() {
619             return null;
620         }
621 
622         @Override
623         public Http2Stream resetSent() {
624             throw new UnsupportedOperationException();
625         }
626 
627         @Override
628         public Http2Stream open(boolean halfClosed) {
629             throw new UnsupportedOperationException();
630         }
631 
632         @Override
633         public Http2Stream close() {
634             throw new UnsupportedOperationException();
635         }
636 
637         @Override
638         public Http2Stream closeLocalSide() {
639             throw new UnsupportedOperationException();
640         }
641 
642         @Override
643         public Http2Stream closeRemoteSide() {
644             throw new UnsupportedOperationException();
645         }
646 
647         @Override
648         public Http2Stream headersSent(boolean isInformational) {
649             throw new UnsupportedOperationException();
650         }
651 
652         @Override
653         public boolean isHeadersSent() {
654             throw new UnsupportedOperationException();
655         }
656 
657         @Override
658         public Http2Stream pushPromiseSent() {
659             throw new UnsupportedOperationException();
660         }
661 
662         @Override
663         public boolean isPushPromiseSent() {
664             throw new UnsupportedOperationException();
665         }
666     }
667 
668     /**
669      * Simple endpoint implementation.
670      */
671     private final class DefaultEndpoint<F extends Http2FlowController> implements Endpoint<F> {
672         private final boolean server;
673         /**
674          * The minimum stream ID allowed when creating the next stream. This only applies at the time the stream is
675          * created. If the ID of the stream being created is less than this value, stream creation will fail. Upon
676          * successful creation of a stream, this value is incremented to the next valid stream ID.
677          */
678         private int nextStreamIdToCreate;
679         /**
680          * Used for reservation of stream IDs. Stream IDs can be reserved in advance by applications before the streams
681          * are actually created.  For example, applications may choose to buffer stream creation attempts as a way of
682          * working around {@code SETTINGS_MAX_CONCURRENT_STREAMS}, in which case they will reserve stream IDs for each
683          * buffered stream.
684          */
685         private int nextReservationStreamId;
686         private int lastStreamKnownByPeer = -1;
687         private boolean pushToAllowed;
688         private F flowController;
689         private int maxStreams;
690         private int maxActiveStreams;
691         private final int maxReservedStreams;
692         // Fields accessed by inner classes
693         int numActiveStreams;
694         int numStreams;
695 
696         DefaultEndpoint(boolean server, int maxReservedStreams) {
697             this.server = server;
698 
699             // Determine the starting stream ID for this endpoint. Client-initiated streams
700             // are odd and server-initiated streams are even. Zero is reserved for the
701             // connection. Stream 1 is reserved client-initiated stream for responding to an
702             // upgrade from HTTP 1.1.
703             if (server) {
704                 nextStreamIdToCreate = 2;
705                 nextReservationStreamId = 0;
706             } else {
707                 nextStreamIdToCreate = 1;
708                 // For manually created client-side streams, 1 is reserved for HTTP upgrade, so start at 3.
709                 nextReservationStreamId = 1;
710             }
711 
712             // Push is disallowed by default for servers and allowed for clients.
713             pushToAllowed = !server;
714             maxActiveStreams = MAX_VALUE;
715             this.maxReservedStreams = checkPositiveOrZero(maxReservedStreams, "maxReservedStreams");
716             updateMaxStreams();
717         }
718 
719         @Override
720         public int incrementAndGetNextStreamId() {
721             return nextReservationStreamId >= 0 ? nextReservationStreamId += 2 : nextReservationStreamId;
722         }
723 
724         private void incrementExpectedStreamId(int streamId) {
725             if (streamId > nextReservationStreamId && nextReservationStreamId >= 0) {
726                 nextReservationStreamId = streamId;
727             }
728             nextStreamIdToCreate = streamId + 2;
729             ++numStreams;
730         }
731 
732         @Override
733         public boolean isValidStreamId(int streamId) {
734             return streamId > 0 && server == ((streamId & 1) == 0);
735         }
736 
737         @Override
738         public boolean mayHaveCreatedStream(int streamId) {
739             return isValidStreamId(streamId) && streamId <= lastStreamCreated();
740         }
741 
742         @Override
743         public boolean canOpenStream() {
744             return numActiveStreams < maxActiveStreams;
745         }
746 
747         @Override
748         public DefaultStream createStream(int streamId, boolean halfClosed) throws Http2Exception {
749             State state = activeState(streamId, IDLE, isLocal(), halfClosed);
750 
751             checkNewStreamAllowed(streamId, state);
752 
753             // Create and initialize the stream.
754             DefaultStream stream = new DefaultStream(streamId, state);
755 
756             incrementExpectedStreamId(streamId);
757 
758             addStream(stream);
759 
760             stream.activate();
761             return stream;
762         }
763 
764         @Override
765         public boolean created(Http2Stream stream) {
766             return stream instanceof DefaultStream && ((DefaultStream) stream).createdBy() == this;
767         }
768 
769         @Override
770         public boolean isServer() {
771             return server;
772         }
773 
774         @Override
775         public DefaultStream reservePushStream(int streamId, Http2Stream parent) throws Http2Exception {
776             if (parent == null) {
777                 throw connectionError(PROTOCOL_ERROR, "Parent stream missing");
778             }
779             if (isLocal() ? !parent.state().localSideOpen() : !parent.state().remoteSideOpen()) {
780                 throw connectionError(PROTOCOL_ERROR, "Stream %d is not open for sending push promise", parent.id());
781             }
782             if (!opposite().allowPushTo()) {
783                 throw connectionError(PROTOCOL_ERROR, "Server push not allowed to opposite endpoint");
784             }
785             State state = isLocal() ? RESERVED_LOCAL : RESERVED_REMOTE;
786             checkNewStreamAllowed(streamId, state);
787 
788             // Create and initialize the stream.
789             DefaultStream stream = new DefaultStream(streamId, state);
790 
791             incrementExpectedStreamId(streamId);
792 
793             // Register the stream.
794             addStream(stream);
795             return stream;
796         }
797 
798         private void addStream(DefaultStream stream) {
799             // Add the stream to the map and priority tree.
800             streamMap.put(stream.id(), stream);
801 
802             // Notify the listeners of the event.
803             for (int i = 0; i < listeners.size(); i++) {
804                 try {
805                     listeners.get(i).onStreamAdded(stream);
806                 } catch (Throwable cause) {
807                     logger.error("Caught Throwable from listener onStreamAdded.", cause);
808                 }
809             }
810         }
811 
812         @Override
813         public void allowPushTo(boolean allow) {
814             if (allow && server) {
815                 throw new IllegalArgumentException("Servers do not allow push");
816             }
817             pushToAllowed = allow;
818         }
819 
820         @Override
821         public boolean allowPushTo() {
822             return pushToAllowed;
823         }
824 
825         @Override
826         public int numActiveStreams() {
827             return numActiveStreams;
828         }
829 
830         @Override
831         public int maxActiveStreams() {
832             return maxActiveStreams;
833         }
834 
835         @Override
836         public void maxActiveStreams(int maxActiveStreams) {
837             this.maxActiveStreams = maxActiveStreams;
838             updateMaxStreams();
839         }
840 
841         @Override
842         public int lastStreamCreated() {
843             return nextStreamIdToCreate > 1 ? nextStreamIdToCreate - 2 : 0;
844         }
845 
846         @Override
847         public int lastStreamKnownByPeer() {
848             return lastStreamKnownByPeer;
849         }
850 
851         private void lastStreamKnownByPeer(int lastKnownStream) {
852             lastStreamKnownByPeer = lastKnownStream;
853         }
854 
855         @Override
856         public F flowController() {
857             return flowController;
858         }
859 
860         @Override
861         public void flowController(F flowController) {
862             this.flowController = checkNotNull(flowController, "flowController");
863         }
864 
865         @Override
866         public Endpoint<? extends Http2FlowController> opposite() {
867             return isLocal() ? remoteEndpoint : localEndpoint;
868         }
869 
870         private void updateMaxStreams() {
871             maxStreams = (int) Math.min(MAX_VALUE, (long) maxActiveStreams + maxReservedStreams);
872         }
873 
874         private void checkNewStreamAllowed(int streamId, State state) throws Http2Exception {
875             assert state != IDLE;
876             if (lastStreamKnownByPeer >= 0 && streamId > lastStreamKnownByPeer) {
877                 throw streamError(streamId, REFUSED_STREAM,
878                         "Cannot create stream %d greater than Last-Stream-ID %d from GOAWAY.",
879                         streamId, lastStreamKnownByPeer);
880             }
881             if (!isValidStreamId(streamId)) {
882                 if (streamId < 0) {
883                     throw new Http2NoMoreStreamIdsException();
884                 }
885                 throw connectionError(PROTOCOL_ERROR, "Request stream %d is not correct for %s connection", streamId,
886                         server ? "server" : "client");
887             }
888             // This check must be after all id validated checks, but before the max streams check because it may be
889             // recoverable to some degree for handling frames which can be sent on closed streams.
890             if (streamId < nextStreamIdToCreate) {
891                 throw closedStreamError(PROTOCOL_ERROR, "Request stream %d is behind the next expected stream %d",
892                         streamId, nextStreamIdToCreate);
893             }
894             if (nextStreamIdToCreate <= 0) {
895                 // We exhausted the stream id space that we  can use. Let's signal this back but also signal that
896                 // we still may want to process active streams.
897                 throw new Http2Exception(REFUSED_STREAM, "Stream IDs are exhausted for this endpoint.",
898                         Http2Exception.ShutdownHint.GRACEFUL_SHUTDOWN);
899             }
900             boolean isReserved = state == RESERVED_LOCAL || state == RESERVED_REMOTE;
901             if (!isReserved && !canOpenStream() || isReserved && numStreams >= maxStreams) {
902                 throw streamError(streamId, REFUSED_STREAM, "Maximum active streams violated for this endpoint: " +
903                         (isReserved ? maxStreams : maxActiveStreams));
904             }
905             if (isClosed()) {
906                 throw connectionError(INTERNAL_ERROR, "Attempted to create stream id %d after connection was closed",
907                                       streamId);
908             }
909         }
910 
911         private boolean isLocal() {
912             return this == localEndpoint;
913         }
914     }
915 
916     /**
917      * Allows events which would modify the collection of active streams to be queued while iterating via {@link
918      * #forEachActiveStream(Http2StreamVisitor)}.
919      */
920     interface Event {
921         /**
922          * Trigger the original intention of this event. Expect to modify the active streams list.
923          * <p/>
924          * If a {@link RuntimeException} object is thrown it will be logged and <strong>not propagated</strong>.
925          * Throwing from this method is not supported and is considered a programming error.
926          */
927         void process();
928     }
929 
930     /**
931      * Manages the list of currently active streams.  Queues any {@link Event}s that would modify the list of
932      * active streams in order to prevent modification while iterating.
933      */
934     private final class ActiveStreams {
935         private final List<Listener> listeners;
936         private final Queue<Event> pendingEvents = new ArrayDeque<Event>(4);
937         private final Set<Http2Stream> streams = new LinkedHashSet<Http2Stream>();
938         private int pendingIterations;
939 
940         ActiveStreams(List<Listener> listeners) {
941             this.listeners = listeners;
942         }
943 
944         public int size() {
945             return streams.size();
946         }
947 
948         public void activate(final DefaultStream stream) {
949             if (allowModifications()) {
950                 addToActiveStreams(stream);
951             } else {
952                 pendingEvents.add(new Event() {
953                     @Override
954                     public void process() {
955                         addToActiveStreams(stream);
956                     }
957                 });
958             }
959         }
960 
961         public void deactivate(final DefaultStream stream, final Iterator<?> itr) {
962             if (allowModifications() || itr != null) {
963                 removeFromActiveStreams(stream, itr);
964             } else {
965                 pendingEvents.add(new Event() {
966                     @Override
967                     public void process() {
968                         removeFromActiveStreams(stream, itr);
969                     }
970                 });
971             }
972         }
973 
974         public Http2Stream forEachActiveStream(Http2StreamVisitor visitor) throws Http2Exception {
975             incrementPendingIterations();
976             try {
977                 for (Http2Stream stream : streams) {
978                     if (!visitor.visit(stream)) {
979                         return stream;
980                     }
981                 }
982                 return null;
983             } finally {
984                 decrementPendingIterations();
985             }
986         }
987 
988         void addToActiveStreams(DefaultStream stream) {
989             if (streams.add(stream)) {
990                 // Update the number of active streams initiated by the endpoint.
991                 stream.createdBy().numActiveStreams++;
992 
993                 for (int i = 0; i < listeners.size(); i++) {
994                     try {
995                         listeners.get(i).onStreamActive(stream);
996                     } catch (Throwable cause) {
997                         logger.error("Caught Throwable from listener onStreamActive.", cause);
998                     }
999                 }
1000             }
1001         }
1002 
1003         void removeFromActiveStreams(DefaultStream stream, Iterator<?> itr) {
1004             if (streams.remove(stream)) {
1005                 // Update the number of active streams initiated by the endpoint.
1006                 stream.createdBy().numActiveStreams--;
1007                 notifyClosed(stream);
1008             }
1009             removeStream(stream, itr);
1010         }
1011 
1012         boolean allowModifications() {
1013             return pendingIterations == 0;
1014         }
1015 
1016         void incrementPendingIterations() {
1017             ++pendingIterations;
1018         }
1019 
1020         void decrementPendingIterations() {
1021             --pendingIterations;
1022             if (allowModifications()) {
1023                 for (;;) {
1024                     Event event = pendingEvents.poll();
1025                     if (event == null) {
1026                         break;
1027                     }
1028                     try {
1029                         event.process();
1030                     } catch (Throwable cause) {
1031                         logger.error("Caught Throwable while processing pending ActiveStreams$Event.", cause);
1032                     }
1033                 }
1034             }
1035         }
1036     }
1037 
1038     /**
1039      * Implementation of {@link PropertyKey} that specifies the index position of the property.
1040      */
1041     final class DefaultPropertyKey implements PropertyKey {
1042         final int index;
1043 
1044         DefaultPropertyKey(int index) {
1045             this.index = index;
1046         }
1047 
1048         DefaultPropertyKey verifyConnection(Http2Connection connection) {
1049             if (connection != DefaultHttp2Connection.this) {
1050                 throw new IllegalArgumentException("Using a key that was not created by this connection");
1051             }
1052             return this;
1053         }
1054     }
1055 
1056     /**
1057      * A registry of all stream property keys known by this connection.
1058      */
1059     private final class PropertyKeyRegistry {
1060         /**
1061          * Initial size of 4 because the default configuration currently has 3 listeners
1062          * (local/remote flow controller and {@link StreamByteDistributor}) and we leave room for 1 extra.
1063          * We could be more aggressive but the ArrayList resize will double the size if we are too small.
1064          */
1065         final List<DefaultPropertyKey> keys = new ArrayList<DefaultPropertyKey>(4);
1066 
1067         /**
1068          * Registers a new property key.
1069          */
1070         DefaultPropertyKey newKey() {
1071             DefaultPropertyKey key = new DefaultPropertyKey(keys.size());
1072             keys.add(key);
1073             return key;
1074         }
1075 
1076         int size() {
1077             return keys.size();
1078         }
1079     }
1080 }