diff --git a/src/main/java/io/grpc/netty/NettyClientHandler.java b/src/main/java/io/grpc/netty/NettyClientHandler.java
index 674dd6c033..55e0e1712b 100644
--- a/src/main/java/io/grpc/netty/NettyClientHandler.java
+++ b/src/main/java/io/grpc/netty/NettyClientHandler.java
@@ -137,21 +137,18 @@ class NettyClientHandler extends AbstractNettyHandler {
   public static final Histogram createStreamWriteHeaderDuration =
       Histogram.build()
           .name("grpc_netty_client_stream_write_header_duration_seconds")
-          .labelNames("path")
           .help("Time taken to write headers for a stream in seconds.")
           .register();
 
   public static final Histogram createStreamAddListenerDuration =
       Histogram.build()
           .name("grpc_netty_client_stream_add_listener_duration_seconds")
-          .labelNames("path")
           .help("Time taken to add listener for a stream future in seconds.")
           .register();
 
   public static final Histogram createStreamCreateNewFuture =
       Histogram.build()
           .name("grpc_netty_client_stream_create_future_duration_seconds")
-          .labelNames("path")
           .help("Time taken to create new stream future in seconds.")
           .register();
 
@@ -650,17 +647,17 @@ class NettyClientHandler extends AbstractNettyHandler {
     // Create an intermediate promise so that we can intercept the failure reported back to the
     // application.
     Histogram.Timer createFutureTimer =
-        createStreamCreateNewFuture.labels(headers.path().toString()).startTimer();
+        createStreamCreateNewFuture.startTimer();
     ChannelPromise tempPromise = ctx().newPromise();
     createFutureTimer.observeDuration();
 
     Histogram.Timer writeHeaderTimer =
-        createStreamWriteHeaderDuration.labels(headers.path().toString()).startTimer();
+        createStreamWriteHeaderDuration.startTimer();
     ChannelFuture future = encoder().writeHeaders(ctx(), streamId, headers, 0, isGet, tempPromise);
     writeHeaderTimer.observeDuration();
 
     Histogram.Timer addListenerTimer =
-        createStreamAddListenerDuration.labels(headers.path().toString()).startTimer();
+        createStreamAddListenerDuration.startTimer();
     future.addListener(
         new ChannelFutureListener() {
           @Override
diff --git a/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java b/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java
new file mode 100644
index 0000000000..1f35c059b2
--- /dev/null
+++ b/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java
@@ -0,0 +1,787 @@
+/*
+ * Copyright 2014 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License, version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at:
+ *
+ * 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.
+ */
+package io.netty.handler.codec.http2;
+
+import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_WINDOW_SIZE;
+import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_WEIGHT;
+import static io.netty.handler.codec.http2.Http2CodecUtil.MIN_WEIGHT;
+import static io.netty.handler.codec.http2.Http2Error.FLOW_CONTROL_ERROR;
+import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
+import static io.netty.handler.codec.http2.Http2Error.STREAM_CLOSED;
+import static io.netty.handler.codec.http2.Http2Exception.streamError;
+import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
+import static io.netty.util.internal.ObjectUtil.checkNotNull;
+import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
+import static java.lang.Math.max;
+import static java.lang.Math.min;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.internal.UnstableApi;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+import io.prometheus.client.Histogram;
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+/**
+ * Basic implementation of {@link Http2RemoteFlowController}.
+ *
+ * <p>This class is <strong>NOT</strong> thread safe. The assumption is all methods must be invoked
+ * from a single thread. Typically this thread is the event loop thread for the {@link
+ * ChannelHandlerContext} managed by this class.
+ */
+@UnstableApi
+public class DefaultHttp2RemoteFlowController implements Http2RemoteFlowController {
+  private static final InternalLogger logger =
+      InternalLoggerFactory.getInstance(DefaultHttp2RemoteFlowController.class);
+  private static final int MIN_WRITABLE_CHUNK = 32 * 1024;
+  private final Http2Connection connection;
+  private final Http2Connection.PropertyKey stateKey;
+  private final StreamByteDistributor streamByteDistributor;
+  private final FlowState connectionState;
+  private int initialWindowSize = DEFAULT_WINDOW_SIZE;
+  private WritabilityMonitor monitor;
+  private ChannelHandlerContext ctx;
+
+  public static final Histogram byteDistributedDuration =
+      Histogram.build()
+          .name("http2_byte_distributed_duration_seconds")
+          .help("The duration of byte distributed to streams.")
+          .register();
+
+  public DefaultHttp2RemoteFlowController(Http2Connection connection) {
+    this(connection, (Listener) null);
+  }
+
+  public DefaultHttp2RemoteFlowController(
+      Http2Connection connection, StreamByteDistributor streamByteDistributor) {
+    this(connection, streamByteDistributor, null);
+  }
+
+  public DefaultHttp2RemoteFlowController(Http2Connection connection, final Listener listener) {
+    this(connection, new WeightedFairQueueByteDistributor(connection), listener);
+  }
+
+  public DefaultHttp2RemoteFlowController(
+      Http2Connection connection,
+      StreamByteDistributor streamByteDistributor,
+      final Listener listener) {
+    this.connection = checkNotNull(connection, "connection");
+    this.streamByteDistributor = checkNotNull(streamByteDistributor, "streamWriteDistributor");
+
+    // Add a flow state for the connection.
+    stateKey = connection.newKey();
+    connectionState = new FlowState(connection.connectionStream());
+    connection.connectionStream().setProperty(stateKey, connectionState);
+
+    // Monitor may depend upon connectionState, and so initialize after connectionState
+    listener(listener);
+    monitor.windowSize(connectionState, initialWindowSize);
+
+    // Register for notification of new streams.
+    connection.addListener(
+        new Http2ConnectionAdapter() {
+          @Override
+          public void onStreamAdded(Http2Stream stream) {
+            // If the stream state is not open then the stream is not yet eligible for flow
+            // controlled frames and
+            // only requires the ReducedFlowState. Otherwise the full amount of memory is required.
+            stream.setProperty(stateKey, new FlowState(stream));
+          }
+
+          @Override
+          public void onStreamActive(Http2Stream stream) {
+            // If the object was previously created, but later activated then we have to ensure the
+            // proper
+            // initialWindowSize is used.
+            monitor.windowSize(state(stream), initialWindowSize);
+          }
+
+          @Override
+          public void onStreamClosed(Http2Stream stream) {
+            // Any pending frames can never be written, cancel and
+            // write errors for any pending frames.
+            state(stream).cancel(STREAM_CLOSED, null);
+          }
+
+          @Override
+          public void onStreamHalfClosed(Http2Stream stream) {
+            if (HALF_CLOSED_LOCAL == stream.state()) {
+              /**
+               * When this method is called there should not be any pending frames left if the API
+               * is used correctly. However, it is possible that a erroneous application can sneak
+               * in a frame even after having already written a frame with the END_STREAM flag set,
+               * as the stream state might not transition immediately to HALF_CLOSED_LOCAL / CLOSED
+               * due to flow control delaying the write.
+               *
+               * <p>This is to cancel any such illegal writes.
+               */
+              state(stream).cancel(STREAM_CLOSED, null);
+            }
+          }
+        });
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * <p>Any queued {@link FlowControlled} objects will be sent.
+   */
+  @Override
+  public void channelHandlerContext(ChannelHandlerContext ctx) throws Http2Exception {
+    this.ctx = checkNotNull(ctx, "ctx");
+
+    // Writing the pending bytes will not check writability change and instead a writability change
+    // notification
+    // to be provided by an explicit call.
+    channelWritabilityChanged();
+
+    // Don't worry about cleaning up queued frames here if ctx is null. It is expected that all
+    // streams will be
+    // closed and the queue cleanup will occur when the stream state transitions occur.
+
+    // If any frames have been queued up, we should send them now that we have a channel context.
+    if (isChannelWritable()) {
+      writePendingBytes();
+    }
+  }
+
+  @Override
+  public ChannelHandlerContext channelHandlerContext() {
+    return ctx;
+  }
+
+  @Override
+  public void initialWindowSize(int newWindowSize) throws Http2Exception {
+    assert ctx == null || ctx.executor().inEventLoop();
+    monitor.initialWindowSize(newWindowSize);
+  }
+
+  @Override
+  public int initialWindowSize() {
+    return initialWindowSize;
+  }
+
+  @Override
+  public int windowSize(Http2Stream stream) {
+    return state(stream).windowSize();
+  }
+
+  @Override
+  public boolean isWritable(Http2Stream stream) {
+    return monitor.isWritable(state(stream));
+  }
+
+  @Override
+  public void channelWritabilityChanged() throws Http2Exception {
+    monitor.channelWritabilityChange();
+  }
+
+  @Override
+  public void updateDependencyTree(
+      int childStreamId, int parentStreamId, short weight, boolean exclusive) {
+    // It is assumed there are all validated at a higher level. For example in the Http2FrameReader.
+    assert weight >= MIN_WEIGHT && weight <= MAX_WEIGHT : "Invalid weight";
+    assert childStreamId != parentStreamId : "A stream cannot depend on itself";
+    assert childStreamId > 0 && parentStreamId >= 0
+        : "childStreamId must be > 0. parentStreamId must be >= 0.";
+
+    streamByteDistributor.updateDependencyTree(childStreamId, parentStreamId, weight, exclusive);
+  }
+
+  private boolean isChannelWritable() {
+    return ctx != null && isChannelWritable0();
+  }
+
+  private boolean isChannelWritable0() {
+    return ctx.channel().isWritable();
+  }
+
+  @Override
+  public void listener(Listener listener) {
+    monitor =
+        listener == null ? new WritabilityMonitor() : new ListenerWritabilityMonitor(listener);
+  }
+
+  @Override
+  public void incrementWindowSize(Http2Stream stream, int delta) throws Http2Exception {
+    assert ctx == null || ctx.executor().inEventLoop();
+    monitor.incrementWindowSize(state(stream), delta);
+  }
+
+  @Override
+  public void addFlowControlled(Http2Stream stream, FlowControlled frame) {
+    // The context can be null assuming the frame will be queued and send later when the context is
+    // set.
+    assert ctx == null || ctx.executor().inEventLoop();
+    checkNotNull(frame, "frame");
+    try {
+      monitor.enqueueFrame(state(stream), frame);
+    } catch (Throwable t) {
+      frame.error(ctx, t);
+    }
+  }
+
+  @Override
+  public boolean hasFlowControlled(Http2Stream stream) {
+    return state(stream).hasFrame();
+  }
+
+  private FlowState state(Http2Stream stream) {
+    return (FlowState) stream.getProperty(stateKey);
+  }
+
+  /** Returns the flow control window for the entire connection. */
+  private int connectionWindowSize() {
+    return connectionState.windowSize();
+  }
+
+  private int minUsableChannelBytes() {
+    // The current allocation algorithm values "fairness" and doesn't give any consideration to
+    // "goodput". It
+    // is possible that 1 byte will be allocated to many streams. In an effort to try to make
+    // "goodput"
+    // reasonable with the current allocation algorithm we have this "cheap" check up front to
+    // ensure there is
+    // an "adequate" amount of connection window before allocation is attempted. This is not
+    // foolproof as if the
+    // number of streams is >= this minimal number then we may still have the issue, but the idea is
+    // to narrow the
+    // circumstances in which this can happen without rewriting the allocation algorithm.
+    return max(ctx.channel().config().getWriteBufferLowWaterMark(), MIN_WRITABLE_CHUNK);
+  }
+
+  private int maxUsableChannelBytes() {
+    // If the channel isWritable, allow at least minUsableChannelBytes.
+    int channelWritableBytes = (int) min(Integer.MAX_VALUE, ctx.channel().bytesBeforeUnwritable());
+    int usableBytes =
+        channelWritableBytes > 0 ? max(channelWritableBytes, minUsableChannelBytes()) : 0;
+
+    // Clip the usable bytes by the connection window.
+    return min(connectionState.windowSize(), usableBytes);
+  }
+
+  /**
+   * The amount of bytes that can be supported by underlying {@link io.netty.channel.Channel}
+   * without queuing "too-much".
+   */
+  private int writableBytes() {
+    return min(connectionWindowSize(), maxUsableChannelBytes());
+  }
+
+  @Override
+  public void writePendingBytes() throws Http2Exception {
+    monitor.writePendingBytes();
+  }
+
+  /** The remote flow control state for a single stream. */
+  private final class FlowState implements StreamByteDistributor.StreamState {
+    private final Http2Stream stream;
+    private final Deque<FlowControlled> pendingWriteQueue;
+    private int window;
+    private long pendingBytes;
+    private boolean markedWritable;
+
+    /** Set to true while a frame is being written, false otherwise. */
+    private boolean writing;
+    /** Set to true if cancel() was called. */
+    private boolean cancelled;
+
+    FlowState(Http2Stream stream) {
+      this.stream = stream;
+      pendingWriteQueue = new ArrayDeque<FlowControlled>(2);
+    }
+
+    /**
+     * Determine if the stream associated with this object is writable.
+     *
+     * @return {@code true} if the stream associated with this object is writable.
+     */
+    boolean isWritable() {
+      return windowSize() > pendingBytes() && !cancelled;
+    }
+
+    /** The stream this state is associated with. */
+    @Override
+    public Http2Stream stream() {
+      return stream;
+    }
+
+    /** Returns the parameter from the last call to {@link #markedWritability(boolean)}. */
+    boolean markedWritability() {
+      return markedWritable;
+    }
+
+    /** Save the state of writability. */
+    void markedWritability(boolean isWritable) {
+      this.markedWritable = isWritable;
+    }
+
+    @Override
+    public int windowSize() {
+      return window;
+    }
+
+    /** Reset the window size for this stream. */
+    void windowSize(int initialWindowSize) {
+      window = initialWindowSize;
+    }
+
+    /**
+     * Write the allocated bytes for this stream.
+     *
+     * @return the number of bytes written for a stream or {@code -1} if no write occurred.
+     */
+    int writeAllocatedBytes(int allocated) {
+      final int initialAllocated = allocated;
+      int writtenBytes;
+      // In case an exception is thrown we want to remember it and pass it to cancel(Throwable).
+      Throwable cause = null;
+      FlowControlled frame;
+      try {
+        assert !writing;
+        writing = true;
+
+        // Write the remainder of frames that we are allowed to
+        boolean writeOccurred = false;
+        while (!cancelled && (frame = peek()) != null) {
+          int maxBytes = min(allocated, writableWindow());
+          if (maxBytes <= 0 && frame.size() > 0) {
+            // The frame still has data, but the amount of allocated bytes has been exhausted.
+            // Don't write needless empty frames.
+            break;
+          }
+          writeOccurred = true;
+          int initialFrameSize = frame.size();
+          try {
+            frame.write(ctx, max(0, maxBytes));
+            if (frame.size() == 0) {
+              // This frame has been fully written, remove this frame and notify it.
+              // Since we remove this frame first, we're guaranteed that its error
+              // method will not be called when we call cancel.
+              pendingWriteQueue.remove();
+              frame.writeComplete();
+            }
+          } finally {
+            // Decrement allocated by how much was actually written.
+            allocated -= initialFrameSize - frame.size();
+          }
+        }
+
+        if (!writeOccurred) {
+          // Either there was no frame, or the amount of allocated bytes has been exhausted.
+          return -1;
+        }
+
+      } catch (Throwable t) {
+        // Mark the state as cancelled, we'll clear the pending queue via cancel() below.
+        cancelled = true;
+        cause = t;
+      } finally {
+        writing = false;
+        // Make sure we always decrement the flow control windows
+        // by the bytes written.
+        writtenBytes = initialAllocated - allocated;
+
+        decrementPendingBytes(writtenBytes, false);
+        decrementFlowControlWindow(writtenBytes);
+
+        // If a cancellation occurred while writing, call cancel again to
+        // clear and error all of the pending writes.
+        if (cancelled) {
+          cancel(INTERNAL_ERROR, cause);
+        }
+      }
+      return writtenBytes;
+    }
+
+    /**
+     * Increments the flow control window for this stream by the given delta and returns the new
+     * value.
+     */
+    int incrementStreamWindow(int delta) throws Http2Exception {
+      if (delta > 0 && Integer.MAX_VALUE - delta < window) {
+        throw streamError(
+            stream.id(), FLOW_CONTROL_ERROR, "Window size overflow for stream: %d", stream.id());
+      }
+      window += delta;
+
+      streamByteDistributor.updateStreamableBytes(this);
+      return window;
+    }
+
+    /** Returns the maximum writable window (minimum of the stream and connection windows). */
+    private int writableWindow() {
+      return min(window, connectionWindowSize());
+    }
+
+    @Override
+    public long pendingBytes() {
+      return pendingBytes;
+    }
+
+    /** Adds the {@code frame} to the pending queue and increments the pending byte count. */
+    void enqueueFrame(FlowControlled frame) {
+      FlowControlled last = pendingWriteQueue.peekLast();
+      if (last == null) {
+        enqueueFrameWithoutMerge(frame);
+        return;
+      }
+
+      int lastSize = last.size();
+      if (last.merge(ctx, frame)) {
+        incrementPendingBytes(last.size() - lastSize, true);
+        return;
+      }
+      enqueueFrameWithoutMerge(frame);
+    }
+
+    private void enqueueFrameWithoutMerge(FlowControlled frame) {
+      pendingWriteQueue.offer(frame);
+      // This must be called after adding to the queue in order so that hasFrame() is
+      // updated before updating the stream state.
+      incrementPendingBytes(frame.size(), true);
+    }
+
+    @Override
+    public boolean hasFrame() {
+      return !pendingWriteQueue.isEmpty();
+    }
+
+    /** Returns the head of the pending queue, or {@code null} if empty. */
+    private FlowControlled peek() {
+      return pendingWriteQueue.peek();
+    }
+
+    /**
+     * Clears the pending queue and writes errors for each remaining frame.
+     *
+     * @param error the {@link Http2Error} to use.
+     * @param cause the {@link Throwable} that caused this method to be invoked.
+     */
+    void cancel(Http2Error error, Throwable cause) {
+      cancelled = true;
+      // Ensure that the queue can't be modified while we are writing.
+      if (writing) {
+        return;
+      }
+
+      FlowControlled frame = pendingWriteQueue.poll();
+      if (frame != null) {
+        // Only create exception once and reuse to reduce overhead of filling in the stacktrace.
+        final Http2Exception exception =
+            streamError(stream.id(), error, cause, "Stream closed before write could take place");
+        do {
+          writeError(frame, exception);
+          frame = pendingWriteQueue.poll();
+        } while (frame != null);
+      }
+
+      streamByteDistributor.updateStreamableBytes(this);
+
+      monitor.stateCancelled(this);
+    }
+
+    /**
+     * Increments the number of pending bytes for this node and optionally updates the {@link
+     * StreamByteDistributor}.
+     */
+    private void incrementPendingBytes(int numBytes, boolean updateStreamableBytes) {
+      pendingBytes += numBytes;
+      monitor.incrementPendingBytes(numBytes);
+      if (updateStreamableBytes) {
+        streamByteDistributor.updateStreamableBytes(this);
+      }
+    }
+
+    /**
+     * If this frame is in the pending queue, decrements the number of pending bytes for the stream.
+     */
+    private void decrementPendingBytes(int bytes, boolean updateStreamableBytes) {
+      incrementPendingBytes(-bytes, updateStreamableBytes);
+    }
+
+    /** Decrement the per stream and connection flow control window by {@code bytes}. */
+    private void decrementFlowControlWindow(int bytes) {
+      try {
+        int negativeBytes = -bytes;
+        connectionState.incrementStreamWindow(negativeBytes);
+        incrementStreamWindow(negativeBytes);
+      } catch (Http2Exception e) {
+        // Should never get here since we're decrementing.
+        throw new IllegalStateException(
+            "Invalid window state when writing frame: " + e.getMessage(), e);
+      }
+    }
+
+    /**
+     * Discards this {@link FlowControlled}, writing an error. If this frame is in the pending
+     * queue, the unwritten bytes are removed from this branch of the priority tree.
+     */
+    private void writeError(FlowControlled frame, Http2Exception cause) {
+      assert ctx != null;
+      decrementPendingBytes(frame.size(), true);
+      frame.error(ctx, cause);
+    }
+  }
+
+  /** Abstract class which provides common functionality for writability monitor implementations. */
+  private class WritabilityMonitor implements StreamByteDistributor.Writer {
+    private boolean inWritePendingBytes;
+    private long totalPendingBytes;
+
+    @Override
+    public final void write(Http2Stream stream, int numBytes) {
+      state(stream).writeAllocatedBytes(numBytes);
+    }
+
+    /**
+     * Called when the writability of the underlying channel changes.
+     *
+     * @throws Http2Exception If a write occurs and an exception happens in the write operation.
+     */
+    void channelWritabilityChange() throws Http2Exception {}
+
+    /**
+     * Called when the state is cancelled.
+     *
+     * @param state the state that was cancelled.
+     */
+    void stateCancelled(FlowState state) {}
+
+    /**
+     * Set the initial window size for {@code state}.
+     *
+     * @param state the state to change the initial window size for.
+     * @param initialWindowSize the size of the window in bytes.
+     */
+    void windowSize(FlowState state, int initialWindowSize) {
+      state.windowSize(initialWindowSize);
+    }
+
+    /**
+     * Increment the window size for a particular stream.
+     *
+     * @param state the state associated with the stream whose window is being incremented.
+     * @param delta The amount to increment by.
+     * @throws Http2Exception If this operation overflows the window for {@code state}.
+     */
+    void incrementWindowSize(FlowState state, int delta) throws Http2Exception {
+      state.incrementStreamWindow(delta);
+    }
+
+    /**
+     * Add a frame to be sent via flow control.
+     *
+     * @param state The state associated with the stream which the {@code frame} is associated with.
+     * @param frame the frame to enqueue.
+     * @throws Http2Exception If a writability error occurs.
+     */
+    void enqueueFrame(FlowState state, FlowControlled frame) throws Http2Exception {
+      state.enqueueFrame(frame);
+    }
+
+    /**
+     * Increment the total amount of pending bytes for all streams. When any stream's pending bytes
+     * changes method should be called.
+     *
+     * @param delta The amount to increment by.
+     */
+    final void incrementPendingBytes(int delta) {
+      totalPendingBytes += delta;
+
+      // Notification of writibilty change should be delayed until the end of the top level event.
+      // This is to ensure the flow controller is more consistent state before calling external
+      // listener methods.
+    }
+
+    /**
+     * Determine if the stream associated with {@code state} is writable.
+     *
+     * @param state The state which is associated with the stream to test writability for.
+     * @return {@code true} if {@link FlowState#stream()} is writable. {@code false} otherwise.
+     */
+    final boolean isWritable(FlowState state) {
+      return isWritableConnection() && state.isWritable();
+    }
+
+    final void writePendingBytes() throws Http2Exception {
+      // Reentry is not permitted during the byte distribution process. It may lead to undesirable
+      // distribution of
+      // bytes and even infinite loops. We protect against reentry and make sure each call has an
+      // opportunity to
+      // cause a distribution to occur. This may be useful for example if the channel's writability
+      // changes from
+      // Writable -> Not Writable (because we are writing) -> Writable (because the user flushed to
+      // make more room
+      // in the channel outbound buffer).
+      if (inWritePendingBytes) {
+        return;
+      }
+      inWritePendingBytes = true;
+      try {
+        int bytesToWrite = writableBytes();
+        // Make sure we always write at least once, regardless if we have bytesToWrite or not.
+        // This ensures that zero-length frames will always be written.
+        for (; ; ) {
+          Histogram.Timer distributedTimer = byteDistributedDuration.startTimer();
+          boolean distributed = streamByteDistributor.distribute(bytesToWrite, this);
+          distributedTimer.observeDuration();
+          if (!distributed || (bytesToWrite = writableBytes()) <= 0 || !isChannelWritable0()) {
+            break;
+          }
+        }
+      } finally {
+        inWritePendingBytes = false;
+      }
+    }
+
+    void initialWindowSize(int newWindowSize) throws Http2Exception {
+      checkPositiveOrZero(newWindowSize, "newWindowSize");
+
+      final int delta = newWindowSize - initialWindowSize;
+      initialWindowSize = newWindowSize;
+      connection.forEachActiveStream(
+          new Http2StreamVisitor() {
+            @Override
+            public boolean visit(Http2Stream stream) throws Http2Exception {
+              state(stream).incrementStreamWindow(delta);
+              return true;
+            }
+          });
+
+      if (delta > 0 && isChannelWritable()) {
+        // The window size increased, send any pending frames for all streams.
+        writePendingBytes();
+      }
+    }
+
+    final boolean isWritableConnection() {
+      return connectionState.windowSize() - totalPendingBytes > 0 && isChannelWritable();
+    }
+  }
+
+  /**
+   * Writability of a {@code stream} is calculated using the following:
+   *
+   * <pre>
+   * Connection Window - Total Queued Bytes > 0 &&
+   * Stream Window - Bytes Queued for Stream > 0 &&
+   * isChannelWritable()
+   * </pre>
+   */
+  private final class ListenerWritabilityMonitor extends WritabilityMonitor
+      implements Http2StreamVisitor {
+    private final Listener listener;
+
+    ListenerWritabilityMonitor(Listener listener) {
+      this.listener = listener;
+    }
+
+    @Override
+    public boolean visit(Http2Stream stream) throws Http2Exception {
+      FlowState state = state(stream);
+      if (isWritable(state) != state.markedWritability()) {
+        notifyWritabilityChanged(state);
+      }
+      return true;
+    }
+
+    @Override
+    void windowSize(FlowState state, int initialWindowSize) {
+      super.windowSize(state, initialWindowSize);
+      try {
+        checkStateWritability(state);
+      } catch (Http2Exception e) {
+        throw new RuntimeException("Caught unexpected exception from window", e);
+      }
+    }
+
+    @Override
+    void incrementWindowSize(FlowState state, int delta) throws Http2Exception {
+      super.incrementWindowSize(state, delta);
+      checkStateWritability(state);
+    }
+
+    @Override
+    void initialWindowSize(int newWindowSize) throws Http2Exception {
+      super.initialWindowSize(newWindowSize);
+      if (isWritableConnection()) {
+        // If the write operation does not occur we still need to check all streams because they
+        // may have transitioned from writable to not writable.
+        checkAllWritabilityChanged();
+      }
+    }
+
+    @Override
+    void enqueueFrame(FlowState state, FlowControlled frame) throws Http2Exception {
+      super.enqueueFrame(state, frame);
+      checkConnectionThenStreamWritabilityChanged(state);
+    }
+
+    @Override
+    void stateCancelled(FlowState state) {
+      try {
+        checkConnectionThenStreamWritabilityChanged(state);
+      } catch (Http2Exception e) {
+        throw new RuntimeException(
+            "Caught unexpected exception from checkAllWritabilityChanged", e);
+      }
+    }
+
+    @Override
+    void channelWritabilityChange() throws Http2Exception {
+      if (connectionState.markedWritability() != isChannelWritable()) {
+        checkAllWritabilityChanged();
+      }
+    }
+
+    private void checkStateWritability(FlowState state) throws Http2Exception {
+      if (isWritable(state) != state.markedWritability()) {
+        if (state == connectionState) {
+          checkAllWritabilityChanged();
+        } else {
+          notifyWritabilityChanged(state);
+        }
+      }
+    }
+
+    private void notifyWritabilityChanged(FlowState state) {
+      state.markedWritability(!state.markedWritability());
+      try {
+        listener.writabilityChanged(state.stream);
+      } catch (Throwable cause) {
+        logger.error("Caught Throwable from listener.writabilityChanged", cause);
+      }
+    }
+
+    private void checkConnectionThenStreamWritabilityChanged(FlowState state)
+        throws Http2Exception {
+      // It is possible that the connection window and/or the individual stream writability could
+      // change.
+      if (isWritableConnection() != connectionState.markedWritability()) {
+        checkAllWritabilityChanged();
+      } else if (isWritable(state) != state.markedWritability()) {
+        notifyWritabilityChanged(state);
+      }
+    }
+
+    private void checkAllWritabilityChanged() throws Http2Exception {
+      // Make sure we mark that we have notified as a result of this change.
+      connectionState.markedWritability(isWritableConnection());
+      connection.forEachActiveStream(this);
+    }
+  }
+}
diff --git a/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java b/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java
new file mode 100644
index 0000000000..f9089dc62e
--- /dev/null
+++ b/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java
@@ -0,0 +1,1110 @@
+/*
+ * Copyright 2014 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License, version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at:
+ *
+ * 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.
+ */
+package io.netty.handler.codec.http2;
+
+import static io.netty.buffer.ByteBufUtil.hexDump;
+import static io.netty.buffer.Unpooled.EMPTY_BUFFER;
+import static io.netty.handler.codec.http2.Http2CodecUtil.HTTP_UPGRADE_STREAM_ID;
+import static io.netty.handler.codec.http2.Http2CodecUtil.connectionPrefaceBuf;
+import static io.netty.handler.codec.http2.Http2CodecUtil.getEmbeddedHttp2Exception;
+import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
+import static io.netty.handler.codec.http2.Http2Error.NO_ERROR;
+import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
+import static io.netty.handler.codec.http2.Http2Exception.connectionError;
+import static io.netty.handler.codec.http2.Http2Exception.isStreamError;
+import static io.netty.handler.codec.http2.Http2FrameTypes.SETTINGS;
+import static io.netty.handler.codec.http2.Http2Stream.State.IDLE;
+import static io.netty.util.CharsetUtil.UTF_8;
+import static io.netty.util.internal.ObjectUtil.checkNotNull;
+import static java.lang.Math.min;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandler;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http2.Http2Exception.CompositeStreamException;
+import io.netty.handler.codec.http2.Http2Exception.StreamException;
+import io.netty.util.CharsetUtil;
+import io.netty.util.concurrent.ScheduledFuture;
+import io.netty.util.internal.UnstableApi;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+import io.prometheus.client.Histogram;
+import java.net.SocketAddress;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Provides the default implementation for processing inbound frame events and delegates to a {@link
+ * Http2FrameListener}
+ *
+ * <p>This class will read HTTP/2 frames and delegate the events to a {@link Http2FrameListener}
+ *
+ * <p>This interface enforces inbound flow control functionality through {@link
+ * Http2LocalFlowController}
+ */
+@UnstableApi
+public class Http2ConnectionHandler extends ByteToMessageDecoder
+    implements Http2LifecycleManager, ChannelOutboundHandler {
+
+  private static final InternalLogger logger =
+      InternalLoggerFactory.getInstance(Http2ConnectionHandler.class);
+
+  private static final Http2Headers HEADERS_TOO_LARGE_HEADERS =
+      ReadOnlyHttp2Headers.serverHeaders(
+          false, HttpResponseStatus.REQUEST_HEADER_FIELDS_TOO_LARGE.codeAsText());
+  private static final ByteBuf HTTP_1_X_BUF =
+      Unpooled.unreleasableBuffer(
+              Unpooled.wrappedBuffer(new byte[] {'H', 'T', 'T', 'P', '/', '1', '.'}))
+          .asReadOnly();
+
+  private final Http2ConnectionDecoder decoder;
+  private final Http2ConnectionEncoder encoder;
+  private final Http2Settings initialSettings;
+  private final boolean decoupleCloseAndGoAway;
+  private ChannelFutureListener closeListener;
+  private BaseDecoder byteDecoder;
+  private long gracefulShutdownTimeoutMillis;
+
+  public static final Histogram flushFlowControlWriteDuration =
+      Histogram.build()
+          .name("netty_http2_flush_flow_control_write_duration_seconds")
+          .help("The time it takes to flush the pending bytes via flow control in seconds.")
+          .register();
+
+  public static final Histogram flushCtxFlushDuration =
+      Histogram.build()
+          .name("netty_http2_flush_ctx_flush_duration_seconds")
+          .help("The time it takes to ctx flush in seconds.")
+          .register();
+
+  protected Http2ConnectionHandler(
+      Http2ConnectionDecoder decoder,
+      Http2ConnectionEncoder encoder,
+      Http2Settings initialSettings) {
+    this(decoder, encoder, initialSettings, false);
+  }
+
+  protected Http2ConnectionHandler(
+      Http2ConnectionDecoder decoder,
+      Http2ConnectionEncoder encoder,
+      Http2Settings initialSettings,
+      boolean decoupleCloseAndGoAway) {
+    this.initialSettings = checkNotNull(initialSettings, "initialSettings");
+    this.decoder = checkNotNull(decoder, "decoder");
+    this.encoder = checkNotNull(encoder, "encoder");
+    this.decoupleCloseAndGoAway = decoupleCloseAndGoAway;
+    if (encoder.connection() != decoder.connection()) {
+      throw new IllegalArgumentException(
+          "Encoder and Decoder do not share the same connection object");
+    }
+  }
+
+  /**
+   * Get the amount of time (in milliseconds) this endpoint will wait for all streams to be closed
+   * before closing the connection during the graceful shutdown process. Returns -1 if this
+   * connection is configured to wait indefinitely for all streams to close.
+   */
+  public long gracefulShutdownTimeoutMillis() {
+    return gracefulShutdownTimeoutMillis;
+  }
+
+  /**
+   * Set the amount of time (in milliseconds) this endpoint will wait for all streams to be closed
+   * before closing the connection during the graceful shutdown process.
+   *
+   * @param gracefulShutdownTimeoutMillis the amount of time (in milliseconds) this endpoint will
+   *     wait for all streams to be closed before closing the connection during the graceful
+   *     shutdown process.
+   */
+  public void gracefulShutdownTimeoutMillis(long gracefulShutdownTimeoutMillis) {
+    if (gracefulShutdownTimeoutMillis < -1) {
+      throw new IllegalArgumentException(
+          "gracefulShutdownTimeoutMillis: "
+              + gracefulShutdownTimeoutMillis
+              + " (expected: -1 for indefinite or >= 0)");
+    }
+    this.gracefulShutdownTimeoutMillis = gracefulShutdownTimeoutMillis;
+  }
+
+  public Http2Connection connection() {
+    return encoder.connection();
+  }
+
+  public Http2ConnectionDecoder decoder() {
+    return decoder;
+  }
+
+  public Http2ConnectionEncoder encoder() {
+    return encoder;
+  }
+
+  private boolean prefaceSent() {
+    return byteDecoder != null && byteDecoder.prefaceSent();
+  }
+
+  /**
+   * Handles the client-side (cleartext) upgrade from HTTP to HTTP/2. Reserves local stream 1 for
+   * the HTTP/2 response.
+   */
+  public void onHttpClientUpgrade() throws Http2Exception {
+    if (connection().isServer()) {
+      throw connectionError(PROTOCOL_ERROR, "Client-side HTTP upgrade requested for a server");
+    }
+    if (!prefaceSent()) {
+      // If the preface was not sent yet it most likely means the handler was not added to the
+      // pipeline before
+      // calling this method.
+      throw connectionError(INTERNAL_ERROR, "HTTP upgrade must occur after preface was sent");
+    }
+    if (decoder.prefaceReceived()) {
+      throw connectionError(
+          PROTOCOL_ERROR, "HTTP upgrade must occur before HTTP/2 preface is received");
+    }
+
+    // Create a local stream used for the HTTP cleartext upgrade.
+    connection().local().createStream(HTTP_UPGRADE_STREAM_ID, true);
+  }
+
+  /**
+   * Handles the server-side (cleartext) upgrade from HTTP to HTTP/2.
+   *
+   * @param settings the settings for the remote endpoint.
+   */
+  public void onHttpServerUpgrade(Http2Settings settings) throws Http2Exception {
+    if (!connection().isServer()) {
+      throw connectionError(PROTOCOL_ERROR, "Server-side HTTP upgrade requested for a client");
+    }
+    if (!prefaceSent()) {
+      // If the preface was not sent yet it most likely means the handler was not added to the
+      // pipeline before
+      // calling this method.
+      throw connectionError(INTERNAL_ERROR, "HTTP upgrade must occur after preface was sent");
+    }
+    if (decoder.prefaceReceived()) {
+      throw connectionError(
+          PROTOCOL_ERROR, "HTTP upgrade must occur before HTTP/2 preface is received");
+    }
+
+    // Apply the settings but no ACK is necessary.
+    encoder.remoteSettings(settings);
+
+    // Create a stream in the half-closed state.
+    connection().remote().createStream(HTTP_UPGRADE_STREAM_ID, true);
+  }
+
+  @Override
+  public void flush(ChannelHandlerContext ctx) {
+    Histogram.Timer writeTimer = flushFlowControlWriteDuration.startTimer();
+    Histogram.Timer flushTimer = flushCtxFlushDuration.startTimer();
+    boolean writeSuccess = false;
+    boolean flushSuccess = false;
+    try {
+      // Trigger pending writes in the remote flow controller.
+      encoder.flowController().writePendingBytes();
+      writeTimer.observeDuration();
+      writeSuccess = true;
+
+      ctx.flush();
+      flushSuccess = true;
+      flushTimer.observeDuration();
+    } catch (Http2Exception e) {
+      onError(ctx, true, e);
+    } catch (Throwable cause) {
+      onError(ctx, true, connectionError(INTERNAL_ERROR, cause, "Error flushing"));
+    } finally {
+      if (!writeSuccess) {
+        writeTimer.observeDuration();
+      }
+      if (!flushSuccess) {
+        flushTimer.observeDuration();
+      }
+    }
+  }
+
+  private abstract class BaseDecoder {
+    public abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
+        throws Exception;
+
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {}
+
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {}
+
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+      // Connection has terminated, close the encoder and decoder.
+      encoder().close();
+      decoder().close();
+
+      // We need to remove all streams (not just the active ones).
+      // See https://github.com/netty/netty/issues/4838.
+      connection().close(ctx.voidPromise());
+    }
+
+    /** Determine if the HTTP/2 connection preface been sent. */
+    public boolean prefaceSent() {
+      return true;
+    }
+  }
+
+  private final class PrefaceDecoder extends BaseDecoder {
+    private ByteBuf clientPrefaceString;
+    private boolean prefaceSent;
+
+    PrefaceDecoder(ChannelHandlerContext ctx) throws Exception {
+      clientPrefaceString = clientPrefaceString(encoder.connection());
+      // This handler was just added to the context. In case it was handled after
+      // the connection became active, send the connection preface now.
+      sendPreface(ctx);
+    }
+
+    @Override
+    public boolean prefaceSent() {
+      return prefaceSent;
+    }
+
+    @Override
+    public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+      try {
+        if (ctx.channel().isActive()
+            && readClientPrefaceString(in)
+            && verifyFirstFrameIsSettings(in)) {
+          // After the preface is read, it is time to hand over control to the post initialized
+          // decoder.
+          byteDecoder = new FrameDecoder();
+          byteDecoder.decode(ctx, in, out);
+        }
+      } catch (Throwable e) {
+        onError(ctx, false, e);
+      }
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+      // The channel just became active - send the connection preface to the remote endpoint.
+      sendPreface(ctx);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+      cleanup();
+      super.channelInactive(ctx);
+    }
+
+    /** Releases the {@code clientPrefaceString}. Any active streams will be left in the open. */
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+      cleanup();
+    }
+
+    /** Releases the {@code clientPrefaceString}. Any active streams will be left in the open. */
+    private void cleanup() {
+      if (clientPrefaceString != null) {
+        clientPrefaceString.release();
+        clientPrefaceString = null;
+      }
+    }
+
+    /**
+     * Decodes the client connection preface string from the input buffer.
+     *
+     * @return {@code true} if processing of the client preface string is complete. Since client
+     *     preface strings can only be received by servers, returns true immediately for client
+     *     endpoints.
+     */
+    private boolean readClientPrefaceString(ByteBuf in) throws Http2Exception {
+      if (clientPrefaceString == null) {
+        return true;
+      }
+
+      int prefaceRemaining = clientPrefaceString.readableBytes();
+      int bytesRead = min(in.readableBytes(), prefaceRemaining);
+
+      // If the input so far doesn't match the preface, break the connection.
+      if (bytesRead == 0
+          || !ByteBufUtil.equals(
+              in,
+              in.readerIndex(),
+              clientPrefaceString,
+              clientPrefaceString.readerIndex(),
+              bytesRead)) {
+        int maxSearch = 1024; // picked because 512 is too little, and 2048 too much
+        int http1Index =
+            ByteBufUtil.indexOf(
+                HTTP_1_X_BUF, in.slice(in.readerIndex(), min(in.readableBytes(), maxSearch)));
+        if (http1Index != -1) {
+          String chunk =
+              in.toString(in.readerIndex(), http1Index - in.readerIndex(), CharsetUtil.US_ASCII);
+          throw connectionError(PROTOCOL_ERROR, "Unexpected HTTP/1.x request: %s", chunk);
+        }
+        String receivedBytes =
+            hexDump(
+                in, in.readerIndex(), min(in.readableBytes(), clientPrefaceString.readableBytes()));
+        throw connectionError(
+            PROTOCOL_ERROR,
+            "HTTP/2 client preface string missing or corrupt. " + "Hex dump for received bytes: %s",
+            receivedBytes);
+      }
+      in.skipBytes(bytesRead);
+      clientPrefaceString.skipBytes(bytesRead);
+
+      if (!clientPrefaceString.isReadable()) {
+        // Entire preface has been read.
+        clientPrefaceString.release();
+        clientPrefaceString = null;
+        return true;
+      }
+      return false;
+    }
+
+    /**
+     * Peeks at that the next frame in the buffer and verifies that it is a non-ack {@code SETTINGS}
+     * frame.
+     *
+     * @param in the inbound buffer.
+     * @return {@code true} if the next frame is a non-ack {@code SETTINGS} frame, {@code false} if
+     *     more data is required before we can determine the next frame type.
+     * @throws Http2Exception thrown if the next frame is NOT a non-ack {@code SETTINGS} frame.
+     */
+    private boolean verifyFirstFrameIsSettings(ByteBuf in) throws Http2Exception {
+      if (in.readableBytes() < 5) {
+        // Need more data before we can see the frame type for the first frame.
+        return false;
+      }
+
+      short frameType = in.getUnsignedByte(in.readerIndex() + 3);
+      short flags = in.getUnsignedByte(in.readerIndex() + 4);
+      if (frameType != SETTINGS || (flags & Http2Flags.ACK) != 0) {
+        throw connectionError(
+            PROTOCOL_ERROR,
+            "First received frame was not SETTINGS. " + "Hex dump for first 5 bytes: %s",
+            hexDump(in, in.readerIndex(), 5));
+      }
+      return true;
+    }
+
+    /**
+     * Sends the HTTP/2 connection preface upon establishment of the connection, if not already
+     * sent.
+     */
+    private void sendPreface(ChannelHandlerContext ctx) throws Exception {
+      if (prefaceSent || !ctx.channel().isActive()) {
+        return;
+      }
+
+      prefaceSent = true;
+
+      final boolean isClient = !connection().isServer();
+      if (isClient) {
+        // Clients must send the preface string as the first bytes on the connection.
+        ctx.write(connectionPrefaceBuf()).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
+      }
+
+      // Both client and server must send their initial settings.
+      encoder
+          .writeSettings(ctx, initialSettings, ctx.newPromise())
+          .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
+
+      if (isClient) {
+        // If this handler is extended by the user and we directly fire the userEvent from this
+        // context then
+        // the user will not see the event. We should fire the event starting with this handler so
+        // this class
+        // (and extending classes) have a chance to process the event.
+        userEventTriggered(ctx, Http2ConnectionPrefaceAndSettingsFrameWrittenEvent.INSTANCE);
+      }
+    }
+  }
+
+  private final class FrameDecoder extends BaseDecoder {
+    @Override
+    public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+      try {
+        decoder.decodeFrame(ctx, in, out);
+      } catch (Throwable e) {
+        onError(ctx, false, e);
+      }
+    }
+  }
+
+  @Override
+  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+    // Initialize the encoder, decoder, flow controllers, and internal state.
+    encoder.lifecycleManager(this);
+    decoder.lifecycleManager(this);
+    encoder.flowController().channelHandlerContext(ctx);
+    decoder.flowController().channelHandlerContext(ctx);
+    byteDecoder = new PrefaceDecoder(ctx);
+  }
+
+  @Override
+  protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
+    if (byteDecoder != null) {
+      byteDecoder.handlerRemoved(ctx);
+      byteDecoder = null;
+    }
+  }
+
+  @Override
+  public void channelActive(ChannelHandlerContext ctx) throws Exception {
+    if (byteDecoder == null) {
+      byteDecoder = new PrefaceDecoder(ctx);
+    }
+    byteDecoder.channelActive(ctx);
+    super.channelActive(ctx);
+  }
+
+  @Override
+  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+    // Call super class first, as this may result in decode being called.
+    super.channelInactive(ctx);
+    if (byteDecoder != null) {
+      byteDecoder.channelInactive(ctx);
+      byteDecoder = null;
+    }
+  }
+
+  @Override
+  public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+    // Writability is expected to change while we are writing. We cannot allow this event to trigger
+    // reentering
+    // the allocation and write loop. Reentering the event loop will lead to over or illegal
+    // allocation.
+    try {
+      if (ctx.channel().isWritable()) {
+        flush(ctx);
+      }
+      encoder.flowController().channelWritabilityChanged();
+    } finally {
+      super.channelWritabilityChanged(ctx);
+    }
+  }
+
+  @Override
+  protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+    byteDecoder.decode(ctx, in, out);
+  }
+
+  @Override
+  public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
+      throws Exception {
+    ctx.bind(localAddress, promise);
+  }
+
+  @Override
+  public void connect(
+      ChannelHandlerContext ctx,
+      SocketAddress remoteAddress,
+      SocketAddress localAddress,
+      ChannelPromise promise)
+      throws Exception {
+    ctx.connect(remoteAddress, localAddress, promise);
+  }
+
+  @Override
+  public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+    ctx.disconnect(promise);
+  }
+
+  @Override
+  public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+    if (decoupleCloseAndGoAway) {
+      ctx.close(promise);
+      return;
+    }
+    promise = promise.unvoid();
+    // Avoid NotYetConnectedException
+    if (!ctx.channel().isActive()) {
+      ctx.close(promise);
+      return;
+    }
+
+    // If the user has already sent a GO_AWAY frame they may be attempting to do a graceful shutdown
+    // which requires
+    // sending multiple GO_AWAY frames. We should only send a GO_AWAY here if one has not already
+    // been sent. If
+    // a GO_AWAY has been sent we send a empty buffer just so we can wait to close until all other
+    // data has been
+    // flushed to the OS.
+    // https://github.com/netty/netty/issues/5307
+    ChannelFuture f =
+        connection().goAwaySent() ? ctx.write(EMPTY_BUFFER) : goAway(ctx, null, ctx.newPromise());
+    ctx.flush();
+    doGracefulShutdown(ctx, f, promise);
+  }
+
+  private ChannelFutureListener newClosingChannelFutureListener(
+      ChannelHandlerContext ctx, ChannelPromise promise) {
+    long gracefulShutdownTimeoutMillis = this.gracefulShutdownTimeoutMillis;
+    return gracefulShutdownTimeoutMillis < 0
+        ? new ClosingChannelFutureListener(ctx, promise)
+        : new ClosingChannelFutureListener(
+            ctx, promise, gracefulShutdownTimeoutMillis, MILLISECONDS);
+  }
+
+  private void doGracefulShutdown(
+      ChannelHandlerContext ctx, ChannelFuture future, final ChannelPromise promise) {
+    final ChannelFutureListener listener = newClosingChannelFutureListener(ctx, promise);
+    if (isGracefulShutdownComplete()) {
+      // If there are no active streams, close immediately after the GO_AWAY write completes or the
+      // timeout
+      // elapsed.
+      future.addListener(listener);
+    } else {
+      // If there are active streams we should wait until they are all closed before closing the
+      // connection.
+
+      // The ClosingChannelFutureListener will cascade promise completion. We need to always notify
+      // the
+      // new ClosingChannelFutureListener when the graceful close completes if the promise is not
+      // null.
+      if (closeListener == null) {
+        closeListener = listener;
+      } else if (promise != null) {
+        final ChannelFutureListener oldCloseListener = closeListener;
+        closeListener =
+            new ChannelFutureListener() {
+              @Override
+              public void operationComplete(ChannelFuture future) throws Exception {
+                try {
+                  oldCloseListener.operationComplete(future);
+                } finally {
+                  listener.operationComplete(future);
+                }
+              }
+            };
+      }
+    }
+  }
+
+  @Override
+  public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+    ctx.deregister(promise);
+  }
+
+  @Override
+  public void read(ChannelHandlerContext ctx) throws Exception {
+    ctx.read();
+  }
+
+  @Override
+  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
+      throws Exception {
+    ctx.write(msg, promise);
+  }
+
+  @Override
+  public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+    // Trigger flush after read on the assumption that flush is cheap if there is nothing to write
+    // and that
+    // for flow-control the read may release window that causes data to be written that can now be
+    // flushed.
+    try {
+      // First call channelReadComplete0(...) as this may produce more data that we want to flush
+      channelReadComplete0(ctx);
+    } finally {
+      flush(ctx);
+    }
+  }
+
+  final void channelReadComplete0(ChannelHandlerContext ctx) {
+    // Discard bytes of the cumulation buffer if needed.
+    discardSomeReadBytes();
+
+    // Ensure we never stale the HTTP/2 Channel. Flow-control is enforced by HTTP/2.
+    //
+    // See https://tools.ietf.org/html/rfc7540#section-5.2.2
+    if (!ctx.channel().config().isAutoRead()) {
+      ctx.read();
+    }
+
+    ctx.fireChannelReadComplete();
+  }
+
+  /**
+   * Handles {@link Http2Exception} objects that were thrown from other handlers. Ignores all other
+   * exceptions.
+   */
+  @Override
+  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+    if (getEmbeddedHttp2Exception(cause) != null) {
+      // Some exception in the causality chain is an Http2Exception - handle it.
+      onError(ctx, false, cause);
+    } else {
+      super.exceptionCaught(ctx, cause);
+    }
+  }
+
+  /**
+   * Closes the local side of the given stream. If this causes the stream to be closed, adds a hook
+   * to close the channel after the given future completes.
+   *
+   * @param stream the stream to be half closed.
+   * @param future If closing, the future after which to close the channel.
+   */
+  @Override
+  public void closeStreamLocal(Http2Stream stream, ChannelFuture future) {
+    switch (stream.state()) {
+      case HALF_CLOSED_LOCAL:
+      case OPEN:
+        stream.closeLocalSide();
+        break;
+      default:
+        closeStream(stream, future);
+        break;
+    }
+  }
+
+  /**
+   * Closes the remote side of the given stream. If this causes the stream to be closed, adds a hook
+   * to close the channel after the given future completes.
+   *
+   * @param stream the stream to be half closed.
+   * @param future If closing, the future after which to close the channel.
+   */
+  @Override
+  public void closeStreamRemote(Http2Stream stream, ChannelFuture future) {
+    switch (stream.state()) {
+      case HALF_CLOSED_REMOTE:
+      case OPEN:
+        stream.closeRemoteSide();
+        break;
+      default:
+        closeStream(stream, future);
+        break;
+    }
+  }
+
+  @Override
+  public void closeStream(final Http2Stream stream, ChannelFuture future) {
+    stream.close();
+
+    if (future.isDone()) {
+      checkCloseConnection(future);
+    } else {
+      future.addListener(
+          new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+              checkCloseConnection(future);
+            }
+          });
+    }
+  }
+
+  /** Central handler for all exceptions caught during HTTP/2 processing. */
+  @Override
+  public void onError(ChannelHandlerContext ctx, boolean outbound, Throwable cause) {
+    Http2Exception embedded = getEmbeddedHttp2Exception(cause);
+    if (isStreamError(embedded)) {
+      onStreamError(ctx, outbound, cause, (StreamException) embedded);
+    } else if (embedded instanceof CompositeStreamException) {
+      CompositeStreamException compositException = (CompositeStreamException) embedded;
+      for (StreamException streamException : compositException) {
+        onStreamError(ctx, outbound, cause, streamException);
+      }
+    } else {
+      onConnectionError(ctx, outbound, cause, embedded);
+    }
+    ctx.flush();
+  }
+
+  /**
+   * Called by the graceful shutdown logic to determine when it is safe to close the connection.
+   * Returns {@code true} if the graceful shutdown has completed and the connection can be safely
+   * closed. This implementation just guarantees that there are no active streams. Subclasses may
+   * override to provide additional checks.
+   */
+  protected boolean isGracefulShutdownComplete() {
+    return connection().numActiveStreams() == 0;
+  }
+
+  /**
+   * Handler for a connection error. Sends a GO_AWAY frame to the remote endpoint. Once all streams
+   * are closed, the connection is shut down.
+   *
+   * @param ctx the channel context
+   * @param outbound {@code true} if the error was caused by an outbound operation.
+   * @param cause the exception that was caught
+   * @param http2Ex the {@link Http2Exception} that is embedded in the causality chain. This may be
+   *     {@code null} if it's an unknown exception.
+   */
+  protected void onConnectionError(
+      ChannelHandlerContext ctx, boolean outbound, Throwable cause, Http2Exception http2Ex) {
+    if (http2Ex == null) {
+      http2Ex = new Http2Exception(INTERNAL_ERROR, cause.getMessage(), cause);
+    }
+
+    ChannelPromise promise = ctx.newPromise();
+    ChannelFuture future = goAway(ctx, http2Ex, ctx.newPromise());
+    if (http2Ex.shutdownHint() == Http2Exception.ShutdownHint.GRACEFUL_SHUTDOWN) {
+      doGracefulShutdown(ctx, future, promise);
+    } else {
+      future.addListener(newClosingChannelFutureListener(ctx, promise));
+    }
+  }
+
+  /**
+   * Handler for a stream error. Sends a {@code RST_STREAM} frame to the remote endpoint and closes
+   * the stream.
+   *
+   * @param ctx the channel context
+   * @param outbound {@code true} if the error was caused by an outbound operation.
+   * @param cause the exception that was caught
+   * @param http2Ex the {@link StreamException} that is embedded in the causality chain.
+   */
+  protected void onStreamError(
+      ChannelHandlerContext ctx,
+      boolean outbound,
+      @SuppressWarnings("unused") Throwable cause,
+      StreamException http2Ex) {
+    final int streamId = http2Ex.streamId();
+    Http2Stream stream = connection().stream(streamId);
+
+    // if this is caused by reading headers that are too large, send a header with status 431
+    if (http2Ex instanceof Http2Exception.HeaderListSizeException
+        && ((Http2Exception.HeaderListSizeException) http2Ex).duringDecode()
+        && connection().isServer()) {
+
+      // NOTE We have to check to make sure that a stream exists before we send our reply.
+      // We likely always create the stream below as the stream isn't created until the
+      // header block is completely processed.
+
+      // The case of a streamId referring to a stream which was already closed is handled
+      // by createStream and will land us in the catch block below
+      if (stream == null) {
+        try {
+          stream = encoder.connection().remote().createStream(streamId, true);
+        } catch (Http2Exception e) {
+          resetUnknownStream(ctx, streamId, http2Ex.error().code(), ctx.newPromise());
+          return;
+        }
+      }
+
+      // ensure that we have not already sent headers on this stream
+      if (stream != null && !stream.isHeadersSent()) {
+        try {
+          handleServerHeaderDecodeSizeError(ctx, stream);
+        } catch (Throwable cause2) {
+          onError(ctx, outbound, connectionError(INTERNAL_ERROR, cause2, "Error DecodeSizeError"));
+        }
+      }
+    }
+
+    if (stream == null) {
+      if (!outbound || connection().local().mayHaveCreatedStream(streamId)) {
+        resetUnknownStream(ctx, streamId, http2Ex.error().code(), ctx.newPromise());
+      }
+    } else {
+      resetStream(ctx, stream, http2Ex.error().code(), ctx.newPromise());
+    }
+  }
+
+  /**
+   * Notifies client that this server has received headers that are larger than what it is willing
+   * to accept. Override to change behavior.
+   *
+   * @param ctx the channel context
+   * @param stream the Http2Stream on which the header was received
+   */
+  protected void handleServerHeaderDecodeSizeError(ChannelHandlerContext ctx, Http2Stream stream) {
+    encoder().writeHeaders(ctx, stream.id(), HEADERS_TOO_LARGE_HEADERS, 0, true, ctx.newPromise());
+  }
+
+  protected Http2FrameWriter frameWriter() {
+    return encoder().frameWriter();
+  }
+
+  /**
+   * Sends a {@code RST_STREAM} frame even if we don't know about the stream. This error condition
+   * is most likely triggered by the first frame of a stream being invalid. That is, there was an
+   * error reading the frame before we could create a new stream.
+   */
+  private ChannelFuture resetUnknownStream(
+      final ChannelHandlerContext ctx, int streamId, long errorCode, ChannelPromise promise) {
+    ChannelFuture future = frameWriter().writeRstStream(ctx, streamId, errorCode, promise);
+    if (future.isDone()) {
+      closeConnectionOnError(ctx, future);
+    } else {
+      future.addListener(
+          new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+              closeConnectionOnError(ctx, future);
+            }
+          });
+    }
+    return future;
+  }
+
+  @Override
+  public ChannelFuture resetStream(
+      final ChannelHandlerContext ctx, int streamId, long errorCode, ChannelPromise promise) {
+    final Http2Stream stream = connection().stream(streamId);
+    if (stream == null) {
+      return resetUnknownStream(ctx, streamId, errorCode, promise.unvoid());
+    }
+
+    return resetStream(ctx, stream, errorCode, promise);
+  }
+
+  private ChannelFuture resetStream(
+      final ChannelHandlerContext ctx,
+      final Http2Stream stream,
+      long errorCode,
+      ChannelPromise promise) {
+    promise = promise.unvoid();
+    if (stream.isResetSent()) {
+      // Don't write a RST_STREAM frame if we have already written one.
+      return promise.setSuccess();
+    }
+    // Synchronously set the resetSent flag to prevent any subsequent calls
+    // from resulting in multiple reset frames being sent.
+    //
+    // This needs to be done before we notify the promise as the promise may have a listener
+    // attached that
+    // call resetStream(...) again.
+    stream.resetSent();
+
+    final ChannelFuture future;
+    // If the remote peer is not aware of the steam, then we are not allowed to send a RST_STREAM
+    // https://tools.ietf.org/html/rfc7540#section-6.4.
+    if (stream.state() == IDLE
+        || connection().local().created(stream)
+            && !stream.isHeadersSent()
+            && !stream.isPushPromiseSent()) {
+      future = promise.setSuccess();
+    } else {
+      future = frameWriter().writeRstStream(ctx, stream.id(), errorCode, promise);
+    }
+    if (future.isDone()) {
+      processRstStreamWriteResult(ctx, stream, future);
+    } else {
+      future.addListener(
+          new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+              processRstStreamWriteResult(ctx, stream, future);
+            }
+          });
+    }
+
+    return future;
+  }
+
+  @Override
+  public ChannelFuture goAway(
+      final ChannelHandlerContext ctx,
+      final int lastStreamId,
+      final long errorCode,
+      final ByteBuf debugData,
+      ChannelPromise promise) {
+    promise = promise.unvoid();
+    final Http2Connection connection = connection();
+    try {
+      if (!connection.goAwaySent(lastStreamId, errorCode, debugData)) {
+        debugData.release();
+        promise.trySuccess();
+        return promise;
+      }
+    } catch (Throwable cause) {
+      debugData.release();
+      promise.tryFailure(cause);
+      return promise;
+    }
+
+    // Need to retain before we write the buffer because if we do it after the refCnt could already
+    // be 0 and
+    // result in an IllegalRefCountException.
+    debugData.retain();
+    ChannelFuture future =
+        frameWriter().writeGoAway(ctx, lastStreamId, errorCode, debugData, promise);
+
+    if (future.isDone()) {
+      processGoAwayWriteResult(ctx, lastStreamId, errorCode, debugData, future);
+    } else {
+      future.addListener(
+          new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+              processGoAwayWriteResult(ctx, lastStreamId, errorCode, debugData, future);
+            }
+          });
+    }
+
+    return future;
+  }
+
+  /**
+   * Closes the connection if the graceful shutdown process has completed.
+   *
+   * @param future Represents the status that will be passed to the {@link #closeListener}.
+   */
+  private void checkCloseConnection(ChannelFuture future) {
+    // If this connection is closing and the graceful shutdown has completed, close the connection
+    // once this operation completes.
+    if (closeListener != null && isGracefulShutdownComplete()) {
+      ChannelFutureListener closeListener = this.closeListener;
+      // This method could be called multiple times
+      // and we don't want to notify the closeListener multiple times.
+      this.closeListener = null;
+      try {
+        closeListener.operationComplete(future);
+      } catch (Exception e) {
+        throw new IllegalStateException("Close listener threw an unexpected exception", e);
+      }
+    }
+  }
+
+  /**
+   * Close the remote endpoint with with a {@code GO_AWAY} frame. Does <strong>not</strong> flush
+   * immediately, this is the responsibility of the caller.
+   */
+  private ChannelFuture goAway(
+      ChannelHandlerContext ctx, Http2Exception cause, ChannelPromise promise) {
+    long errorCode = cause != null ? cause.error().code() : NO_ERROR.code();
+    int lastKnownStream = connection().remote().lastStreamCreated();
+    return goAway(ctx, lastKnownStream, errorCode, Http2CodecUtil.toByteBuf(ctx, cause), promise);
+  }
+
+  private void processRstStreamWriteResult(
+      ChannelHandlerContext ctx, Http2Stream stream, ChannelFuture future) {
+    if (future.isSuccess()) {
+      closeStream(stream, future);
+    } else {
+      // The connection will be closed and so no need to change the resetSent flag to false.
+      onConnectionError(ctx, true, future.cause(), null);
+    }
+  }
+
+  private void closeConnectionOnError(ChannelHandlerContext ctx, ChannelFuture future) {
+    if (!future.isSuccess()) {
+      onConnectionError(ctx, true, future.cause(), null);
+    }
+  }
+
+  /**
+   * Returns the client preface string if this is a client connection, otherwise returns {@code
+   * null}.
+   */
+  private static ByteBuf clientPrefaceString(Http2Connection connection) {
+    return connection.isServer() ? connectionPrefaceBuf() : null;
+  }
+
+  private static void processGoAwayWriteResult(
+      final ChannelHandlerContext ctx,
+      final int lastStreamId,
+      final long errorCode,
+      final ByteBuf debugData,
+      ChannelFuture future) {
+    try {
+      if (future.isSuccess()) {
+        if (errorCode != NO_ERROR.code()) {
+          if (logger.isDebugEnabled()) {
+            logger.debug(
+                "{} Sent GOAWAY: lastStreamId '{}', errorCode '{}', "
+                    + "debugData '{}'. Forcing shutdown of the connection.",
+                ctx.channel(),
+                lastStreamId,
+                errorCode,
+                debugData.toString(UTF_8),
+                future.cause());
+          }
+          ctx.close();
+        }
+      } else {
+        if (logger.isDebugEnabled()) {
+          logger.debug(
+              "{} Sending GOAWAY failed: lastStreamId '{}', errorCode '{}', "
+                  + "debugData '{}'. Forcing shutdown of the connection.",
+              ctx.channel(),
+              lastStreamId,
+              errorCode,
+              debugData.toString(UTF_8),
+              future.cause());
+        }
+        ctx.close();
+      }
+    } finally {
+      // We're done with the debug data now.
+      debugData.release();
+    }
+  }
+
+  /** Closes the channel when the future completes. */
+  private static final class ClosingChannelFutureListener implements ChannelFutureListener {
+    private final ChannelHandlerContext ctx;
+    private final ChannelPromise promise;
+    private final ScheduledFuture<?> timeoutTask;
+    private boolean closed;
+
+    ClosingChannelFutureListener(ChannelHandlerContext ctx, ChannelPromise promise) {
+      this.ctx = ctx;
+      this.promise = promise;
+      timeoutTask = null;
+    }
+
+    ClosingChannelFutureListener(
+        final ChannelHandlerContext ctx,
+        final ChannelPromise promise,
+        long timeout,
+        TimeUnit unit) {
+      this.ctx = ctx;
+      this.promise = promise;
+      timeoutTask =
+          ctx.executor()
+              .schedule(
+                  new Runnable() {
+                    @Override
+                    public void run() {
+                      doClose();
+                    }
+                  },
+                  timeout,
+                  unit);
+    }
+
+    @Override
+    public void operationComplete(ChannelFuture sentGoAwayFuture) {
+      if (timeoutTask != null) {
+        timeoutTask.cancel(false);
+      }
+      doClose();
+    }
+
+    private void doClose() {
+      // We need to guard against multiple calls as the timeout may trigger close() first and then
+      // it will be
+      // triggered again because of operationComplete(...) is called.
+      if (closed) {
+        // This only happens if we also scheduled a timeout task.
+        assert timeoutTask != null;
+        return;
+      }
+      closed = true;
+      if (promise == null) {
+        ctx.close();
+      } else {
+        ctx.close(promise);
+      }
+    }
+  }
+}
diff --git a/src/test/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java b/src/test/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java
new file mode 100644
index 0000000000..764b874471
--- /dev/null
+++ b/src/test/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java
@@ -0,0 +1,3 @@
+package io.netty.handler.codec.http2;
+
+public class DefaultHttp2RemoteFlowController {}