/*
 * ====================================================================
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF 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
 *
 *   http://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.
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of the Apache Software Foundation.  For more
 * information on the Apache Software Foundation, please see
 * <http://www.apache.org/>.
 *
 */
package org.apache.hc.core5.http2.impl.nio;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.charset.CharacterCodingException;
import java.util.Deque;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import javax.net.ssl.SSLSession;

import org.apache.hc.core5.concurrent.CancellableDependency;
import org.apache.hc.core5.http.ConnectionClosedException;
import org.apache.hc.core5.http.EndpointDetails;
import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.HttpConnection;
import org.apache.hc.core5.http.HttpException;
import org.apache.hc.core5.http.HttpStreamResetException;
import org.apache.hc.core5.http.HttpVersion;
import org.apache.hc.core5.http.ProtocolException;
import org.apache.hc.core5.http.ProtocolVersion;
import org.apache.hc.core5.http.config.CharCodingConfig;
import org.apache.hc.core5.http.impl.BasicEndpointDetails;
import org.apache.hc.core5.http.impl.BasicHttpConnectionMetrics;
import org.apache.hc.core5.http.impl.CharCodingSupport;
import org.apache.hc.core5.http.nio.AsyncPushConsumer;
import org.apache.hc.core5.http.nio.AsyncPushProducer;
import org.apache.hc.core5.http.nio.HandlerFactory;
import org.apache.hc.core5.http.nio.command.ExecutableCommand;
import org.apache.hc.core5.http.nio.command.ShutdownCommand;
import org.apache.hc.core5.http.protocol.HttpCoreContext;
import org.apache.hc.core5.http.protocol.HttpProcessor;
import org.apache.hc.core5.http2.H2ConnectionException;
import org.apache.hc.core5.http2.H2Error;
import org.apache.hc.core5.http2.H2StreamResetException;
import org.apache.hc.core5.http2.config.H2Config;
import org.apache.hc.core5.http2.config.H2Param;
import org.apache.hc.core5.http2.config.H2Setting;
import org.apache.hc.core5.http2.frame.FrameFactory;
import org.apache.hc.core5.http2.frame.FrameFlag;
import org.apache.hc.core5.http2.frame.FrameType;
import org.apache.hc.core5.http2.frame.RawFrame;
import org.apache.hc.core5.http2.frame.StreamIdGenerator;
import org.apache.hc.core5.http2.hpack.HPackDecoder;
import org.apache.hc.core5.http2.hpack.HPackEncoder;
import org.apache.hc.core5.http2.impl.BasicH2TransportMetrics;
import org.apache.hc.core5.http2.nio.AsyncPingHandler;
import org.apache.hc.core5.http2.nio.command.PingCommand;
import org.apache.hc.core5.io.CloseMode;
import org.apache.hc.core5.reactor.Command;
import org.apache.hc.core5.reactor.ProtocolIOSession;
import org.apache.hc.core5.reactor.ssl.TlsDetails;
import org.apache.hc.core5.util.Args;
import org.apache.hc.core5.util.ByteArrayBuffer;
import org.apache.hc.core5.util.Identifiable;
import org.apache.hc.core5.util.Timeout;

abstract class AbstractH2StreamMultiplexer implements Identifiable, HttpConnection {

    private static final long LINGER_TIME = 1000; // 1 second
    private static final long CONNECTION_WINDOW_LOW_MARK = 10 * 1024 * 1024; // 10 MiB

    enum ConnectionHandshake { READY, ACTIVE, GRACEFUL_SHUTDOWN, SHUTDOWN}
    enum SettingsHandshake { READY, TRANSMITTED, ACKED }

    private final ProtocolIOSession ioSession;
    private final FrameFactory frameFactory;
    private final StreamIdGenerator idGenerator;
    private final HttpProcessor httpProcessor;
    private final H2Config localConfig;
    private final BasicH2TransportMetrics inputMetrics;
    private final BasicH2TransportMetrics outputMetrics;
    private final BasicHttpConnectionMetrics connMetrics;
    private final FrameInputBuffer inputBuffer;
    private final FrameOutputBuffer outputBuffer;
    private final Deque<RawFrame> outputQueue;
    private final HPackEncoder hPackEncoder;
    private final HPackDecoder hPackDecoder;
    private final Map<Integer, H2Stream> streamMap;
    private final Queue<AsyncPingHandler> pingHandlers;
    private final AtomicInteger connInputWindow;
    private final AtomicInteger connOutputWindow;
    private final AtomicInteger outputRequests;
    private final AtomicInteger lastStreamId;
    private final H2StreamListener streamListener;

    private ConnectionHandshake connState = ConnectionHandshake.READY;
    private SettingsHandshake localSettingState = SettingsHandshake.READY;
    private SettingsHandshake remoteSettingState = SettingsHandshake.READY;

    private int initInputWinSize;
    private int initOutputWinSize;
    private int lowMark;

    private volatile H2Config remoteConfig;

    private Continuation continuation;

    private int processedRemoteStreamId;
    private EndpointDetails endpointDetails;

    AbstractH2StreamMultiplexer(
            final ProtocolIOSession ioSession,
            final FrameFactory frameFactory,
            final StreamIdGenerator idGenerator,
            final HttpProcessor httpProcessor,
            final CharCodingConfig charCodingConfig,
            final H2Config h2Config,
            final H2StreamListener streamListener) {
        this.ioSession = Args.notNull(ioSession, "IO session");
        this.frameFactory = Args.notNull(frameFactory, "Frame factory");
        this.idGenerator = Args.notNull(idGenerator, "Stream id generator");
        this.httpProcessor = Args.notNull(httpProcessor, "HTTP processor");
        this.localConfig = h2Config != null ? h2Config : H2Config.DEFAULT;
        this.inputMetrics = new BasicH2TransportMetrics();
        this.outputMetrics = new BasicH2TransportMetrics();
        this.connMetrics = new BasicHttpConnectionMetrics(this.inputMetrics, this.outputMetrics);
        this.inputBuffer = new FrameInputBuffer(this.inputMetrics, this.localConfig.getMaxFrameSize());
        this.outputBuffer = new FrameOutputBuffer(this.outputMetrics, this.localConfig.getMaxFrameSize());
        this.outputQueue = new ConcurrentLinkedDeque<>();
        this.pingHandlers = new ConcurrentLinkedQueue<>();
        this.outputRequests = new AtomicInteger(0);
        this.lastStreamId = new AtomicInteger(0);
        this.hPackEncoder = new HPackEncoder(CharCodingSupport.createEncoder(charCodingConfig));
        this.hPackDecoder = new HPackDecoder(CharCodingSupport.createDecoder(charCodingConfig));
        this.streamMap = new ConcurrentHashMap<>();
        this.remoteConfig = H2Config.INIT;
        this.connInputWindow = new AtomicInteger(H2Config.INIT.getInitialWindowSize());
        this.connOutputWindow = new AtomicInteger(H2Config.INIT.getInitialWindowSize());

        this.initInputWinSize = H2Config.INIT.getInitialWindowSize();
        this.initOutputWinSize = H2Config.INIT.getInitialWindowSize();

        this.hPackDecoder.setMaxTableSize(H2Config.INIT.getHeaderTableSize());
        this.hPackEncoder.setMaxTableSize(H2Config.INIT.getHeaderTableSize());
        this.hPackDecoder.setMaxListSize(H2Config.INIT.getMaxHeaderListSize());

        this.lowMark = H2Config.INIT.getInitialWindowSize() / 2;
        this.streamListener = streamListener;
    }

