DefaultHttp2Connection.java

/*
 * Copyright 2014 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License, version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at:
 *
 * https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package io.netty.handler.codec.http2;

import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http2.Http2Stream.State;
import io.netty.util.collection.IntObjectHashMap;
import io.netty.util.collection.IntObjectMap;
import io.netty.util.collection.IntObjectMap.PrimitiveEntry;
import io.netty.util.concurrent.Promise;
import io.netty.util.concurrent.UnaryPromiseNotifier;
import io.netty.util.internal.EmptyArrays;
import io.netty.util.internal.UnstableApi;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Queue;
import java.util.Set;

import static io.netty.handler.codec.http2.Http2CodecUtil.CONNECTION_STREAM_ID;
import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_MAX_RESERVED_STREAMS;
import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
import static io.netty.handler.codec.http2.Http2Error.REFUSED_STREAM;
import static io.netty.handler.codec.http2.Http2Exception.closedStreamError;
import static io.netty.handler.codec.http2.Http2Exception.connectionError;
import static io.netty.handler.codec.http2.Http2Exception.streamError;
import static io.netty.handler.codec.http2.Http2Stream.State.CLOSED;
import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_REMOTE;
import static io.netty.handler.codec.http2.Http2Stream.State.IDLE;
import static io.netty.handler.codec.http2.Http2Stream.State.OPEN;
import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_LOCAL;
import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_REMOTE;
import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
import static java.lang.Integer.MAX_VALUE;
import static java.util.Objects.requireNonNull;

/**
 * Simple implementation of {@link Http2Connection}.
 */
