/*
    * Copyright 2019 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.ByteBufAllocator;
  import io.netty.channel.Channel;
  import io.netty.channel.ChannelConfig;
  import io.netty.channel.ChannelFuture;
  import io.netty.channel.ChannelFutureListener;
  import io.netty.channel.ChannelHandler;
  import io.netty.channel.ChannelHandlerContext;
  import io.netty.channel.ChannelId;
  import io.netty.channel.ChannelMetadata;
  import io.netty.channel.ChannelOption;
  import io.netty.channel.ChannelOutboundBuffer;
  import io.netty.channel.ChannelPipeline;
  import io.netty.channel.ChannelProgressivePromise;
  import io.netty.channel.ChannelPromise;
  import io.netty.channel.DefaultChannelConfig;
  import io.netty.channel.DefaultChannelPipeline;
  import io.netty.channel.EventLoop;
  import io.netty.channel.MessageSizeEstimator;
  import io.netty.channel.RecvByteBufAllocator;
  import io.netty.channel.VoidChannelPromise;
  import io.netty.channel.WriteBufferWaterMark;
  import io.netty.channel.socket.ChannelInputShutdownReadComplete;
  import io.netty.channel.socket.ChannelOutputShutdownEvent;
  import io.netty.handler.codec.http2.Http2FrameCodec.DefaultHttp2FrameStream;
  import io.netty.handler.ssl.SslCloseCompletionEvent;
  import io.netty.util.DefaultAttributeMap;
  import io.netty.util.ReferenceCountUtil;
  import io.netty.util.internal.ObjectUtil;
  import io.netty.util.internal.StringUtil;
  import io.netty.util.internal.logging.InternalLogger;
  import io.netty.util.internal.logging.InternalLoggerFactory;
  
  import java.io.IOException;
  import java.net.SocketAddress;
  import java.nio.channels.ClosedChannelException;
  import java.util.ArrayDeque;
  import java.util.Map;
  import java.util.Queue;
  import java.util.concurrent.RejectedExecutionException;
  import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
  import java.util.concurrent.atomic.AtomicLongFieldUpdater;
  
  import static io.netty.handler.codec.http2.Http2CodecUtil.isStreamIdValid;
  import static io.netty.util.internal.ObjectUtil.checkNotNull;
  import static java.lang.Math.min;
  
  abstract class AbstractHttp2StreamChannel extends DefaultAttributeMap implements Http2StreamChannel {
  
      static final Http2FrameStreamVisitor WRITABLE_VISITOR = new Http2FrameStreamVisitor() {
          @Override
          public boolean visit(Http2FrameStream stream) {
              final AbstractHttp2StreamChannel childChannel = (AbstractHttp2StreamChannel)
                      ((DefaultHttp2FrameStream) stream).attachment;
              childChannel.trySetWritable();
              return true;
          }
      };
  
      static final Http2FrameStreamVisitor CHANNEL_INPUT_SHUTDOWN_READ_COMPLETE_VISITOR =
              new UserEventStreamVisitor(ChannelInputShutdownReadComplete.INSTANCE);
  
      static final Http2FrameStreamVisitor CHANNEL_OUTPUT_SHUTDOWN_EVENT_VISITOR =
              new UserEventStreamVisitor(ChannelOutputShutdownEvent.INSTANCE);
  
      static final Http2FrameStreamVisitor SSL_CLOSE_COMPLETION_EVENT_VISITOR =
              new UserEventStreamVisitor(SslCloseCompletionEvent.SUCCESS);
  
      private static final InternalLogger logger = InternalLoggerFactory.getInstance(AbstractHttp2StreamChannel.class);
  
      private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16);
  
      /**
       * Number of bytes to consider non-payload messages. 9 is arbitrary, but also the minimum size of an HTTP/2 frame.
       * Primarily is non-zero.
       */
      private static final int MIN_HTTP2_FRAME_SIZE = 9;
  
      /**
       * {@link Http2FrameStreamVisitor} that fires the user event for every active stream pipeline.
       */
      private static final class UserEventStreamVisitor implements Http2FrameStreamVisitor {
  
         private final Object event;
 
         UserEventStreamVisitor(Object event) {
             this.event = checkNotNull(event, "event");
         }
 
         @Override
         public boolean visit(Http2FrameStream stream) {
             final AbstractHttp2StreamChannel childChannel = (AbstractHttp2StreamChannel)
                     ((DefaultHttp2FrameStream) stream).attachment;
             childChannel.pipeline().fireUserEventTriggered(event);
             return true;
         }
     }
 
     /**
      * Returns the flow-control size for DATA frames, and {@value MIN_HTTP2_FRAME_SIZE} for all other frames.
      */
     private static final class FlowControlledFrameSizeEstimator implements MessageSizeEstimator {
 
         static final FlowControlledFrameSizeEstimator INSTANCE = new FlowControlledFrameSizeEstimator();
 
         private static final Handle HANDLE_INSTANCE = new Handle() {
             @Override
             public int size(Object msg) {
                 return msg instanceof Http2DataFrame ?
                         // Guard against overflow.
                         (int) min(Integer.MAX_VALUE, ((Http2DataFrame) msg).initialFlowControlledBytes() +
                                 (long) MIN_HTTP2_FRAME_SIZE) : MIN_HTTP2_FRAME_SIZE;
             }
         };
 
         @Override
         public Handle newHandle() {
             return HANDLE_INSTANCE;
         }
     }
 
     private static final AtomicLongFieldUpdater<AbstractHttp2StreamChannel> TOTAL_PENDING_SIZE_UPDATER =
             AtomicLongFieldUpdater.newUpdater(AbstractHttp2StreamChannel.class, "totalPendingSize");
 
     private static final AtomicIntegerFieldUpdater<AbstractHttp2StreamChannel> UNWRITABLE_UPDATER =
             AtomicIntegerFieldUpdater.newUpdater(AbstractHttp2StreamChannel.class, "unwritable");
 
     private static void windowUpdateFrameWriteComplete(ChannelFuture future, Channel streamChannel) {
         Throwable cause = future.cause();
         if (cause != null) {
             Throwable unwrappedCause;
             // Unwrap if needed
             if (cause instanceof Http2FrameStreamException && (unwrappedCause = cause.getCause()) != null) {
                 cause = unwrappedCause;
             }
 
             // Notify the child-channel and close it.
             streamChannel.pipeline().fireExceptionCaught(cause);
             streamChannel.unsafe().close(streamChannel.unsafe().voidPromise());
         }
     }
 
     private final ChannelFutureListener windowUpdateFrameWriteListener = new ChannelFutureListener() {
         @Override
         public void operationComplete(ChannelFuture future) {
             windowUpdateFrameWriteComplete(future, AbstractHttp2StreamChannel.this);
         }
     };
 
     /**
      * The current status of the read-processing for a {@link AbstractHttp2StreamChannel}.
      */
     private enum ReadStatus {
         /**
          * No read in progress and no read was requested (yet)
          */
         IDLE,
 
         /**
          * Reading in progress
          */
         IN_PROGRESS,
 
         /**
          * A read operation was requested.
          */
         REQUESTED
     }
 
     private final Http2StreamChannelConfig config = new Http2StreamChannelConfig(this);
     private final Http2ChannelUnsafe unsafe = new Http2ChannelUnsafe();
     private final ChannelId channelId;
     private final ChannelPipeline pipeline;
     private final DefaultHttp2FrameStream stream;
     private final ChannelPromise closePromise;
 
     private volatile boolean registered;
 
     private volatile long totalPendingSize;
     private volatile int unwritable;
 
     // Cached to reduce GC
     private Runnable fireChannelWritabilityChangedTask;
 
     private boolean outboundClosed;
     private int flowControlledBytes;
 
     /**
      * This variable represents if a read is in progress for the current channel or was requested.
      * Note that depending upon the {@link RecvByteBufAllocator} behavior a read may extend beyond the
      * {@link Http2ChannelUnsafe#beginRead()} method scope. The {@link Http2ChannelUnsafe#beginRead()} loop may
      * drain all pending data, and then if the parent channel is reading this channel may still accept frames.
      */
     private ReadStatus readStatus = ReadStatus.IDLE;
 
     private Queue<Object> inboundBuffer;
 
     /** {@code true} after the first HEADERS frame has been written **/
     private boolean firstFrameWritten;
     private boolean readCompletePending;
 
     AbstractHttp2StreamChannel(DefaultHttp2FrameStream stream, int id, ChannelHandler inboundHandler) {
         this.stream = stream;
         stream.attachment = this;
         pipeline = new DefaultChannelPipeline(this) {
             @Override
             protected void incrementPendingOutboundBytes(long size) {
                 AbstractHttp2StreamChannel.this.incrementPendingOutboundBytes(size, true);
             }
 
             @Override
             protected void decrementPendingOutboundBytes(long size) {
                 AbstractHttp2StreamChannel.this.decrementPendingOutboundBytes(size, true);
             }
 
             @Override
             protected void onUnhandledInboundException(Throwable cause) {
                 // Ensure we use the correct Http2Error to close the channel.
                 if (cause instanceof Http2FrameStreamException) {
                     closeWithError(((Http2FrameStreamException) cause).error());
                     return;
                 } else {
                     Http2Exception exception = Http2CodecUtil.getEmbeddedHttp2Exception(cause);
                     if (exception != null) {
                         closeWithError(exception.error());
                         return;
                     }
                 }
                 super.onUnhandledInboundException(cause);
             }
         };
 
         closePromise = pipeline.newPromise();
         channelId = new Http2StreamChannelId(parent().id(), id);
 
         if (inboundHandler != null) {
             // Add the handler to the pipeline now that we are registered.
             pipeline.addLast(inboundHandler);
         }
     }
 
     private void incrementPendingOutboundBytes(long size, boolean invokeLater) {
         if (size == 0) {
             return;
         }
 
         long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, size);
         if (newWriteBufferSize > config().getWriteBufferHighWaterMark()) {
             setUnwritable(invokeLater);
         }
     }
 
     private void decrementPendingOutboundBytes(long size, boolean invokeLater) {
         if (size == 0) {
             return;
         }
 
         long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);
         // Once the totalPendingSize dropped below the low water-mark we can mark the child channel
         // as writable again. Before doing so we also need to ensure the parent channel is writable to
         // prevent excessive buffering in the parent outbound buffer. If the parent is not writable
         // we will mark the child channel as writable once the parent becomes writable by calling
         // trySetWritable() later.
         if (newWriteBufferSize < config().getWriteBufferLowWaterMark() && parent().isWritable()) {
             setWritable(invokeLater);
         }
     }
 
     final void trySetWritable() {
         // The parent is writable again but the child channel itself may still not be writable.
         // Lets try to set the child channel writable to match the state of the parent channel
         // if (and only if) the totalPendingSize is smaller then the low water-mark.
         // If this is not the case we will try again later once we drop under it.
         if (totalPendingSize < config().getWriteBufferLowWaterMark()) {
             setWritable(false);
         }
     }
 
     private void setWritable(boolean invokeLater) {
         for (;;) {
             final int oldValue = unwritable;
             final int newValue = oldValue & ~1;
             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
                 if (oldValue != 0 && newValue == 0) {
                     fireChannelWritabilityChanged(invokeLater);
                 }
                 break;
             }
         }
     }
 
     private void setUnwritable(boolean invokeLater) {
         for (;;) {
             final int oldValue = unwritable;
             final int newValue = oldValue | 1;
             if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
                 if (oldValue == 0) {
                     fireChannelWritabilityChanged(invokeLater);
                 }
                 break;
             }
         }
     }
 
     private void fireChannelWritabilityChanged(boolean invokeLater) {
         final ChannelPipeline pipeline = pipeline();
         if (invokeLater) {
             Runnable task = fireChannelWritabilityChangedTask;
             if (task == null) {
                 fireChannelWritabilityChangedTask = task = new Runnable() {
                     @Override
                     public void run() {
                         pipeline.fireChannelWritabilityChanged();
                     }
                 };
             }
             eventLoop().execute(task);
         } else {
             pipeline.fireChannelWritabilityChanged();
         }
     }
     @Override
     public Http2FrameStream stream() {
         return stream;
     }
 
     void closeOutbound() {
         outboundClosed = true;
     }
 
     void streamClosed() {
         unsafe.readEOS();
         // Attempt to drain any queued data from the queue and deliver it to the application before closing this
         // channel.
         unsafe.doBeginRead();
     }
 
     @Override
     public ChannelMetadata metadata() {
         return METADATA;
     }
 
     @Override
     public ChannelConfig config() {
         return config;
     }
 
     @Override
     public boolean isOpen() {
         return !closePromise.isDone();
     }
 
     @Override
     public boolean isActive() {
         return isOpen();
     }
 
     @Override
     public boolean isWritable() {
         return unwritable == 0;
     }
 
     @Override
     public ChannelId id() {
         return channelId;
     }
 
     @Override
     public EventLoop eventLoop() {
         return parent().eventLoop();
     }
 
     @Override
     public Channel parent() {
         return parentContext().channel();
     }
 
     @Override
     public boolean isRegistered() {
         return registered;
     }
 
     @Override
     public SocketAddress localAddress() {
         return parent().localAddress();
     }
 
     @Override
     public SocketAddress remoteAddress() {
         return parent().remoteAddress();
     }
 
     @Override
     public ChannelFuture closeFuture() {
         return closePromise;
     }
 
     @Override
     public long bytesBeforeUnwritable() {
         // +1 because writability doesn't change until the threshold is crossed (not equal to).
         long bytes = config().getWriteBufferHighWaterMark() - totalPendingSize + 1;
         // If bytes is negative we know we are not writable, but if bytes is non-negative we have to check
         // writability. Note that totalPendingSize and isWritable() use different volatile variables that are not
         // synchronized together. totalPendingSize will be updated before isWritable().
         return bytes > 0 && isWritable() ? bytes : 0;
     }
 
     @Override
     public long bytesBeforeWritable() {
         // +1 because writability doesn't change until the threshold is crossed (not equal to).
         long bytes = totalPendingSize - config().getWriteBufferLowWaterMark() + 1;
         // If bytes is negative we know we are writable, but if bytes is non-negative we have to check writability.
         // Note that totalPendingSize and isWritable() use different volatile variables that are not synchronized
         // together. totalPendingSize will be updated before isWritable().
         return bytes <= 0 || isWritable() ? 0 : bytes;
     }
 
     @Override
     public Unsafe unsafe() {
         return unsafe;
     }
 
     @Override
     public ChannelPipeline pipeline() {
         return pipeline;
     }
 
     @Override
     public ByteBufAllocator alloc() {
         return config().getAllocator();
     }
 
     @Override
     public Channel read() {
         pipeline().read();
         return this;
     }
 
     @Override
     public Channel flush() {
         pipeline().flush();
         return this;
     }
 
     @Override
     public ChannelFuture bind(SocketAddress localAddress) {
         return pipeline().bind(localAddress);
     }
 
     @Override
     public ChannelFuture connect(SocketAddress remoteAddress) {
         return pipeline().connect(remoteAddress);
     }
 
     @Override
     public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) {
         return pipeline().connect(remoteAddress, localAddress);
     }
 
     @Override
     public ChannelFuture disconnect() {
         return pipeline().disconnect();
     }
 
     @Override
     public ChannelFuture close() {
         return pipeline().close();
     }
 
     @Override
     public ChannelFuture deregister() {
         return pipeline().deregister();
     }
 
     @Override
     public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
         return pipeline().bind(localAddress, promise);
     }
 
     @Override
     public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
         return pipeline().connect(remoteAddress, promise);
     }
 
     @Override
     public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
         return pipeline().connect(remoteAddress, localAddress, promise);
     }
 
     @Override
     public ChannelFuture disconnect(ChannelPromise promise) {
         return pipeline().disconnect(promise);
     }
 
     @Override
     public ChannelFuture close(ChannelPromise promise) {
         return pipeline().close(promise);
     }
 
     @Override
     public ChannelFuture deregister(ChannelPromise promise) {
         return pipeline().deregister(promise);
     }
 
     @Override
     public ChannelFuture write(Object msg) {
         return pipeline().write(msg);
     }
 
     @Override
     public ChannelFuture write(Object msg, ChannelPromise promise) {
         return pipeline().write(msg, promise);
     }
 
     @Override
     public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
         return pipeline().writeAndFlush(msg, promise);
     }
 
     @Override
     public ChannelFuture writeAndFlush(Object msg) {
         return pipeline().writeAndFlush(msg);
     }
 
     @Override
     public ChannelPromise newPromise() {
         return pipeline().newPromise();
     }
 
     @Override
     public ChannelProgressivePromise newProgressivePromise() {
         return pipeline().newProgressivePromise();
     }
 
     @Override
     public ChannelFuture newSucceededFuture() {
         return pipeline().newSucceededFuture();
     }
 
     @Override
     public ChannelFuture newFailedFuture(Throwable cause) {
         return pipeline().newFailedFuture(cause);
     }
 
     @Override
     public ChannelPromise voidPromise() {
         return pipeline().voidPromise();
     }
 
     @Override
     public int hashCode() {
         return id().hashCode();
     }
 
     @Override
     public boolean equals(Object o) {
         return this == o;
     }
 
     @Override
     public int compareTo(Channel o) {
         if (this == o) {
             return 0;
         }
 
         return id().compareTo(o.id());
     }
 
     @Override
     public String toString() {
         return parent().toString() + "(H2 - " + stream + ')';
     }
 
     /**
      * Receive a read message. This does not notify handlers unless a read is in progress on the
      * channel.
      */
     void fireChildRead(Http2Frame frame) {
         assert eventLoop().inEventLoop();
         if (!isActive()) {
             ReferenceCountUtil.release(frame);
         } else if (readStatus != ReadStatus.IDLE) {
             // If a read is in progress or has been requested, there cannot be anything in the queue,
             // otherwise we would have drained it from the queue and processed it during the read cycle.
             assert inboundBuffer == null || inboundBuffer.isEmpty();
             final RecvByteBufAllocator.Handle allocHandle = unsafe.recvBufAllocHandle();
 
             unsafe.doRead0(frame, allocHandle);
             // We currently don't need to check for readEOS because the parent channel and child channel are limited
             // to the same EventLoop thread. There are a limited number of frame types that may come after EOS is
             // read (unknown, reset) and the trade off is less conditionals for the hot path (headers/data) at the
             // cost of additional readComplete notifications on the rare path.
             if (allocHandle.continueReading()) {
                 maybeAddChannelToReadCompletePendingQueue();
             } else {
                 unsafe.notifyReadComplete(allocHandle, true, false);
             }
         } else {
             if (inboundBuffer == null) {
                 inboundBuffer = new ArrayDeque<Object>(4);
             }
             inboundBuffer.add(frame);
         }
     }
 
     void fireChildReadComplete() {
         assert eventLoop().inEventLoop();
         assert readStatus != ReadStatus.IDLE || !readCompletePending;
         unsafe.notifyReadComplete(unsafe.recvBufAllocHandle(), false, false);
     }
 
     final void closeWithError(Http2Error error) {
         assert eventLoop().inEventLoop();
         unsafe.close(unsafe.voidPromise(), error);
     }
 
     private final class Http2ChannelUnsafe implements Unsafe {
         private final VoidChannelPromise unsafeVoidPromise =
                 new VoidChannelPromise(AbstractHttp2StreamChannel.this, false);
         @SuppressWarnings("deprecation")
         private RecvByteBufAllocator.Handle recvHandle;
         private boolean writeDoneAndNoFlush;
         private boolean closeInitiated;
         private boolean readEOS;
 
         private boolean receivedEndOfStream;
         private boolean sentEndOfStream;
 
         @Override
         public void connect(final SocketAddress remoteAddress,
                             SocketAddress localAddress, final ChannelPromise promise) {
             if (!promise.setUncancellable()) {
                 return;
             }
             promise.setFailure(new UnsupportedOperationException());
         }
 
         @Override
         public RecvByteBufAllocator.Handle recvBufAllocHandle() {
             if (recvHandle == null) {
                 recvHandle = config().getRecvByteBufAllocator().newHandle();
                 recvHandle.reset(config());
             }
             return recvHandle;
         }
 
         @Override
         public SocketAddress localAddress() {
             return parent().unsafe().localAddress();
         }
 
         @Override
         public SocketAddress remoteAddress() {
             return parent().unsafe().remoteAddress();
         }
 
         @Override
         public void register(EventLoop eventLoop, ChannelPromise promise) {
             if (!promise.setUncancellable()) {
                 return;
             }
             if (registered) {
                 promise.setFailure(new UnsupportedOperationException("Re-register is not supported"));
                 return;
             }
 
             registered = true;
 
             promise.setSuccess();
 
             pipeline().fireChannelRegistered();
             if (isActive()) {
                 pipeline().fireChannelActive();
             }
         }
 
         @Override
         public void bind(SocketAddress localAddress, ChannelPromise promise) {
             if (!promise.setUncancellable()) {
                 return;
             }
             promise.setFailure(new UnsupportedOperationException());
         }
 
         @Override
         public void disconnect(ChannelPromise promise) {
             close(promise);
         }
 
         @Override
         public void close(final ChannelPromise promise) {
             close(promise, Http2Error.CANCEL);
         }
 
         void close(final ChannelPromise promise, Http2Error error) {
             if (!promise.setUncancellable()) {
                 return;
             }
             if (closeInitiated) {
                 if (closePromise.isDone()) {
                     // Closed already.
                     promise.setSuccess();
                 } else if (!(promise instanceof VoidChannelPromise)) { // Only needed if no VoidChannelPromise.
                     // This means close() was called before so we just register a listener and return
                     closePromise.addListener(new ChannelFutureListener() {
                         @Override
                         public void operationComplete(ChannelFuture future) {
                             promise.setSuccess();
                         }
                     });
                 }
                 return;
             }
             closeInitiated = true;
             // Just set to false as removing from an underlying queue would even be more expensive.
             readCompletePending = false;
 
             final boolean wasActive = isActive();
 
             // There is no need to update the local window as once the stream is closed all the pending bytes will be
             // given back to the connection window by the controller itself.
 
             // Only ever send a reset frame if the connection is still alive and if the stream was created before
             // as otherwise we may send a RST on a stream in an invalid state and cause a connection error.
             if (parent().isActive() && isStreamIdValid(stream.id()) &&
                     // Also ensure the stream was never "closed" before.
                     !readEOS && !(receivedEndOfStream && sentEndOfStream)) {
                 Http2StreamFrame resetFrame = new DefaultHttp2ResetFrame(error).stream(stream());
                 write(resetFrame, unsafe().voidPromise());
                 flush();
             }
 
             if (inboundBuffer != null) {
                 for (;;) {
                     Object msg = inboundBuffer.poll();
                     if (msg == null) {
                         break;
                     }
                     ReferenceCountUtil.release(msg);
                 }
                 inboundBuffer = null;
             }
 
             // The promise should be notified before we call fireChannelInactive().
             outboundClosed = true;
             closePromise.setSuccess();
             promise.setSuccess();
 
             fireChannelInactiveAndDeregister(voidPromise(), wasActive);
         }
 
         @Override
         public void closeForcibly() {
             close(unsafe().voidPromise());
         }
 
         @Override
         public void deregister(ChannelPromise promise) {
             fireChannelInactiveAndDeregister(promise, false);
         }
 
         private void fireChannelInactiveAndDeregister(final ChannelPromise promise,
                                                       final boolean fireChannelInactive) {
             if (!promise.setUncancellable()) {
                 return;
             }
 
             if (!registered) {
                 promise.setSuccess();
                 return;
             }
 
             // As a user may call deregister() from within any method while doing processing in the ChannelPipeline,
             // we need to ensure we do the actual deregister operation later. This is necessary to preserve the
             // behavior of the AbstractChannel, which always invokes channelUnregistered and channelInactive
             // events 'later' to ensure the current events in the handler are completed before these events.
             //
             // See:
             // https://github.com/netty/netty/issues/4435
             invokeLater(promise.channel(), new Runnable() {
                 @Override
                 public void run() {
                     if (fireChannelInactive) {
                         pipeline.fireChannelInactive();
                     }
                     // The user can fire `deregister` events multiple times but we only want to fire the pipeline
                     // event if the channel was actually registered.
                     if (registered) {
                         registered = false;
                         pipeline.fireChannelUnregistered();
                     }
                     safeSetSuccess(promise);
                 }
             });
         }
 
         private void safeSetSuccess(ChannelPromise promise) {
             if (!(promise instanceof VoidChannelPromise) && !promise.trySuccess()) {
                 logger.warn("{} Failed to mark a promise as success because it is done already: {}",
                         promise.channel(), promise);
             }
         }
 
         private void invokeLater(Channel channel, Runnable task) {
             try {
                 // This method is used by outbound operation implementations to trigger an inbound event later.
                 // They do not trigger an inbound event immediately because an outbound operation might have been
                 // triggered by another inbound event handler method.  If fired immediately, the call stack
                 // will look like this for example:
                 //
                 //   handlerA.inboundBufferUpdated() - (1) an inbound handler method closes a connection.
                 //   -> handlerA.ctx.close()
                 //     -> channel.unsafe.close()
                 //       -> handlerA.channelInactive() - (2) another inbound handler method called while in (1) yet
                 //
                 // which means the execution of two inbound handler methods of the same handler overlap undesirably.
                 eventLoop().execute(task);
             } catch (RejectedExecutionException e) {
                 logger.warn("{} Can't invoke task later as EventLoop rejected it", channel, e);
             }
         }
 
         @Override
         public void beginRead() {
             if (!isActive()) {
                 return;
             }
             updateLocalWindowIfNeeded();
 
             switch (readStatus) {
                 case IDLE:
                     readStatus = ReadStatus.IN_PROGRESS;
                     doBeginRead();
                     break;
                 case IN_PROGRESS:
                     readStatus = ReadStatus.REQUESTED;
                     break;
                 default:
                     break;
             }
         }
 
         private Object pollQueuedMessage() {
             return inboundBuffer == null ? null : inboundBuffer.poll();
         }
 
         void doBeginRead() {
             if (readStatus == ReadStatus.IDLE) {
                 // Don't wait for the user to request a read to notify of channel closure.
                 if (readEOS && (inboundBuffer == null || inboundBuffer.isEmpty())) {
                     // Double check there is nothing left to flush such as a window update frame.
                     flush();
                     unsafe.closeForcibly();
                 }
             } else {
                 do { // Process messages until there are none left (or the user stopped requesting) and also handle EOS.
                     Object message = pollQueuedMessage();
                     if (message == null) {
                         // Double check there is nothing left to flush such as a window update frame.
                         flush();
                         if (readEOS) {
                             unsafe.closeForcibly();
                         }
                         break;
                     }
                     final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
                     allocHandle.reset(config());
                     boolean continueReading = false;
                     do {
                         doRead0((Http2Frame) message, allocHandle);
                     } while ((readEOS || (continueReading = allocHandle.continueReading()))
                             && (message = pollQueuedMessage()) != null);
 
                     if (continueReading && isParentReadInProgress() && !readEOS) {
                         // Currently the parent and child channel are on the same EventLoop thread. If the parent is
                         // currently reading it is possible that more frames will be delivered to this child channel. In
                         // the case that this child channel still wants to read we delay the channelReadComplete on this
                         // child channel until the parent is done reading.
                         maybeAddChannelToReadCompletePendingQueue();
                     } else {
                         notifyReadComplete(allocHandle, true, true);
 
                         // While in the read loop reset the readState AFTER calling readComplete (or other pipeline
                         // callbacks) to prevents re-entry into this method (if autoRead is disabled and the user calls
                         // read on each readComplete) and StackOverflowException.
                         resetReadStatus();
                     }
                 } while (readStatus != ReadStatus.IDLE);
             }
         }
 
         void readEOS() {
             readEOS = true;
         }
 
         private boolean updateLocalWindowIfNeeded() {
             if (flowControlledBytes != 0 && !parentContext().isRemoved() && config.autoStreamFlowControl) {
                 int bytes = flowControlledBytes;
                 flowControlledBytes = 0;
                 writeWindowUpdateFrame(new DefaultHttp2WindowUpdateFrame(bytes).stream(stream));
                 return true;
             }
             return false;
         }
 
         void updateLocalWindowIfNeededAndFlush() {
             if (updateLocalWindowIfNeeded()) {
                 flush();
             }
         }
 
         private void resetReadStatus() {
             readStatus = readStatus == ReadStatus.REQUESTED ? ReadStatus.IN_PROGRESS : ReadStatus.IDLE;
         }
 
         void notifyReadComplete(RecvByteBufAllocator.Handle allocHandle, boolean forceReadComplete,
                                 boolean inReadLoop) {
             if (!readCompletePending && !forceReadComplete) {
                 return;
             }
             // Set to false just in case we added the channel multiple times before.
             readCompletePending = false;
 
             if (!inReadLoop) {
                 // While in the read loop we reset the state after calling pipeline methods to prevent StackOverflow.
                 resetReadStatus();
             }
 
             allocHandle.readComplete();
             pipeline().fireChannelReadComplete();
             // Reading data may result in frames being written (e.g. WINDOW_UPDATE, RST, etc..). If the parent
             // channel is not currently reading we need to force a flush at the child channel, because we cannot
             // rely upon flush occurring in channelReadComplete on the parent channel.
             flush();
             if (readEOS) {
                 unsafe.closeForcibly();
             }
         }
 
         @SuppressWarnings("deprecation")
         void doRead0(Http2Frame frame, RecvByteBufAllocator.Handle allocHandle) {
             final int bytes;
             if (frame instanceof Http2DataFrame) {
                 bytes = ((Http2DataFrame) frame).initialFlowControlledBytes();
                 // It is important that we increment the flowControlledBytes before we call fireChannelRead(...)
                 // as it may cause a read() that will call updateLocalWindowIfNeeded() and we need to ensure
                 // in this case that we accounted for it.
                 //
                 // See https://github.com/netty/netty/issues/9663
                 flowControlledBytes += bytes;
             } else {
                 bytes = MIN_HTTP2_FRAME_SIZE;
             }
 
             // Let's keep track of what we received as the stream state itself will only be updated once the frame
             // was dispatched for reading which might cause problems if we try to close the channel in a write future.
             receivedEndOfStream |= isEndOfStream(frame);
 
             // Update before firing event through the pipeline to be consistent with other Channel implementation.
             allocHandle.attemptedBytesRead(bytes);
             allocHandle.lastBytesRead(bytes);
             allocHandle.incMessagesRead(1);
 
             pipeline().fireChannelRead(frame);
         }
 
         private ChannelFuture writeWindowUpdateFrame(Http2WindowUpdateFrame windowUpdateFrame) {
             ChannelFuture future = write0(parentContext(), windowUpdateFrame);
             // window update frames are commonly swallowed by the Http2FrameCodec and the promise is synchronously
             // completed but the flow controller _may_ have generated a wire level WINDOW_UPDATE. Therefore we need,
             // to assume there was a write done that needs to be flushed or we risk flow control starvation.
             writeDoneAndNoFlush = true;
             // Add a listener which will notify and teardown the stream
             // when a window update fails if needed or check the result of the future directly if it was completed
             // already.
             // See https://github.com/netty/netty/issues/9663
             if (future.isDone()) {
                 windowUpdateFrameWriteComplete(future, AbstractHttp2StreamChannel.this);
             } else {
                 future.addListener(windowUpdateFrameWriteListener);
             }
             return future;
         }
 
         @Override
         public void write(Object msg, final ChannelPromise promise) {
             // After this point its not possible to cancel a write anymore.
             if (!promise.setUncancellable()) {
                 ReferenceCountUtil.release(msg);
                 return;
             }
 
             if (!isActive() ||
                     // Once the outbound side was closed we should not allow header / data frames
                     outboundClosed && (msg instanceof Http2HeadersFrame || msg instanceof Http2DataFrame)) {
                 ReferenceCountUtil.release(msg);
                 promise.setFailure(new ClosedChannelException());
                 return;
             }
 
             try {
                 if (msg instanceof Http2StreamFrame) {
                     Http2StreamFrame frame = validateStreamFrame((Http2StreamFrame) msg).stream(stream());
                     if (msg instanceof Http2WindowUpdateFrame) {
                         Http2WindowUpdateFrame updateFrame = (Http2WindowUpdateFrame) msg;
                         if (config.autoStreamFlowControl) {
                             ReferenceCountUtil.release(msg);
                             promise.setFailure(new UnsupportedOperationException(
                                     Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL + " is set to false"));
                             return;
                         }
                         try {
                             ObjectUtil.checkInRange(updateFrame.windowSizeIncrement(), 0,
                                     flowControlledBytes, "windowSizeIncrement");
                         } catch (RuntimeException e) {
                             ReferenceCountUtil.release(updateFrame);
                             promise.setFailure(e);
                             return;
                         }
                         flowControlledBytes -= updateFrame.windowSizeIncrement();
                         if (parentContext().isRemoved()) {
                             ReferenceCountUtil.release(msg);
                             promise.setFailure(new ClosedChannelException());
                             return;
                         }
                         ChannelFuture f = writeWindowUpdateFrame(updateFrame);
                         if (f.isDone()) {
                             writeComplete(f, promise);
                         } else {
                             f.addListener(new ChannelFutureListener() {
                                 @Override
                                 public void operationComplete(ChannelFuture future) {
                                     writeComplete(future, promise);
                                 }
                             });
                         }
                     } else {
                         writeHttp2StreamFrame(frame, promise);
                     }
                 } else {
                     String msgStr = msg.toString();
                     ReferenceCountUtil.release(msg);
                     promise.setFailure(new IllegalArgumentException(
                             "Message must be an " + StringUtil.simpleClassName(Http2StreamFrame.class) +
                                     ": " + msgStr));
                 }
             } catch (Throwable t) {
                 promise.tryFailure(t);
             }
         }
 
         private boolean isEndOfStream(Http2Frame frame) {
             if (frame instanceof Http2HeadersFrame) {
                 return ((Http2HeadersFrame) frame).isEndStream();
             }
             if (frame instanceof Http2DataFrame) {
                 return ((Http2DataFrame) frame).isEndStream();
             }
             return false;
         }
 
         private void writeHttp2StreamFrame(Http2StreamFrame frame, final ChannelPromise promise) {
             if (!firstFrameWritten && !isStreamIdValid(stream().id()) && !(frame instanceof Http2HeadersFrame)) {
                 ReferenceCountUtil.release(frame);
                 promise.setFailure(
                     new IllegalArgumentException("The first frame must be a headers frame. Was: "
                         + frame.name()));
                 return;
             }
 
             final boolean firstWrite;
             if (firstFrameWritten) {
                 firstWrite = false;
             } else {
                 firstWrite = firstFrameWritten = true;
             }
 
             // Let's keep track of what we send as the stream state itself will only be updated once the frame
             // was written which might cause problems if we try to close the channel in a write future.
             sentEndOfStream |= isEndOfStream(frame);
             ChannelFuture f = write0(parentContext(), frame);
             if (f.isDone()) {
                 if (firstWrite) {
                     firstWriteComplete(f, promise);
                 } else {
                     writeComplete(f, promise);
                 }
             } else {
                 final long bytes = FlowControlledFrameSizeEstimator.HANDLE_INSTANCE.size(frame);
                 incrementPendingOutboundBytes(bytes, false);
                 f.addListener(new ChannelFutureListener() {
                     @Override
                     public void operationComplete(ChannelFuture future) {
                         if (firstWrite) {
                             firstWriteComplete(future, promise);
                         } else {
                             writeComplete(future, promise);
                         }
                         decrementPendingOutboundBytes(bytes, false);
                     }
                 });
                 writeDoneAndNoFlush = true;
             }
         }
 
         private void firstWriteComplete(ChannelFuture future, ChannelPromise promise) {
             Throwable cause = future.cause();
             if (cause == null) {
                 promise.setSuccess();
             } else {
                 // If the first write fails there is not much we can do, just close
                 closeForcibly();
                 promise.setFailure(wrapStreamClosedError(cause));
             }
         }
 
         private void writeComplete(ChannelFuture future, ChannelPromise promise) {
             Throwable cause = future.cause();
             if (cause == null) {
                 promise.setSuccess();
             } else {
                 Throwable error = wrapStreamClosedError(cause);
                 // To make it more consistent with AbstractChannel we handle all IOExceptions here.
                 if (error instanceof IOException) {
                     if (config.isAutoClose()) {
                         // Close channel if needed.
                         closeForcibly();
                     } else {
                         // TODO: Once Http2StreamChannel extends DuplexChannel we should call shutdownOutput(...)
                         outboundClosed = true;
                     }
                 }
                 promise.setFailure(error);
             }
         }
 
         private Throwable wrapStreamClosedError(Throwable cause) {
             // If the error was caused by STREAM_CLOSED we should use a ClosedChannelException to better
             // mimic other transports and make it easier to reason about what exceptions to expect.
             if (cause instanceof Http2Exception && ((Http2Exception) cause).error() == Http2Error.STREAM_CLOSED) {
                 return new ClosedChannelException().initCause(cause);
             }
             return cause;
         }
 
         private Http2StreamFrame validateStreamFrame(Http2StreamFrame frame) {
             if (frame.stream() != null && frame.stream() != stream) {
                 String msgString = frame.toString();
                 ReferenceCountUtil.release(frame);
                 throw new IllegalArgumentException(
                         "Stream " + frame.stream() + " must not be set on the frame: " + msgString);
             }
             return frame;
         }
 
         @Override
         public void flush() {
             // If we are currently in the parent channel's read loop we should just ignore the flush.
             // We will ensure we trigger ctx.flush() after we processed all Channels later on and
             // so aggregate the flushes. This is done as ctx.flush() is expensive when as it may trigger an
             // write(...) or writev(...) operation on the socket.
             if (!writeDoneAndNoFlush || isParentReadInProgress()) {
                 // There is nothing to flush so this is a NOOP.
                 return;
             }
             // We need to set this to false before we call flush0(...) as ChannelFutureListener may produce more data
             // that are explicit flushed.
             writeDoneAndNoFlush = false;
             flush0(parentContext());
         }
 
         @Override
         public ChannelPromise voidPromise() {
             return unsafeVoidPromise;
         }
 
         @Override
         public ChannelOutboundBuffer outboundBuffer() {
             // Always return null as we not use the ChannelOutboundBuffer and not even support it.
             return null;
         }
     }
 
     /**
      * {@link ChannelConfig} so that the high and low writebuffer watermarks can reflect the outbound flow control
      * window, without having to create a new {@link WriteBufferWaterMark} object whenever the flow control window
      * changes.
      */
     private static final class Http2StreamChannelConfig extends DefaultChannelConfig {
 
         volatile boolean autoStreamFlowControl = true;
         Http2StreamChannelConfig(Channel channel) {
             super(channel);
         }
 
         @Override
         public MessageSizeEstimator getMessageSizeEstimator() {
             return FlowControlledFrameSizeEstimator.INSTANCE;
         }
 
         @Override
         public ChannelConfig setMessageSizeEstimator(MessageSizeEstimator estimator) {
             throw new UnsupportedOperationException();
         }
 
         @Override
         public ChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
             if (!(allocator.newHandle() instanceof RecvByteBufAllocator.ExtendedHandle)) {
                 throw new IllegalArgumentException("allocator.newHandle() must return an object of type: " +
                         RecvByteBufAllocator.ExtendedHandle.class);
             }
             super.setRecvByteBufAllocator(allocator);
             return this;
         }
 
         @Override
         public Map<ChannelOption<?>, Object> getOptions() {
             return getOptions(
                     super.getOptions(),
                     Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL);
         }
 
         @SuppressWarnings("unchecked")
         @Override
         public <T> T getOption(ChannelOption<T> option) {
             if (option == Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL) {
                 return (T) Boolean.valueOf(autoStreamFlowControl);
             }
             return super.getOption(option);
         }
 
         @Override
         public <T> boolean setOption(ChannelOption<T> option, T value) {
             validate(option, value);
             if (option == Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL) {
                 boolean newValue = (Boolean) value;
                 boolean changed = newValue && !autoStreamFlowControl;
                 autoStreamFlowControl = (Boolean) value;
                 if (changed) {
                     if (channel.isRegistered()) {
                         final Http2ChannelUnsafe unsafe = (Http2ChannelUnsafe) channel.unsafe();
                         if (channel.eventLoop().inEventLoop()) {
                             unsafe.updateLocalWindowIfNeededAndFlush();
                         } else {
                             channel.eventLoop().execute(new Runnable() {
                                 @Override
                                 public void run() {
                                     unsafe.updateLocalWindowIfNeededAndFlush();
                                 }
                             });
                         }
                     }
                 }
                 return true;
             }
             return super.setOption(option, value);
         }
     }
 
     private void maybeAddChannelToReadCompletePendingQueue() {
         if (!readCompletePending) {
             readCompletePending = true;
             addChannelToReadCompletePendingQueue();
         }
     }
 
     protected void flush0(ChannelHandlerContext ctx) {
         ctx.flush();
     }
 
     protected ChannelFuture write0(ChannelHandlerContext ctx, Object msg) {
         ChannelPromise promise = ctx.newPromise();
         ctx.write(msg, promise);
         return promise;
     }
 
     protected abstract boolean isParentReadInProgress();
     protected abstract void addChannelToReadCompletePendingQueue();
     protected abstract ChannelHandlerContext parentContext();
 }