    @Override
    public String getId() {
        return ioSession.getId();
    }

    abstract void acceptHeaderFrame() throws H2ConnectionException;

    abstract void acceptPushRequest() throws H2ConnectionException;

    abstract void acceptPushFrame() throws H2ConnectionException;

    abstract H2StreamHandler createRemotelyInitiatedStream(
            H2StreamChannel channel,
            HttpProcessor httpProcessor,
            BasicHttpConnectionMetrics connMetrics,
            HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException;

    abstract H2StreamHandler createLocallyInitiatedStream(
            ExecutableCommand command,
            H2StreamChannel channel,
            HttpProcessor httpProcessor,
            BasicHttpConnectionMetrics connMetrics) throws IOException;

    private int updateWindow(final AtomicInteger window, final int delta) throws ArithmeticException {
        for (;;) {
            final int current = window.get();
            long newValue = (long) current + delta;

            //TODO: work-around for what looks like a bug in Ngnix (1.11)
            // Tolerate if the update window exceeded by one
            if (newValue == 0x80000000L) {
                newValue = Integer.MAX_VALUE;
            }
            //TODO: needs to be removed

            if (Math.abs(newValue) > 0x7fffffffL) {
                throw new ArithmeticException("Update causes flow control window to exceed " + Integer.MAX_VALUE);
            }
            if (window.compareAndSet(current, (int) newValue)) {
                return (int) newValue;
            }
        }
    }

    private int updateInputWindow(
            final int streamId, final AtomicInteger window, final int delta) throws ArithmeticException {
        final int newSize = updateWindow(window, delta);
        if (streamListener != null) {
            streamListener.onInputFlowControl(this, streamId, delta, newSize);
        }
        return newSize;
    }

    private int updateOutputWindow(
            final int streamId, final AtomicInteger window, final int delta) throws ArithmeticException {
        final int newSize = updateWindow(window, delta);
        if (streamListener != null) {
            streamListener.onOutputFlowControl(this, streamId, delta, newSize);
        }
        return newSize;
    }

    private void commitFrameInternal(final RawFrame frame) throws IOException {
        if (outputBuffer.isEmpty() && outputQueue.isEmpty()) {
            if (streamListener != null) {
                streamListener.onFrameOutput(this, frame.getStreamId(), frame);
            }
            outputBuffer.write(frame, ioSession);
        } else {
            outputQueue.addLast(frame);
        }
        ioSession.setEvent(SelectionKey.OP_WRITE);
    }

    private void commitFrame(final RawFrame frame) throws IOException {
        Args.notNull(frame, "Frame");
        ioSession.getLock().lock();
        try {
            commitFrameInternal(frame);
        } finally {
            ioSession.getLock().unlock();
        }
    }

    private void commitHeaders(
            final int streamId, final List<? extends Header> headers, final boolean endStream) throws IOException {
        if (streamListener != null) {
            streamListener.onHeaderOutput(this, streamId, headers);
        }
        final ByteArrayBuffer buf = new ByteArrayBuffer(512);
        hPackEncoder.encodeHeaders(buf, headers, localConfig.isCompressionEnabled());

        int off = 0;
        int remaining = buf.length();
        boolean continuation = false;

        while (remaining > 0) {
            final int chunk = Math.min(remoteConfig.getMaxFrameSize(), remaining);
            final ByteBuffer payload = ByteBuffer.wrap(buf.array(), off, chunk);

            remaining -= chunk;
            off += chunk;

            final boolean endHeaders = remaining == 0;
            final RawFrame frame;
            if (!continuation) {
                frame = frameFactory.createHeaders(streamId, payload, endHeaders, endStream);
                continuation = true;
            } else {
                frame = frameFactory.createContinuation(streamId, payload, endHeaders);
            }
            commitFrameInternal(frame);
        }
    }

    private void commitPushPromise(
            final int streamId, final int promisedStreamId, final List<Header> headers) throws IOException {
        if (headers == null || headers.isEmpty()) {
            throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Message headers are missing");
        }
        if (streamListener != null) {
            streamListener.onHeaderOutput(this, streamId, headers);
        }
        final ByteArrayBuffer buf = new ByteArrayBuffer(512);
        buf.append((byte)(promisedStreamId >> 24));
        buf.append((byte)(promisedStreamId >> 16));
        buf.append((byte)(promisedStreamId >> 8));
        buf.append((byte)(promisedStreamId));

        hPackEncoder.encodeHeaders(buf, headers, localConfig.isCompressionEnabled());

        int off = 0;
        int remaining = buf.length();
        boolean continuation = false;

        while (remaining > 0) {
            final int chunk = Math.min(remoteConfig.getMaxFrameSize(), remaining);
            final ByteBuffer payload = ByteBuffer.wrap(buf.array(), off, chunk);

            remaining -= chunk;
            off += chunk;

            final boolean endHeaders = remaining == 0;
            final RawFrame frame;
            if (!continuation) {
                frame = frameFactory.createPushPromise(streamId, payload, endHeaders);
                continuation = true;
            } else {
                frame = frameFactory.createContinuation(streamId, payload, endHeaders);
            }
            commitFrameInternal(frame);
        }
    }

    private void streamDataFrame(
            final int streamId,
            final AtomicInteger streamOutputWindow,
            final ByteBuffer payload,
            final int chunk) throws IOException {
        final RawFrame dataFrame = frameFactory.createData(streamId, payload, false);
        if (streamListener != null) {
            streamListener.onFrameOutput(this, streamId, dataFrame);
        }
        updateOutputWindow(0, connOutputWindow, -chunk);
        updateOutputWindow(streamId, streamOutputWindow, -chunk);
        outputBuffer.write(dataFrame, ioSession);
    }

    private int streamData(
            final int streamId, final AtomicInteger streamOutputWindow, final ByteBuffer payload) throws IOException {
        if (outputBuffer.isEmpty() && outputQueue.isEmpty()) {
            final int capacity = Math.min(connOutputWindow.get(), streamOutputWindow.get());
            if (capacity <= 0) {
                return 0;
            }
            final int maxPayloadSize = Math.min(capacity, remoteConfig.getMaxFrameSize());
            final int chunk;
            if (payload.remaining() <= maxPayloadSize) {
                chunk = payload.remaining();
                streamDataFrame(streamId, streamOutputWindow, payload, chunk);
            } else {
                chunk = maxPayloadSize;
                final int originalLimit = payload.limit();
                try {
                    payload.limit(payload.position() + chunk);
                    streamDataFrame(streamId, streamOutputWindow, payload, chunk);
                } finally {
                    payload.limit(originalLimit);
                }
            }
            payload.position(payload.position() + chunk);
            ioSession.setEvent(SelectionKey.OP_WRITE);
            return chunk;
        }
        return 0;
    }

    private void incrementInputCapacity(
            final int streamId, final AtomicInteger inputWindow, final int inputCapacity) throws IOException {
        if (inputCapacity > 0) {
            final int streamWinSize = inputWindow.get();
            final int remainingCapacity = Integer.MAX_VALUE - streamWinSize;
            final int chunk = Math.min(inputCapacity, remainingCapacity);
            if (chunk != 0) {
                final RawFrame windowUpdateFrame = frameFactory.createWindowUpdate(streamId, chunk);
                commitFrame(windowUpdateFrame);
                updateInputWindow(streamId, inputWindow, chunk);
            }
        }
    }

    private void requestSessionOutput() {
        outputRequests.incrementAndGet();
        ioSession.setEvent(SelectionKey.OP_WRITE);
    }

    private void updateLastStreamId(final int streamId) {
        final int currentId = lastStreamId.get();
        if (streamId > currentId) {
            lastStreamId.compareAndSet(currentId, streamId);
        }
    }

    private int generateStreamId() {
        for (;;) {
            final int currentId = lastStreamId.get();
            final int newStreamId = idGenerator.generate(currentId);
            if (lastStreamId.compareAndSet(currentId, newStreamId)) {
                return newStreamId;
            }
        }
    }

    public final void onConnect() throws HttpException, IOException {
        connState = ConnectionHandshake.ACTIVE;
        final RawFrame settingsFrame = frameFactory.createSettings(
                new H2Setting(H2Param.HEADER_TABLE_SIZE, localConfig.getHeaderTableSize()),
                new H2Setting(H2Param.ENABLE_PUSH, localConfig.isPushEnabled() ? 1 : 0),
                new H2Setting(H2Param.MAX_CONCURRENT_STREAMS, localConfig.getMaxConcurrentStreams()),
                new H2Setting(H2Param.INITIAL_WINDOW_SIZE, localConfig.getInitialWindowSize()),
                new H2Setting(H2Param.MAX_FRAME_SIZE, localConfig.getMaxFrameSize()),
                new H2Setting(H2Param.MAX_HEADER_LIST_SIZE, localConfig.getMaxHeaderListSize()));

        commitFrame(settingsFrame);
        localSettingState = SettingsHandshake.TRANSMITTED;
        maximizeConnWindow(connInputWindow.get());

        if (streamListener != null) {
            final int initInputWindow = connInputWindow.get();
            streamListener.onInputFlowControl(this, 0, initInputWindow, initInputWindow);
            final int initOutputWindow = connOutputWindow.get();
            streamListener.onOutputFlowControl(this, 0, initOutputWindow, initOutputWindow);
        }
    }

    public final void onInput(final ByteBuffer src) throws HttpException, IOException {
        if (connState == ConnectionHandshake.SHUTDOWN) {
            ioSession.clearEvent(SelectionKey.OP_READ);
        } else {
            if (src != null) {
                inputBuffer.put(src);
            }
            RawFrame frame;
            while ((frame = inputBuffer.read(ioSession)) != null) {
                if (streamListener != null) {
                    streamListener.onFrameInput(this, frame.getStreamId(), frame);
                }
                consumeFrame(frame);
            }
        }
    }

    public final void onOutput() throws HttpException, IOException {
        ioSession.getLock().lock();
        try {
            if (!outputBuffer.isEmpty()) {
                outputBuffer.flush(ioSession);
            }
            while (outputBuffer.isEmpty()) {
                final RawFrame frame = outputQueue.poll();
                if (frame != null) {
                    if (streamListener != null) {
                        streamListener.onFrameOutput(this, frame.getStreamId(), frame);
                    }
                    outputBuffer.write(frame, ioSession);
                } else {
                    break;
                }
            }
        } finally {
            ioSession.getLock().unlock();
        }

        if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {

            if (connOutputWindow.get() > 0 && remoteSettingState == SettingsHandshake.ACKED) {
                produceOutput();
            }
            final int pendingOutputRequests = outputRequests.get();
            boolean outputPending = false;
            if (!streamMap.isEmpty() && connOutputWindow.get() > 0) {
                for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                    final Map.Entry<Integer, H2Stream> entry = it.next();
                    final H2Stream stream = entry.getValue();
                    if (!stream.isLocalClosed()
                            && stream.getOutputWindow().get() > 0
                            && stream.isOutputReady()) {
                        outputPending = true;
                        break;
                    }
                }
            }
            ioSession.getLock().lock();
            try {
                if (!outputPending && outputBuffer.isEmpty() && outputQueue.isEmpty()
                        && outputRequests.compareAndSet(pendingOutputRequests, 0)) {
                    ioSession.clearEvent(SelectionKey.OP_WRITE);
                } else {
                    outputRequests.addAndGet(-pendingOutputRequests);
                }
            } finally {
                ioSession.getLock().unlock();
            }
        }

        if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0 && remoteSettingState == SettingsHandshake.ACKED) {
            processPendingCommands();
        }
        if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) == 0) {
            for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                final Map.Entry<Integer, H2Stream> entry = it.next();
                final H2Stream stream = entry.getValue();
                if (stream.isLocalClosed() && stream.isRemoteClosed()) {
                    stream.releaseResources();
                    it.remove();
                }
            }
            if (streamMap.isEmpty()) {
                connState = ConnectionHandshake.SHUTDOWN;
            }
        }
        if (connState.compareTo(ConnectionHandshake.SHUTDOWN) >= 0) {
            ioSession.getLock().lock();
            try {
                if (outputBuffer.isEmpty() && outputQueue.isEmpty()) {
                    ioSession.close();
                }
            } finally {
                ioSession.getLock().unlock();
            }
        }
    }

    public final void onTimeout(final Timeout timeout) throws HttpException, IOException {
        connState = ConnectionHandshake.SHUTDOWN;

        final RawFrame goAway;
        if (localSettingState != SettingsHandshake.ACKED) {
            goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.SETTINGS_TIMEOUT,
                            "Setting timeout (" + timeout + ")");
        } else {
            goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.NO_ERROR,
                            "Timeout due to inactivity (" + timeout + ")");
        }
        commitFrame(goAway);
        for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
            final Map.Entry<Integer, H2Stream> entry = it.next();
            final H2Stream stream = entry.getValue();
            stream.reset(new H2StreamResetException(H2Error.NO_ERROR, "Timeout due to inactivity (" + timeout + ")"));
        }
        streamMap.clear();
    }

    public final void onDisconnect() {
        for (;;) {
            final AsyncPingHandler pingHandler = pingHandlers.poll();
            if (pingHandler != null) {
                pingHandler.cancel();
            } else {
                break;
            }
        }
        for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
            final Map.Entry<Integer, H2Stream> entry = it.next();
            final H2Stream stream = entry.getValue();
            stream.cancel();
        }
        for (;;) {
            final Command command = ioSession.poll();
            if (command != null) {
                if (command instanceof ExecutableCommand) {
                    ((ExecutableCommand) command).failed(new ConnectionClosedException());
                } else {
                    command.cancel();
                }
            } else {
                break;
            }
        }
    }

    private void processPendingCommands() throws IOException, HttpException {
        while (streamMap.size() < remoteConfig.getMaxConcurrentStreams()) {
            final Command command = ioSession.poll();
            if (command == null) {
                break;
            }
            if (command instanceof ShutdownCommand) {
                final ShutdownCommand shutdownCommand = (ShutdownCommand) command;
                if (shutdownCommand.getType() == CloseMode.IMMEDIATE) {
                    for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                        final Map.Entry<Integer, H2Stream> entry = it.next();
                        final H2Stream stream = entry.getValue();
                        stream.cancel();
                    }
                    streamMap.clear();
                    connState = ConnectionHandshake.SHUTDOWN;
                } else {
                    if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
                        final RawFrame goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.NO_ERROR, "Graceful shutdown");
                        commitFrame(goAway);
                        connState = streamMap.isEmpty() ? ConnectionHandshake.SHUTDOWN : ConnectionHandshake.GRACEFUL_SHUTDOWN;
                    }
                }
                break;
            } else if (command instanceof PingCommand) {
                final PingCommand pingCommand = (PingCommand) command;
                final AsyncPingHandler handler = pingCommand.getHandler();
                pingHandlers.add(handler);
                final RawFrame ping = frameFactory.createPing(handler.getData());
                commitFrame(ping);
            } else if (command instanceof ExecutableCommand) {
                final int streamId = generateStreamId();
                final H2StreamChannelImpl channel = new H2StreamChannelImpl(
                        streamId, true, initInputWinSize, initOutputWinSize);
                final ExecutableCommand executableCommand = (ExecutableCommand) command;
                final H2StreamHandler streamHandler = createLocallyInitiatedStream(
                        executableCommand, channel, httpProcessor, connMetrics);

                final H2Stream stream = new H2Stream(channel, streamHandler, false);
                streamMap.put(streamId, stream);

                if (streamListener != null) {
                    final int initInputWindow = stream.getInputWindow().get();
                    streamListener.onInputFlowControl(this, streamId, initInputWindow, initInputWindow);
                    final int initOutputWindow = stream.getOutputWindow().get();
                    streamListener.onOutputFlowControl(this, streamId, initOutputWindow, initOutputWindow);
                }

                if (stream.isOutputReady()) {
                    stream.produceOutput();
                }
                final CancellableDependency cancellableDependency = executableCommand.getCancellableDependency();
                if (cancellableDependency != null) {
                    cancellableDependency.setDependency(stream::abort);
                }
                if (!outputQueue.isEmpty()) {
                    return;
                }
            }
        }
    }

    public final void onException(final Exception cause) {
        try {
            for (;;) {
                final AsyncPingHandler pingHandler = pingHandlers.poll();
                if (pingHandler != null) {
                    pingHandler.failed(cause);
                } else {
                    break;
                }
            }
            for (;;) {
                final Command command = ioSession.poll();
                if (command != null) {
                    if (command instanceof ExecutableCommand) {
                        ((ExecutableCommand) command).failed(new ConnectionClosedException());
                    } else {
                        command.cancel();
                    }
                } else {
                    break;
                }
            }
            for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                final Map.Entry<Integer, H2Stream> entry = it.next();
                final H2Stream stream = entry.getValue();
                stream.reset(cause);
            }
            streamMap.clear();
            if (!(cause instanceof ConnectionClosedException)) {
                if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) <= 0) {
                    final H2Error errorCode;
                    if (cause instanceof H2ConnectionException) {
                        errorCode = H2Error.getByCode(((H2ConnectionException) cause).getCode());
                    } else if (cause instanceof ProtocolException){
                        errorCode = H2Error.PROTOCOL_ERROR;
                    } else {
                        errorCode = H2Error.INTERNAL_ERROR;
                    }
                    final RawFrame goAway = frameFactory.createGoAway(processedRemoteStreamId, errorCode, cause.getMessage());
                    commitFrame(goAway);
                }
            }
        } catch (final IOException ignore) {
        } finally {
            connState = ConnectionHandshake.SHUTDOWN;
            final CloseMode closeMode;
            if (cause instanceof ConnectionClosedException) {
                closeMode = CloseMode.GRACEFUL;
            } else if (cause instanceof IOException) {
                closeMode = CloseMode.IMMEDIATE;
            } else {
                closeMode = CloseMode.GRACEFUL;
            }
            ioSession.close(closeMode);
        }
    }

    private H2Stream getValidStream(final int streamId) throws H2ConnectionException {
        if (streamId == 0) {
            throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id: " + streamId);
        }
        final H2Stream stream = streamMap.get(streamId);
        if (stream == null) {
            if (streamId <= lastStreamId.get()) {
                throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
            } else {
                throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected stream id: " + streamId);
            }
        }
        return stream;
    }

    private void consumeFrame(final RawFrame frame) throws HttpException, IOException {
        final FrameType frameType = FrameType.valueOf(frame.getType());
        final int streamId = frame.getStreamId();
        if (continuation != null && frameType != FrameType.CONTINUATION) {
            throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "CONTINUATION frame expected");
        }
        if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) >= 0) {
            if (streamId > processedRemoteStreamId && !idGenerator.isSameSide(streamId)) {
                // ignore the frame
                return;
            }
        }
        switch (frameType) {
            case DATA: {
                final H2Stream stream = getValidStream(streamId);
                try {
                    consumeDataFrame(frame, stream);
                } catch (final H2StreamResetException ex) {
                    stream.localReset(ex);
                } catch (final HttpStreamResetException ex) {
                    stream.localReset(ex, ex.getCause() != null ? H2Error.INTERNAL_ERROR : H2Error.CANCEL);
                }

                if (stream.isTerminated()) {
                    streamMap.remove(streamId);
                    stream.releaseResources();
                }
            }
            break;
            case HEADERS: {
                if (streamId == 0) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id: " + streamId);
                }
                H2Stream stream = streamMap.get(streamId);
                if (stream == null) {
                    acceptHeaderFrame();

                    updateLastStreamId(streamId);

                    final H2StreamChannelImpl channel = new H2StreamChannelImpl(
                            streamId, false, initInputWinSize, initOutputWinSize);
                    final H2StreamHandler streamHandler = createRemotelyInitiatedStream(
                            channel, httpProcessor, connMetrics, null);
                    stream = new H2Stream(channel, streamHandler, true);
                    if (stream.isOutputReady()) {
                        stream.produceOutput();
                    }
                    streamMap.put(streamId, stream);
                }

                try {
                    consumeHeaderFrame(frame, stream);

                    if (stream.isOutputReady()) {
                        stream.produceOutput();
                    }
                } catch (final H2StreamResetException ex) {
                    stream.localReset(ex);
                } catch (final HttpStreamResetException ex) {
                    stream.localReset(ex, ex.getCause() != null ? H2Error.INTERNAL_ERROR : H2Error.CANCEL);
                } catch (final HttpException ex) {
                    stream.handle(ex);
                }

                if (stream.isTerminated()) {
                    streamMap.remove(streamId);
                    stream.releaseResources();
                }
            }
            break;
            case CONTINUATION: {
                if (continuation == null) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected CONTINUATION frame");
                }
                if (streamId != continuation.streamId) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected CONTINUATION stream id: " + streamId);
                }

                final H2Stream stream = getValidStream(streamId);
                try {

                    consumeContinuationFrame(frame, stream);
                } catch (final H2StreamResetException ex) {
                    stream.localReset(ex);
                } catch (final HttpStreamResetException ex) {
                    stream.localReset(ex, ex.getCause() != null ? H2Error.INTERNAL_ERROR : H2Error.CANCEL);
                }

                if (stream.isTerminated()) {
                    streamMap.remove(streamId);
                    stream.releaseResources();
                }
            }
            break;
            case WINDOW_UPDATE: {
                final ByteBuffer payload = frame.getPayload();
                if (payload == null || payload.remaining() != 4) {
                    throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid WINDOW_UPDATE frame payload");
                }
                final int delta = payload.getInt();
                if (delta <= 0) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Invalid WINDOW_UPDATE delta");
                }
                if (streamId == 0) {
                    try {
                        updateOutputWindow(0, connOutputWindow, delta);
                    } catch (final ArithmeticException ex) {
                        throw new H2ConnectionException(H2Error.FLOW_CONTROL_ERROR, ex.getMessage());
                    }
                } else {
                    final H2Stream stream = streamMap.get(streamId);
                    if (stream != null) {
                        try {
                            updateOutputWindow(streamId, stream.getOutputWindow(), delta);
                        } catch (final ArithmeticException ex) {
                            throw new H2ConnectionException(H2Error.FLOW_CONTROL_ERROR, ex.getMessage());
                        }
                    }
                }
                ioSession.setEvent(SelectionKey.OP_WRITE);
            }
            break;
            case RST_STREAM: {
                if (streamId == 0) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id: " + streamId);
                }
                final H2Stream stream = streamMap.get(streamId);
                if (stream == null) {
                    if (streamId > lastStreamId.get()) {
                        throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected stream id: " + streamId);
                    }
                } else {
                    final ByteBuffer payload = frame.getPayload();
                    if (payload == null || payload.remaining() != 4) {
                        throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid RST_STREAM frame payload");
                    }
                    final int errorCode = payload.getInt();
                    stream.reset(new H2StreamResetException(errorCode, "Stream reset (" + errorCode + ")"));
                    streamMap.remove(streamId);
                    stream.releaseResources();
                }
            }
            break;
            case PING: {
                if (streamId != 0) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
                }
                final ByteBuffer ping = frame.getPayloadContent();
                if (ping == null || ping.remaining() != 8) {
                    throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid PING frame payload");
                }
                if (frame.isFlagSet(FrameFlag.ACK)) {
                    final AsyncPingHandler pingHandler = pingHandlers.poll();
                    if (pingHandler != null) {
                        pingHandler.consumeResponse(ping);
                    }
                } else {
                    final ByteBuffer pong = ByteBuffer.allocate(ping.remaining());
                    pong.put(ping);
                    pong.flip();
                    final RawFrame response = frameFactory.createPingAck(pong);
                    commitFrame(response);
                }
            }
            break;
            case SETTINGS: {
                if (streamId != 0) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
                }
                if (frame.isFlagSet(FrameFlag.ACK)) {
                    if (localSettingState == SettingsHandshake.TRANSMITTED) {
                        localSettingState = SettingsHandshake.ACKED;
                        ioSession.setEvent(SelectionKey.OP_WRITE);
                        applyLocalSettings();
                    }
                } else {
                    final ByteBuffer payload = frame.getPayload();
                    if (payload != null) {
                        if ((payload.remaining() % 6) != 0) {
                            throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid SETTINGS payload");
                        }
                        consumeSettingsFrame(payload);
                        remoteSettingState = SettingsHandshake.TRANSMITTED;
                    }
                    // Send ACK
                    final RawFrame response = frameFactory.createSettingsAck();
                    commitFrame(response);
                    remoteSettingState = SettingsHandshake.ACKED;
                }
            }
            break;
            case PRIORITY:
                // Stream priority not supported
                break;
            case PUSH_PROMISE: {
                acceptPushFrame();

                if (!localConfig.isPushEnabled()) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Push is disabled");
                }

                final H2Stream stream = getValidStream(streamId);
                if (stream.isRemoteClosed()) {
                    stream.localReset(new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream closed"));
                    break;
                }

                final ByteBuffer payload = frame.getPayloadContent();
                if (payload == null || payload.remaining() < 4) {
                    throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid PUSH_PROMISE payload");
                }
                final int promisedStreamId = payload.getInt();
                if (promisedStreamId == 0 || idGenerator.isSameSide(promisedStreamId)) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal promised stream id: " + promisedStreamId);
                }
                if (streamMap.get(promisedStreamId) != null) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected promised stream id: " + promisedStreamId);
                }

                updateLastStreamId(promisedStreamId);

                final H2StreamChannelImpl channel = new H2StreamChannelImpl(
                        promisedStreamId, false, initInputWinSize, initOutputWinSize);
                final H2StreamHandler streamHandler = createRemotelyInitiatedStream(
                        channel, httpProcessor, connMetrics, stream.getPushHandlerFactory());
                final H2Stream promisedStream = new H2Stream(channel, streamHandler, true);
                streamMap.put(promisedStreamId, promisedStream);

                try {
                    consumePushPromiseFrame(frame, payload, promisedStream);
                } catch (final H2StreamResetException ex) {
                    promisedStream.localReset(ex);
                } catch (final HttpStreamResetException ex) {
                    promisedStream.localReset(ex, ex.getCause() != null ? H2Error.INTERNAL_ERROR : H2Error.NO_ERROR);
                }
            }
            break;
            case GOAWAY: {
                if (streamId != 0) {
                    throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
                }
                final ByteBuffer payload = frame.getPayload();
                if (payload == null || payload.remaining() < 8) {
                    throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid GOAWAY payload");
                }
                final int processedLocalStreamId = payload.getInt();
                final int errorCode = payload.getInt();
                if (errorCode == H2Error.NO_ERROR.getCode()) {
                    if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
                        for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                            final Map.Entry<Integer, H2Stream> entry = it.next();
                            final int activeStreamId = entry.getKey();
                            if (!idGenerator.isSameSide(activeStreamId) && activeStreamId > processedLocalStreamId) {
                                final H2Stream stream = entry.getValue();
                                stream.cancel();
                                it.remove();
                            }
                        }
                    }
                    connState = streamMap.isEmpty() ? ConnectionHandshake.SHUTDOWN : ConnectionHandshake.GRACEFUL_SHUTDOWN;
                } else {
                    for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                        final Map.Entry<Integer, H2Stream> entry = it.next();
                        final H2Stream stream = entry.getValue();
                        stream.reset(new H2StreamResetException(errorCode, "Connection terminated by the peer (" + errorCode + ")"));
                    }
                    streamMap.clear();
                    connState = ConnectionHandshake.SHUTDOWN;
                }
            }
            ioSession.setEvent(SelectionKey.OP_WRITE);
            break;
        }
    }

    private void consumeDataFrame(final RawFrame frame, final H2Stream stream) throws HttpException, IOException {
        final int streamId = stream.getId();
        final ByteBuffer payload = frame.getPayloadContent();
        if (payload != null) {
            final int frameLength = frame.getLength();
            final int streamWinSize = updateInputWindow(streamId, stream.getInputWindow(), -frameLength);
            if (streamWinSize < lowMark && !stream.isRemoteClosed()) {
                stream.produceInputCapacityUpdate();
            }
            final int connWinSize = updateInputWindow(0, connInputWindow, -frameLength);
            if (connWinSize < CONNECTION_WINDOW_LOW_MARK) {
                maximizeConnWindow(connWinSize);
            }
        }
        if (stream.isRemoteClosed()) {
            throw new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream already closed");
        }
        if (frame.isFlagSet(FrameFlag.END_STREAM)) {
            stream.setRemoteEndStream();
        }
        if (stream.isLocalReset()) {
            return;
        }
        stream.consumeData(payload);
    }

    private void maximizeConnWindow(final int connWinSize) throws IOException {
        final int delta = Integer.MAX_VALUE - connWinSize;
        if (delta > 0) {
            final RawFrame windowUpdateFrame = frameFactory.createWindowUpdate(0, delta);
            commitFrame(windowUpdateFrame);
            updateInputWindow(0, connInputWindow, delta);
        }
    }

    private void consumePushPromiseFrame(final RawFrame frame, final ByteBuffer payload, final H2Stream promisedStream) throws HttpException, IOException {
        final int promisedStreamId = promisedStream.getId();
        if (!frame.isFlagSet(FrameFlag.END_HEADERS)) {
            continuation = new Continuation(promisedStreamId, frame.getType(), true);
        }
        if (continuation == null) {
            final List<Header> headers = hPackDecoder.decodeHeaders(payload);
            if (promisedStreamId > processedRemoteStreamId) {
                processedRemoteStreamId = promisedStreamId;
            }
            if (streamListener != null) {
                streamListener.onHeaderInput(this, promisedStreamId, headers);
            }
            if (connState == ConnectionHandshake.GRACEFUL_SHUTDOWN) {
                throw new H2StreamResetException(H2Error.REFUSED_STREAM, "Stream refused");
            }
            promisedStream.consumePromise(headers);
        } else {
            continuation.copyPayload(payload);
        }
    }

    List<Header> decodeHeaders(final ByteBuffer payload) throws HttpException {
        return hPackDecoder.decodeHeaders(payload);
    }

    private void consumeHeaderFrame(final RawFrame frame, final H2Stream stream) throws HttpException, IOException {
        final int streamId = stream.getId();
        if (!frame.isFlagSet(FrameFlag.END_HEADERS)) {
            continuation = new Continuation(streamId, frame.getType(), frame.isFlagSet(FrameFlag.END_STREAM));
        }
        final ByteBuffer payload = frame.getPayloadContent();
        if (frame.isFlagSet(FrameFlag.PRIORITY)) {
            // Priority not supported
            payload.getInt();
            payload.get();
        }
        if (continuation == null) {
            final List<Header> headers = decodeHeaders(payload);
            if (stream.isRemoteInitiated() && streamId > processedRemoteStreamId) {
                processedRemoteStreamId = streamId;
            }
            if (streamListener != null) {
                streamListener.onHeaderInput(this, streamId, headers);
            }
            if (stream.isRemoteClosed()) {
                throw new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream already closed");
            }
            if (stream.isLocalReset()) {
                return;
            }
            if (connState == ConnectionHandshake.GRACEFUL_SHUTDOWN) {
                throw new H2StreamResetException(H2Error.PROTOCOL_ERROR, "Stream refused");
            }
            if (frame.isFlagSet(FrameFlag.END_STREAM)) {
                stream.setRemoteEndStream();
            }
            stream.consumeHeader(headers);
        } else {
            continuation.copyPayload(payload);
        }
    }

    private void consumeContinuationFrame(final RawFrame frame, final H2Stream stream) throws HttpException, IOException {
        final int streamId = frame.getStreamId();
        final ByteBuffer payload = frame.getPayload();
        continuation.copyPayload(payload);
        if (frame.isFlagSet(FrameFlag.END_HEADERS)) {
            final List<Header> headers = decodeHeaders(continuation.getContent());
            if (stream.isRemoteInitiated() && streamId > processedRemoteStreamId) {
                processedRemoteStreamId = streamId;
            }
            if (streamListener != null) {
                streamListener.onHeaderInput(this, streamId, headers);
            }
            if (connState == ConnectionHandshake.GRACEFUL_SHUTDOWN) {
                throw new H2StreamResetException(H2Error.PROTOCOL_ERROR, "Stream refused");
            }
            if (stream.isRemoteClosed()) {
                throw new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream already closed");
            }
            if (stream.isLocalReset()) {
                return;
            }
            if (continuation.endStream) {
                stream.setRemoteEndStream();
            }
            if (continuation.type == FrameType.PUSH_PROMISE.getValue()) {
                stream.consumePromise(headers);
            } else {
                stream.consumeHeader(headers);
            }
            continuation = null;
        }
    }

    private void consumeSettingsFrame(final ByteBuffer payload) throws HttpException, IOException {
        final H2Config.Builder configBuilder = H2Config.initial();
        while (payload.hasRemaining()) {
            final int code = payload.getShort();
            final H2Param param = H2Param.valueOf(code);
            if (param != null) {
                final int value = payload.getInt();
                switch (param) {
                    case HEADER_TABLE_SIZE:
                        try {
                            configBuilder.setHeaderTableSize(value);
                        } catch (final IllegalArgumentException ex) {
                            throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, ex.getMessage());
                        }
                        break;
                    case MAX_CONCURRENT_STREAMS:
                        try {
                            configBuilder.setMaxConcurrentStreams(value);
                        } catch (final IllegalArgumentException ex) {
                            throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, ex.getMessage());
                        }
                        break;
                    case ENABLE_PUSH:
                        configBuilder.setPushEnabled(value == 1);
                        break;
                    case INITIAL_WINDOW_SIZE:
                        try {
                            configBuilder.setInitialWindowSize(value);
                        } catch (final IllegalArgumentException ex) {
                            throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, ex.getMessage());
                        }
                        break;
                    case MAX_FRAME_SIZE:
                        try {
                            configBuilder.setMaxFrameSize(value);
                        } catch (final IllegalArgumentException ex) {
                            throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, ex.getMessage());
                        }
                        break;
                    case MAX_HEADER_LIST_SIZE:
                        try {
                            configBuilder.setMaxHeaderListSize(value);
                        } catch (final IllegalArgumentException ex) {
                            throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, ex.getMessage());
                        }
                        break;
                }
            }
        }
        applyRemoteSettings(configBuilder.build());
    }

    private void produceOutput() throws HttpException, IOException {
        for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
            final Map.Entry<Integer, H2Stream> entry = it.next();
            final H2Stream stream = entry.getValue();
            if (!stream.isLocalClosed() && stream.getOutputWindow().get() > 0) {
                stream.produceOutput();
            }
            if (stream.isTerminated()) {
                it.remove();
                stream.releaseResources();
            }
            if (!outputQueue.isEmpty()) {
                break;
            }
        }
    }

    private void applyRemoteSettings(final H2Config config) throws H2ConnectionException {
        remoteConfig = config;

        hPackEncoder.setMaxTableSize(remoteConfig.getHeaderTableSize());
        final int delta = remoteConfig.getInitialWindowSize() - initOutputWinSize;
        initOutputWinSize = remoteConfig.getInitialWindowSize();

        if (delta != 0) {
            updateOutputWindow(0, connOutputWindow, delta);
            if (!streamMap.isEmpty()) {
                for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                    final Map.Entry<Integer, H2Stream> entry = it.next();
                    final H2Stream stream = entry.getValue();
                    try {
                        updateOutputWindow(stream.getId(), stream.getOutputWindow(), delta);
                    } catch (final ArithmeticException ex) {
                        throw new H2ConnectionException(H2Error.FLOW_CONTROL_ERROR, ex.getMessage());
                    }
                }
            }
        }
    }

    private void applyLocalSettings() throws H2ConnectionException {
        hPackDecoder.setMaxTableSize(localConfig.getHeaderTableSize());
        hPackDecoder.setMaxListSize(localConfig.getMaxHeaderListSize());

        final int delta = localConfig.getInitialWindowSize() - initInputWinSize;
        initInputWinSize = localConfig.getInitialWindowSize();

        if (delta != 0 && !streamMap.isEmpty()) {
            for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                final Map.Entry<Integer, H2Stream> entry = it.next();
                final H2Stream stream = entry.getValue();
                try {
                    updateInputWindow(stream.getId(), stream.getInputWindow(), delta);
                } catch (final ArithmeticException ex) {
                    throw new H2ConnectionException(H2Error.FLOW_CONTROL_ERROR, ex.getMessage());
                }
            }
        }
        lowMark = initInputWinSize / 2;
    }

    @Override
    public void close() throws IOException {
        ioSession.enqueue(ShutdownCommand.GRACEFUL, Command.Priority.IMMEDIATE);
    }

    @Override
    public void close(final CloseMode closeMode) {
        ioSession.close(closeMode);
    }

    @Override
    public boolean isOpen() {
        return connState == ConnectionHandshake.ACTIVE;
    }

    @Override
    public void setSocketTimeout(final Timeout timeout) {
        ioSession.setSocketTimeout(timeout);
    }

    @Override
    public SSLSession getSSLSession() {
        final TlsDetails tlsDetails = ioSession.getTlsDetails();
        return tlsDetails != null ? tlsDetails.getSSLSession() : null;
    }

    @Override
    public EndpointDetails getEndpointDetails() {
        if (endpointDetails == null) {
            endpointDetails = new BasicEndpointDetails(
                    ioSession.getRemoteAddress(),
                    ioSession.getLocalAddress(),
                    connMetrics,
                    ioSession.getSocketTimeout());
        }
        return endpointDetails;
    }

    @Override
    public Timeout getSocketTimeout() {
        return ioSession.getSocketTimeout();
    }

    @Override
    public ProtocolVersion getProtocolVersion() {
        return HttpVersion.HTTP_2;
    }

    @Override
    public SocketAddress getRemoteAddress() {
        return ioSession.getRemoteAddress();
    }

    @Override
    public SocketAddress getLocalAddress() {
        return ioSession.getLocalAddress();
    }

    void appendState(final StringBuilder buf) {
        buf.append("connState=").append(connState)
                .append(", connInputWindow=").append(connInputWindow)
                .append(", connOutputWindow=").append(connOutputWindow)
                .append(", outputQueue=").append(outputQueue.size())
                .append(", streamMap=").append(streamMap.size())
                .append(", processedRemoteStreamId=").append(processedRemoteStreamId);
    }

    private static class Continuation {

        final int streamId;
        final int type;
        final boolean endStream;
        final ByteArrayBuffer headerBuffer;

        private Continuation(final int streamId, final int type, final boolean endStream) {
            this.streamId = streamId;
            this.type = type;
            this.endStream = endStream;
            this.headerBuffer = new ByteArrayBuffer(1024);
        }

        void copyPayload(final ByteBuffer payload) {
            if (payload == null) {
                return;
            }
            headerBuffer.ensureCapacity(payload.remaining());
            payload.get(headerBuffer.array(), headerBuffer.length(), payload.remaining());
        }

        ByteBuffer getContent() {
            return ByteBuffer.wrap(headerBuffer.array(), 0, headerBuffer.length());
        }

    }

    private class H2StreamChannelImpl implements H2StreamChannel {

        private final int id;
        private final AtomicInteger inputWindow;
        private final AtomicInteger outputWindow;

        private volatile boolean idle;
        private volatile boolean remoteEndStream;
        private volatile boolean localEndStream;

        private volatile long deadline;

        H2StreamChannelImpl(final int id, final boolean idle, final int initialInputWindowSize, final int initialOutputWindowSize) {
            this.id = id;
            this.idle = idle;
            this.inputWindow = new AtomicInteger(initialInputWindowSize);
            this.outputWindow = new AtomicInteger(initialOutputWindowSize);
        }

        int getId() {
            return id;
        }

        AtomicInteger getOutputWindow() {
            return outputWindow;
        }

        AtomicInteger getInputWindow() {
            return inputWindow;
        }

        @Override
        public void submit(final List<Header> headers, final boolean endStream) throws IOException {
            ioSession.getLock().lock();
            try {
                if (headers == null || headers.isEmpty()) {
                    throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Message headers are missing");
                }
                if (localEndStream) {
                    return;
                }
                idle = false;
                commitHeaders(id, headers, endStream);
                if (endStream) {
                    localEndStream = true;
                }
            } finally {
                ioSession.getLock().unlock();
            }
        }

        @Override
        public void push(final List<Header> headers, final AsyncPushProducer pushProducer) throws HttpException, IOException {
            acceptPushRequest();
            final int promisedStreamId = generateStreamId();
            final H2StreamChannelImpl channel = new H2StreamChannelImpl(
                    promisedStreamId,
                    true,
                    localConfig.getInitialWindowSize(),
                    remoteConfig.getInitialWindowSize());
            final HttpCoreContext context = HttpCoreContext.create();
            context.setAttribute(HttpCoreContext.SSL_SESSION, getSSLSession());
            context.setAttribute(HttpCoreContext.CONNECTION_ENDPOINT, getEndpointDetails());
            final H2StreamHandler streamHandler = new ServerPushH2StreamHandler(
                    channel, httpProcessor, connMetrics, pushProducer, context);
            final H2Stream stream = new H2Stream(channel, streamHandler, false);
            streamMap.put(promisedStreamId, stream);

            ioSession.getLock().lock();
            try {
                if (localEndStream) {
                    stream.releaseResources();
                    return;
                }
                commitPushPromise(id, promisedStreamId, headers);
                idle = false;
            } finally {
                ioSession.getLock().unlock();
            }
        }

        @Override
        public void update(final int increment) throws IOException {
            if (remoteEndStream) {
                return;
            }
            incrementInputCapacity(0, connInputWindow, increment);
            incrementInputCapacity(id, inputWindow, increment);
        }

        @Override
        public int write(final ByteBuffer payload) throws IOException {
            ioSession.getLock().lock();
            try {
                if (localEndStream) {
                    return 0;
                }
                return streamData(id, outputWindow, payload);
            } finally {
                ioSession.getLock().unlock();
            }
        }

        @Override
        public void endStream(final List<? extends Header> trailers) throws IOException {
            ioSession.getLock().lock();
            try {
                if (localEndStream) {
                    return;
                }
                localEndStream = true;
                if (trailers != null && !trailers.isEmpty()) {
                    commitHeaders(id, trailers, true);
                } else {
                    final RawFrame frame = frameFactory.createData(id, null, true);
                    commitFrameInternal(frame);
                }
            } finally {
                ioSession.getLock().unlock();
            }
        }

        @Override
        public void endStream() throws IOException {
            endStream(null);
        }

        @Override
        public void requestOutput() {
            requestSessionOutput();
        }

        boolean isRemoteClosed() {
            return remoteEndStream;
        }

        void setRemoteEndStream() {
            remoteEndStream = true;
        }

        boolean isLocalClosed() {
            return localEndStream;
        }

        void setLocalEndStream() {
            localEndStream = true;
        }

        boolean isLocalReset() {
            return deadline > 0;
        }

        boolean isResetDeadline() {
            final long l = deadline;
            return l > 0 && l < System.currentTimeMillis();
        }

        boolean localReset(final int code) throws IOException {
            ioSession.getLock().lock();
            try {
                if (localEndStream) {
                    return false;
                }
                localEndStream = true;
                deadline = System.currentTimeMillis() + LINGER_TIME;
                if (!idle) {
                    final RawFrame resetStream = frameFactory.createResetStream(id, code);
                    commitFrameInternal(resetStream);
                    return true;
                }
                return false;
            } finally {
                ioSession.getLock().unlock();
            }
        }

        boolean localReset(final H2Error error) throws IOException {
            return localReset(error!= null ? error.getCode() : H2Error.INTERNAL_ERROR.getCode());
        }

        @Override
        public boolean cancel() {
            try {
                return localReset(H2Error.CANCEL);
            } catch (final IOException ignore) {
                return false;
            }
        }

        void appendState(final StringBuilder buf) {
            buf.append("id=").append(id)
                    .append(", connState=").append(connState)
                    .append(", inputWindow=").append(inputWindow)
                    .append(", outputWindow=").append(outputWindow)
                    .append(", localEndStream=").append(localEndStream)
                    .append(", idle=").append(idle);
        }

        @Override
        public String toString() {
            final StringBuilder buf = new StringBuilder();
            buf.append("[");
            appendState(buf);
            buf.append("]");
            return buf.toString();
        }

    }

    static class H2Stream {

        private final H2StreamChannelImpl channel;
        private final H2StreamHandler handler;
        private final boolean remoteInitiated;

        private H2Stream(
                final H2StreamChannelImpl channel,
                final H2StreamHandler handler,
                final boolean remoteInitiated) {
            this.channel = channel;
            this.handler = handler;
            this.remoteInitiated = remoteInitiated;
        }

        int getId() {
            return channel.getId();
        }

        boolean isRemoteInitiated() {
            return remoteInitiated;
        }

        AtomicInteger getOutputWindow() {
            return channel.getOutputWindow();
        }

        AtomicInteger getInputWindow() {
            return channel.getInputWindow();
        }

        boolean isTerminated() {
            return channel.isLocalClosed() && (channel.isRemoteClosed() || channel.isResetDeadline());
        }

        boolean isRemoteClosed() {
            return channel.isRemoteClosed();
        }

        boolean isLocalClosed() {
            return channel.isLocalClosed();
        }

        boolean isLocalReset() {
            return channel.isLocalReset();
        }

        void setRemoteEndStream() {
            channel.setRemoteEndStream();
        }

        void consumePromise(final List<Header> headers) throws HttpException, IOException {
            try {
                handler.consumePromise(headers);
                channel.setLocalEndStream();
            } catch (final ProtocolException ex) {
                localReset(ex, H2Error.PROTOCOL_ERROR);
            }
        }

        void consumeHeader(final List<Header> headers) throws HttpException, IOException {
            try {
                handler.consumeHeader(headers, channel.isRemoteClosed());
            } catch (final ProtocolException ex) {
                localReset(ex, H2Error.PROTOCOL_ERROR);
            }
        }

        void consumeData(final ByteBuffer src) throws HttpException, IOException {
            try {
                handler.consumeData(src, channel.isRemoteClosed());
            } catch (final CharacterCodingException ex) {
                localReset(ex, H2Error.INTERNAL_ERROR);
            } catch (final ProtocolException ex) {
                localReset(ex, H2Error.PROTOCOL_ERROR);
            }
        }

        boolean isOutputReady() {
            return handler.isOutputReady();
        }

        void produceOutput() throws HttpException, IOException {
            try {
                handler.produceOutput();
            } catch (final ProtocolException ex) {
                localReset(ex, H2Error.PROTOCOL_ERROR);
            }
        }

        void produceInputCapacityUpdate() throws IOException {
            handler.updateInputCapacity();
        }

        void reset(final Exception cause) {
            channel.setRemoteEndStream();
            channel.setLocalEndStream();
            handler.failed(cause);
        }

        void localReset(final Exception cause, final int code) throws IOException {
            channel.localReset(code);
            handler.failed(cause);
        }

        void localReset(final Exception cause, final H2Error error) throws IOException {
            localReset(cause, error != null ? error.getCode() : H2Error.INTERNAL_ERROR.getCode());
        }

        void localReset(final H2StreamResetException ex) throws IOException {
            localReset(ex, ex.getCode());
        }

        void handle(final HttpException ex) throws IOException, HttpException {
            handler.handle(ex, channel.isRemoteClosed());
        }

        HandlerFactory<AsyncPushConsumer> getPushHandlerFactory() {
            return handler.getPushHandlerFactory();
        }

        void cancel() {
            reset(new CancellationException("HTTP/2 message exchange cancelled"));
        }

        boolean abort() {
            final boolean cancelled = channel.cancel();
            handler.failed(new CancellationException("HTTP/2 message exchange cancelled"));
            return cancelled;
        }

        void releaseResources() {
            handler.releaseResources();
            channel.requestOutput();
        }

        void appendState(final StringBuilder buf) {
            buf.append("channel=[");
            channel.appendState(buf);
            buf.append("]");
        }

        @Override
        public String toString() {
            final StringBuilder buf = new StringBuilder();
            buf.append("[");
            appendState(buf);
            buf.append("]");
            return buf.toString();
        }

    }

}