@UnstableApi
public class DefaultHttp2Connection implements Http2Connection {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultHttp2Connection.class);
    // Fields accessed by inner classes
    final IntObjectMap<Http2Stream> streamMap = new IntObjectHashMap<Http2Stream>();
    final PropertyKeyRegistry propertyKeyRegistry = new PropertyKeyRegistry();
    final ConnectionStream connectionStream = new ConnectionStream();
    final DefaultEndpoint<Http2LocalFlowController> localEndpoint;
    final DefaultEndpoint<Http2RemoteFlowController> remoteEndpoint;

    /**
     * We chose a {@link List} over a {@link Set} to avoid allocating an {@link Iterator} objects when iterating over
     * the listeners.
     * <p>
     * Initial size of 4 because the default configuration currently has 3 listeners
     * (local/remote flow controller and {@link StreamByteDistributor}) and we leave room for 1 extra.
     * We could be more aggressive but the ArrayList resize will double the size if we are too small.
     */
    final List<Listener> listeners = new ArrayList<>(4);
    final ActiveStreams activeStreams;
    Promise<Void> closePromise;

    /**
     * Creates a new connection with the given settings.
     * @param server whether or not this end-point is the server-side of the HTTP/2 connection.
     */
    public DefaultHttp2Connection(boolean server) {
        this(server, DEFAULT_MAX_RESERVED_STREAMS);
    }

    /**
     * Creates a new connection with the given settings.
     * @param server whether or not this end-point is the server-side of the HTTP/2 connection.
     * @param maxReservedStreams The maximum amount of streams which can exist in the reserved state for each endpoint.
     */
    public DefaultHttp2Connection(boolean server, int maxReservedStreams) {
        activeStreams = new ActiveStreams(listeners);
        // Reserved streams are excluded from the SETTINGS_MAX_CONCURRENT_STREAMS limit according to [1] and the RFC
        // doesn't define a way to communicate the limit on reserved streams. We rely upon the peer to send RST_STREAM
        // in response to any locally enforced limits being exceeded [2].
        // [1] https://tools.ietf.org/html/rfc7540#section-5.1.2
        // [2] https://tools.ietf.org/html/rfc7540#section-8.2.2
        localEndpoint = new DefaultEndpoint<>(server, server ? MAX_VALUE : maxReservedStreams);
        remoteEndpoint = new DefaultEndpoint<>(!server, maxReservedStreams);

        // Add the connection stream to the map.
        streamMap.put(connectionStream.id(), connectionStream);
    }

    /**
     * Determine if {@link #close(Promise)} has been called and no more streams are allowed to be created.
     */
    final boolean isClosed() {
        return closePromise != null;
    }

    @Override
    public void close(final Promise<Void> promise) {
        requireNonNull(promise, "promise");
        // Since we allow this method to be called multiple times, we must make sure that all the promises are notified
        // when all streams are removed and the close operation completes.
        if (closePromise != null) {
            if (closePromise == promise) {
                // Do nothing
            } else {
                closePromise.asFuture().addListener(new UnaryPromiseNotifier<>(promise));
            }
        } else {
            closePromise = promise;
        }
        if (isStreamMapEmpty()) {
            promise.trySuccess(null);
            return;
        }

        Iterator<PrimitiveEntry<Http2Stream>> itr = streamMap.entries().iterator();
        // We must take care while iterating the streamMap as to not modify while iterating in case there are other code
        // paths iterating over the active streams.
        if (activeStreams.allowModifications()) {
            activeStreams.incrementPendingIterations();
            try {
                while (itr.hasNext()) {
                    DefaultStream stream = (DefaultStream) itr.next().value();
                    if (stream.id() != CONNECTION_STREAM_ID) {
                        // If modifications of the activeStream map is allowed, then a stream close operation will also
                        // modify the streamMap. Pass the iterator in so that remove will be called to prevent
                        // concurrent modification exceptions.
                        stream.close(itr);
                    }
                }
            } finally {
                activeStreams.decrementPendingIterations();
            }
        } else {
            while (itr.hasNext()) {
                Http2Stream stream = itr.next().value();
                if (stream.id() != CONNECTION_STREAM_ID) {
                    // We are not allowed to make modifications, so the close calls will be executed after this
                    // iteration completes.
                    stream.close();
                }
            }
        }
    }

    @Override
    public void addListener(Listener listener) {
        listeners.add(listener);
    }

    @Override
    public void removeListener(Listener listener) {
        listeners.remove(listener);
    }

    @Override
    public boolean isServer() {
        return localEndpoint.isServer();
    }

    @Override
    public Http2Stream connectionStream() {
        return connectionStream;
    }

    @Override
    public Http2Stream stream(int streamId) {
        return streamMap.get(streamId);
    }

    @Override
    public boolean streamMayHaveExisted(int streamId) {
        return remoteEndpoint.mayHaveCreatedStream(streamId) || localEndpoint.mayHaveCreatedStream(streamId);
    }

    @Override
    public int numActiveStreams() {
        return activeStreams.size();
    }

    @Override
    public Http2Stream forEachActiveStream(Http2StreamVisitor visitor) throws Http2Exception {
        return activeStreams.forEachActiveStream(visitor);
    }

    @Override
    public Endpoint<Http2LocalFlowController> local() {
        return localEndpoint;
    }

    @Override
    public Endpoint<Http2RemoteFlowController> remote() {
        return remoteEndpoint;
    }

    @Override
    public boolean goAwayReceived() {
        return localEndpoint.lastStreamKnownByPeer >= 0;
    }

    @Override
    public void goAwayReceived(final int lastKnownStream, long errorCode, ByteBuf debugData) throws Http2Exception {
        if (localEndpoint.lastStreamKnownByPeer() >= 0 && localEndpoint.lastStreamKnownByPeer() < lastKnownStream) {
            throw connectionError(PROTOCOL_ERROR, "lastStreamId MUST NOT increase. Current value: %d new value: %d",
                    localEndpoint.lastStreamKnownByPeer(), lastKnownStream);
        }

        localEndpoint.lastStreamKnownByPeer(lastKnownStream);
        for (int i = 0; i < listeners.size(); ++i) {
            try {
                listeners.get(i).onGoAwayReceived(lastKnownStream, errorCode, debugData);
            } catch (Throwable cause) {
                logger.error("Caught Throwable from listener onGoAwayReceived.", cause);
            }
        }

        closeStreamsGreaterThanLastKnownStreamId(lastKnownStream, localEndpoint);
    }

    @Override
    public boolean goAwaySent() {
        return remoteEndpoint.lastStreamKnownByPeer >= 0;
    }

    @Override
    public boolean goAwaySent(final int lastKnownStream, long errorCode, ByteBuf debugData) throws Http2Exception {
        if (remoteEndpoint.lastStreamKnownByPeer() >= 0) {
            // Protect against re-entrancy. Could happen if writing the frame fails, and error handling
            // treating this is a connection handler and doing a graceful shutdown...
            if (lastKnownStream == remoteEndpoint.lastStreamKnownByPeer()) {
                return false;
            }
            if (lastKnownStream > remoteEndpoint.lastStreamKnownByPeer()) {
                throw connectionError(PROTOCOL_ERROR, "Last stream identifier must not increase between " +
                                "sending multiple GOAWAY frames (was '%d', is '%d').",
                        remoteEndpoint.lastStreamKnownByPeer(), lastKnownStream);
            }
        }

        remoteEndpoint.lastStreamKnownByPeer(lastKnownStream);
        for (int i = 0; i < listeners.size(); ++i) {
            try {
                listeners.get(i).onGoAwaySent(lastKnownStream, errorCode, debugData);
            } catch (Throwable cause) {
                logger.error("Caught Throwable from listener onGoAwaySent.", cause);
            }
        }

        closeStreamsGreaterThanLastKnownStreamId(lastKnownStream, remoteEndpoint);
        return true;
    }

    private void closeStreamsGreaterThanLastKnownStreamId(final int lastKnownStream,
                                                          final DefaultEndpoint<?> endpoint) throws Http2Exception {
        forEachActiveStream(stream -> {
            if (stream.id() > lastKnownStream && endpoint.isValidStreamId(stream.id())) {
                stream.close();
            }
            return true;
        });
    }

    /**
     * Determine if {@link #streamMap} only contains the connection stream.
     */
    private boolean isStreamMapEmpty() {
        return streamMap.size() == 1;
    }

    /**
     * Remove a stream from the {@link #streamMap}.
     * @param stream the stream to remove.
     * @param itr an iterator that may be pointing to the stream during iteration and {@link Iterator#remove()} will be
     * used if non-{@code null}.
     */
    void removeStream(DefaultStream stream, Iterator<?> itr) {
        final boolean removed;
        if (itr == null) {
            removed = streamMap.remove(stream.id()) != null;
        } else {
            itr.remove();
            removed = true;
        }

        if (removed) {
            for (int i = 0; i < listeners.size(); i++) {
                try {
                    listeners.get(i).onStreamRemoved(stream);
                } catch (Throwable cause) {
                    logger.error("Caught Throwable from listener onStreamRemoved.", cause);
                }
            }

            if (closePromise != null && isStreamMapEmpty()) {
                closePromise.trySuccess(null);
            }
        }
    }

    static State activeState(int streamId, State initialState, boolean isLocal, boolean halfClosed)
            throws Http2Exception {
        switch (initialState) {
        case IDLE:
            return halfClosed ? isLocal ? HALF_CLOSED_LOCAL : HALF_CLOSED_REMOTE : OPEN;
        case RESERVED_LOCAL:
            return HALF_CLOSED_REMOTE;
        case RESERVED_REMOTE:
            return HALF_CLOSED_LOCAL;
        default:
            throw streamError(streamId, PROTOCOL_ERROR, "Attempting to open a stream in an invalid state: "
                    + initialState);
        }
    }

    void notifyHalfClosed(Http2Stream stream) {
        for (int i = 0; i < listeners.size(); i++) {
            try {
                listeners.get(i).onStreamHalfClosed(stream);
            } catch (Throwable cause) {
                logger.error("Caught Throwable from listener onStreamHalfClosed.", cause);
            }
        }
    }

    void notifyClosed(Http2Stream stream) {
        for (int i = 0; i < listeners.size(); i++) {
            try {
                listeners.get(i).onStreamClosed(stream);
            } catch (Throwable cause) {
                logger.error("Caught Throwable from listener onStreamClosed.", cause);
            }
        }
    }

    @Override
    public PropertyKey newKey() {
        return propertyKeyRegistry.newKey();
    }

    /**
     * Verifies that the key is valid and returns it as the internal {@link DefaultPropertyKey} type.
     *
     * @throws NullPointerException if the key is {@code null}.
     * @throws ClassCastException if the key is not of type {@link DefaultPropertyKey}.
     * @throws IllegalArgumentException if the key was not created by this connection.
     */
    final DefaultPropertyKey verifyKey(PropertyKey key) {
        return requireNonNull((DefaultPropertyKey) key, "key").verifyConnection(this);
    }

    /**
     * Simple stream implementation. Streams can be compared to each other by priority.
     */
    private class DefaultStream implements Http2Stream {
        private static final byte META_STATE_SENT_RST = 1;
        private static final byte META_STATE_SENT_HEADERS = 1 << 1;
        private static final byte META_STATE_SENT_TRAILERS = 1 << 2;
        private static final byte META_STATE_SENT_PUSHPROMISE = 1 << 3;
        private static final byte META_STATE_RECV_HEADERS = 1 << 4;
        private static final byte META_STATE_RECV_TRAILERS = 1 << 5;
        private final int id;
        private final PropertyMap properties = new PropertyMap();
        private State state;
        private byte metaState;

        DefaultStream(int id, State state) {
            this.id = id;
            this.state = state;
        }

        @Override
        public final int id() {
            return id;
        }

        @Override
        public final State state() {
            return state;
        }

        @Override
        public boolean isResetSent() {
            return (metaState & META_STATE_SENT_RST) != 0;
        }

        @Override
        public Http2Stream resetSent() {
            metaState |= META_STATE_SENT_RST;
            return this;
        }

        @Override
        public Http2Stream headersSent(boolean isInformational) {
            if (!isInformational) {
                metaState |= isHeadersSent() ? META_STATE_SENT_TRAILERS : META_STATE_SENT_HEADERS;
            }
            return this;
        }

        @Override
        public boolean isHeadersSent() {
            return (metaState & META_STATE_SENT_HEADERS) != 0;
        }

        @Override
        public boolean isTrailersSent() {
            return (metaState & META_STATE_SENT_TRAILERS) != 0;
        }

        @Override
        public Http2Stream headersReceived(boolean isInformational) {
            if (!isInformational) {
                metaState |= isHeadersReceived() ? META_STATE_RECV_TRAILERS : META_STATE_RECV_HEADERS;
            }
            return this;
        }

        @Override
        public boolean isHeadersReceived() {
            return (metaState & META_STATE_RECV_HEADERS) != 0;
        }

        @Override
        public boolean isTrailersReceived() {
            return (metaState & META_STATE_RECV_TRAILERS) != 0;
        }

        @Override
        public Http2Stream pushPromiseSent() {
            metaState |= META_STATE_SENT_PUSHPROMISE;
            return this;
        }

        @Override
        public boolean isPushPromiseSent() {
            return (metaState & META_STATE_SENT_PUSHPROMISE) != 0;
        }

        @Override
        public final <V> V setProperty(PropertyKey key, V value) {
            return properties.add(verifyKey(key), value);
        }

        @Override
        public final <V> V getProperty(PropertyKey key) {
            return properties.get(verifyKey(key));
        }

        @Override
        public final <V> V removeProperty(PropertyKey key) {
            return properties.remove(verifyKey(key));
        }

        @Override
        public Http2Stream open(boolean halfClosed) throws Http2Exception {
            state = activeState(id, state, isLocal(), halfClosed);
            final DefaultEndpoint<? extends Http2FlowController> endpoint = createdBy();
            if (!endpoint.canOpenStream()) {
                throw connectionError(PROTOCOL_ERROR, "Maximum active streams violated for this endpoint: " +
                        endpoint.maxActiveStreams());
            }

            activate();
            return this;
        }

        void activate() {
            // If the stream is opened in a half-closed state, the headers must have either
            // been sent if this is a local stream, or received if it is a remote stream.
            if (state == HALF_CLOSED_LOCAL) {
                headersSent(/*isInformational*/ false);
            } else if (state == HALF_CLOSED_REMOTE) {
                headersReceived(/*isInformational*/ false);
            }
            activeStreams.activate(this);
        }

        Http2Stream close(Iterator<?> itr) {
            if (state == CLOSED) {
                return this;
            }

            state = CLOSED;

            --createdBy().numStreams;
            activeStreams.deactivate(this, itr);
            return this;
        }

        @Override
        public Http2Stream close() {
            return close(null);
        }

        @Override
        public Http2Stream closeLocalSide() {
            switch (state) {
            case OPEN:
                state = HALF_CLOSED_LOCAL;
                notifyHalfClosed(this);
                break;
            case HALF_CLOSED_LOCAL:
                break;
            default:
                close();
                break;
            }
            return this;
        }

        @Override
        public Http2Stream closeRemoteSide() {
            switch (state) {
            case OPEN:
                state = HALF_CLOSED_REMOTE;
                notifyHalfClosed(this);
                break;
            case HALF_CLOSED_REMOTE:
                break;
            default:
                close();
                break;
            }
            return this;
        }

        DefaultEndpoint<? extends Http2FlowController> createdBy() {
            return localEndpoint.isValidStreamId(id) ? localEndpoint : remoteEndpoint;
        }

        final boolean isLocal() {
            return localEndpoint.isValidStreamId(id);
        }

        /**
         * Provides the lazy initialization for the {@link DefaultStream} data map.
         */
        private class PropertyMap {
            Object[] values = EmptyArrays.EMPTY_OBJECTS;

            <V> V add(DefaultPropertyKey key, V value) {
                resizeIfNecessary(key.index);
                @SuppressWarnings("unchecked")
                V prevValue = (V) values[key.index];
                values[key.index] = value;
                return prevValue;
            }

            @SuppressWarnings("unchecked")
            <V> V get(DefaultPropertyKey key) {
                if (key.index >= values.length) {
                    return null;
                }
                return (V) values[key.index];
            }

            @SuppressWarnings("unchecked")
            <V> V remove(DefaultPropertyKey key) {
                V prevValue = null;
                if (key.index < values.length) {
                    prevValue = (V) values[key.index];
                    values[key.index] = null;
                }
                return prevValue;
            }

            void resizeIfNecessary(int index) {
                if (index >= values.length) {
                    values = Arrays.copyOf(values, propertyKeyRegistry.size());
                }
            }
        }
    }

    /**
     * Stream class representing the connection, itself.
     */
    private final class ConnectionStream extends DefaultStream {
        ConnectionStream() {
            super(CONNECTION_STREAM_ID, IDLE);
        }

        @Override
        public boolean isResetSent() {
            return false;
        }

        @Override
        DefaultEndpoint<? extends Http2FlowController> createdBy() {
            return null;
        }

        @Override
        public Http2Stream resetSent() {
            throw new UnsupportedOperationException();
        }

        @Override
        public Http2Stream open(boolean halfClosed) {
            throw new UnsupportedOperationException();
        }

        @Override
        public Http2Stream close() {
            throw new UnsupportedOperationException();
        }

        @Override
        public Http2Stream closeLocalSide() {
            throw new UnsupportedOperationException();
        }

        @Override
        public Http2Stream closeRemoteSide() {
            throw new UnsupportedOperationException();
        }

        @Override
        public Http2Stream headersSent(boolean isInformational) {
            throw new UnsupportedOperationException();
        }

        @Override
        public boolean isHeadersSent() {
            throw new UnsupportedOperationException();
        }

        @Override
        public Http2Stream pushPromiseSent() {
            throw new UnsupportedOperationException();
        }

        @Override
        public boolean isPushPromiseSent() {
            throw new UnsupportedOperationException();
        }
    }

    /**
     * Simple endpoint implementation.
     */
    private final class DefaultEndpoint<F extends Http2FlowController> implements Endpoint<F> {
        private final boolean server;
        /**
         * The minimum stream ID allowed when creating the next stream. This only applies at the time the stream is
         * created. If the ID of the stream being created is less than this value, stream creation will fail. Upon
         * successful creation of a stream, this value is incremented to the next valid stream ID.
         */
        private int nextStreamIdToCreate;
        /**
         * Used for reservation of stream IDs. Stream IDs can be reserved in advance by applications before the streams
         * are actually created.  For example, applications may choose to buffer stream creation attempts as a way of
         * working around {@code SETTINGS_MAX_CONCURRENT_STREAMS}, in which case they will reserve stream IDs for each
         * buffered stream.
         */
        private int nextReservationStreamId;
        private int lastStreamKnownByPeer = -1;
        private boolean pushToAllowed;
        private F flowController;
        private int maxStreams;
        private int maxActiveStreams;
        private final int maxReservedStreams;
        // Fields accessed by inner classes
        int numActiveStreams;
        int numStreams;

        DefaultEndpoint(boolean server, int maxReservedStreams) {
            this.server = server;

            // Determine the starting stream ID for this endpoint. Client-initiated streams
            // are odd and server-initiated streams are even. Zero is reserved for the
            // connection. Stream 1 is reserved client-initiated stream for responding to an
            // upgrade from HTTP 1.1.
            if (server) {
                nextStreamIdToCreate = 2;
                nextReservationStreamId = 0;
            } else {
                nextStreamIdToCreate = 1;
                // For manually created client-side streams, 1 is reserved for HTTP upgrade, so start at 3.
                nextReservationStreamId = 1;
            }

            // Push is disallowed by default for servers and allowed for clients.
            pushToAllowed = !server;
            maxActiveStreams = MAX_VALUE;
            this.maxReservedStreams = checkPositiveOrZero(maxReservedStreams, "maxReservedStreams");
            updateMaxStreams();
        }

        @Override
        public int incrementAndGetNextStreamId() {
            return nextReservationStreamId >= 0 ? nextReservationStreamId += 2 : nextReservationStreamId;
        }

        private void incrementExpectedStreamId(int streamId) {
            if (streamId > nextReservationStreamId && nextReservationStreamId >= 0) {
                nextReservationStreamId = streamId;
            }
            nextStreamIdToCreate = streamId + 2;
            ++numStreams;
        }

        @Override
        public boolean isValidStreamId(int streamId) {
            return streamId > 0 && server == ((streamId & 1) == 0);
        }

        @Override
        public boolean mayHaveCreatedStream(int streamId) {
            return isValidStreamId(streamId) && streamId <= lastStreamCreated();
        }

        @Override
        public boolean canOpenStream() {
            return numActiveStreams < maxActiveStreams;
        }

        @Override
        public DefaultStream createStream(int streamId, boolean halfClosed) throws Http2Exception {
            State state = activeState(streamId, IDLE, isLocal(), halfClosed);

            checkNewStreamAllowed(streamId, state);

            // Create and initialize the stream.
            DefaultStream stream = new DefaultStream(streamId, state);

            incrementExpectedStreamId(streamId);

            addStream(stream);

            stream.activate();
            return stream;
        }

        @Override
        public boolean created(Http2Stream stream) {
            return stream instanceof DefaultStream && ((DefaultStream) stream).createdBy() == this;
        }

        @Override
        public boolean isServer() {
            return server;
        }

        @Override
        public DefaultStream reservePushStream(int streamId, Http2Stream parent) throws Http2Exception {
            if (parent == null) {
                throw connectionError(PROTOCOL_ERROR, "Parent stream missing");
            }
            if (isLocal() ? !parent.state().localSideOpen() : !parent.state().remoteSideOpen()) {
                throw connectionError(PROTOCOL_ERROR, "Stream %d is not open for sending push promise", parent.id());
            }
            if (!opposite().allowPushTo()) {
                throw connectionError(PROTOCOL_ERROR, "Server push not allowed to opposite endpoint");
            }
            State state = isLocal() ? RESERVED_LOCAL : RESERVED_REMOTE;
            checkNewStreamAllowed(streamId, state);

            // Create and initialize the stream.
            DefaultStream stream = new DefaultStream(streamId, state);

            incrementExpectedStreamId(streamId);

            // Register the stream.
            addStream(stream);
            return stream;
        }

        private void addStream(DefaultStream stream) {
            // Add the stream to the map and priority tree.
            streamMap.put(stream.id(), stream);

            // Notify the listeners of the event.
            for (int i = 0; i < listeners.size(); i++) {
                try {
                    listeners.get(i).onStreamAdded(stream);
                } catch (Throwable cause) {
                    logger.error("Caught Throwable from listener onStreamAdded.", cause);
                }
            }
        }

        @Override
        public void allowPushTo(boolean allow) {
            if (allow && server) {
                throw new IllegalArgumentException("Servers do not allow push");
            }
            pushToAllowed = allow;
        }

        @Override
        public boolean allowPushTo() {
            return pushToAllowed;
        }

        @Override
        public int numActiveStreams() {
            return numActiveStreams;
        }

        @Override
        public int maxActiveStreams() {
            return maxActiveStreams;
        }

        @Override
        public void maxActiveStreams(int maxActiveStreams) {
            this.maxActiveStreams = maxActiveStreams;
            updateMaxStreams();
        }

        @Override
        public int lastStreamCreated() {
            return nextStreamIdToCreate > 1 ? nextStreamIdToCreate - 2 : 0;
        }

        @Override
        public int lastStreamKnownByPeer() {
            return lastStreamKnownByPeer;
        }

        private void lastStreamKnownByPeer(int lastKnownStream) {
            lastStreamKnownByPeer = lastKnownStream;
        }

        @Override
        public F flowController() {
            return flowController;
        }

        @Override
        public void flowController(F flowController) {
            this.flowController = requireNonNull(flowController, "flowController");
        }

        @Override
        public Endpoint<? extends Http2FlowController> opposite() {
            return isLocal() ? remoteEndpoint : localEndpoint;
        }

        private void updateMaxStreams() {
            maxStreams = (int) Math.min(MAX_VALUE, (long) maxActiveStreams + maxReservedStreams);
        }

        private void checkNewStreamAllowed(int streamId, State state) throws Http2Exception {
            assert state != IDLE;
            if (lastStreamKnownByPeer >= 0 && streamId > lastStreamKnownByPeer) {
                throw streamError(streamId, REFUSED_STREAM,
                        "Cannot create stream %d greater than Last-Stream-ID %d from GOAWAY.",
                        streamId, lastStreamKnownByPeer);
            }
            if (!isValidStreamId(streamId)) {
                if (streamId < 0) {
                    throw new Http2NoMoreStreamIdsException();
                }
                throw connectionError(PROTOCOL_ERROR, "Request stream %d is not correct for %s connection", streamId,
                        server ? "server" : "client");
            }
            // This check must be after all id validated checks, but before the max streams check because it may be
            // recoverable to some degree for handling frames which can be sent on closed streams.
            if (streamId < nextStreamIdToCreate) {
                throw closedStreamError(PROTOCOL_ERROR, "Request stream %d is behind the next expected stream %d",
                        streamId, nextStreamIdToCreate);
            }
            if (nextStreamIdToCreate <= 0) {
                // We exhausted the stream id space that we  can use. Let's signal this back but also signal that
                // we still may want to process active streams.
                throw new Http2Exception(REFUSED_STREAM, "Stream IDs are exhausted for this endpoint.",
                        Http2Exception.ShutdownHint.GRACEFUL_SHUTDOWN);
            }
            boolean isReserved = state == RESERVED_LOCAL || state == RESERVED_REMOTE;
            if (!isReserved && !canOpenStream() || isReserved && numStreams >= maxStreams) {
                throw streamError(streamId, REFUSED_STREAM, "Maximum active streams violated for this endpoint: " +
                        (isReserved ? maxStreams : maxActiveStreams));
            }
            if (isClosed()) {
                throw connectionError(INTERNAL_ERROR, "Attempted to create stream id %d after connection was closed",
                                      streamId);
            }
        }

        private boolean isLocal() {
            return this == localEndpoint;
        }
    }

    /**
     * Allows events which would modify the collection of active streams to be queued while iterating via {@link
     * #forEachActiveStream(Http2StreamVisitor)}.
     */
    interface Event {
        /**
         * Trigger the original intention of this event. Expect to modify the active streams list.
         * <p/>
         * If a {@link RuntimeException} object is thrown it will be logged and <strong>not propagated</strong>.
         * Throwing from this method is not supported and is considered a programming error.
         */
        void process();
    }

    /**
     * Manages the list of currently active streams.  Queues any {@link Event}s that would modify the list of
     * active streams in order to prevent modification while iterating.
     */
    private final class ActiveStreams {
        private final List<Listener> listeners;
        private final Queue<Event> pendingEvents = new ArrayDeque<>(4);
        private final Set<Http2Stream> streams = new LinkedHashSet<>();
        private int pendingIterations;

        ActiveStreams(List<Listener> listeners) {
            this.listeners = listeners;
        }

        public int size() {
            return streams.size();
        }

        public void activate(final DefaultStream stream) {
            if (allowModifications()) {
                addToActiveStreams(stream);
            } else {
                pendingEvents.add(() -> addToActiveStreams(stream));
            }
        }

        public void deactivate(final DefaultStream stream, final Iterator<?> itr) {
            if (allowModifications() || itr != null) {
                removeFromActiveStreams(stream, itr);
            } else {
                pendingEvents.add(() -> removeFromActiveStreams(stream, itr));
            }
        }

        public Http2Stream forEachActiveStream(Http2StreamVisitor visitor) throws Http2Exception {
            incrementPendingIterations();
            try {
                for (Http2Stream stream : streams) {
                    if (!visitor.visit(stream)) {
                        return stream;
                    }
                }
                return null;
            } finally {
                decrementPendingIterations();
            }
        }

        void addToActiveStreams(DefaultStream stream) {
            if (streams.add(stream)) {
                // Update the number of active streams initiated by the endpoint.
                stream.createdBy().numActiveStreams++;

                for (int i = 0; i < listeners.size(); i++) {
                    try {
                        listeners.get(i).onStreamActive(stream);
                    } catch (Throwable cause) {
                        logger.error("Caught Throwable from listener onStreamActive.", cause);
                    }
                }
            }
        }

        void removeFromActiveStreams(DefaultStream stream, Iterator<?> itr) {
            if (streams.remove(stream)) {
                // Update the number of active streams initiated by the endpoint.
                stream.createdBy().numActiveStreams--;
                notifyClosed(stream);
            }
            removeStream(stream, itr);
        }

        boolean allowModifications() {
            return pendingIterations == 0;
        }

        void incrementPendingIterations() {
            ++pendingIterations;
        }

        void decrementPendingIterations() {
            --pendingIterations;
            if (allowModifications()) {
                for (;;) {
                    Event event = pendingEvents.poll();
                    if (event == null) {
                        break;
                    }
                    try {
                        event.process();
                    } catch (Throwable cause) {
                        logger.error("Caught Throwable while processing pending ActiveStreams$Event.", cause);
                    }
                }
            }
        }
    }

    /**
     * Implementation of {@link PropertyKey} that specifies the index position of the property.
     */
    final class DefaultPropertyKey implements PropertyKey {
        final int index;

        DefaultPropertyKey(int index) {
            this.index = index;
        }

        DefaultPropertyKey verifyConnection(Http2Connection connection) {
            if (connection != DefaultHttp2Connection.this) {
                throw new IllegalArgumentException("Using a key that was not created by this connection");
            }
            return this;
        }
    }

    /**
     * A registry of all stream property keys known by this connection.
     */
    private final class PropertyKeyRegistry {
        /**
         * Initial size of 4 because the default configuration currently has 3 listeners
         * (local/remote flow controller and {@link StreamByteDistributor}) and we leave room for 1 extra.
         * We could be more aggressive but the ArrayList resize will double the size if we are too small.
         */
        final List<DefaultPropertyKey> keys = new ArrayList<>(4);

        /**
         * Registers a new property key.
         */
        DefaultPropertyKey newKey() {
            DefaultPropertyKey key = new DefaultPropertyKey(keys.size());
            keys.add(key);
            return key;
        }

        int size() {
            return keys.size();
        }
    }
}