Http2RemoteFlowController stream writibility listener

Motivation:
For implementations that want to manage flow control down to the stream level it is useful to be notified when stream writability changes.

Modifications:
- Add writabilityChanged to Http2RemoteFlowController.Listener
- Add isWritable to Http2RemoteFlowController

Result:
The Http2RemoteFlowController provides notification when writability of a stream changes.

codec-http2/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java

@@ -438,6 +438,7 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exceptio
             if (ctx.channel().isWritable()) {
                 flush(ctx);
             }
+            encoder.flowController().channelWritabilityChanged();
         } finally {
             super.channelWritabilityChanged(ctx);
         }

codec-http2/src/main/java/io/netty/handler/codec/http2/Http2RemoteFlowController.java

@@ -56,11 +56,21 @@ public interface Http2RemoteFlowController extends Http2FlowController {
     void listener(Listener listener);
 
     /**
-     * Get the current listener to flow-control events.
-     *
-     * @return the current listener or {@code null} if one is not set.
+     * Determine if the {@code stream} has bytes remaining for use in the flow control window.
+     * <p>
+     * Note that this only takes into account HTTP/2 flow control. It does <strong>not</strong> take into account
+     * the underlying {@link io.netty.channel.Channel#isWritable()}.
+     * @param stream The stream to test.
+     * @return {@code true} if if the {@code stream} has bytes remaining for use in the flow control window.
+     * {@code false} otherwise.
+     */
+    boolean isWritable(Http2Stream stream);
+
+    /**
+     * Notification that the writability of {@link #channelHandlerContext()} has changed.
+     * @throws Http2Exception If any writes occur as a result of this call and encounter errors.
      */
-    Listener listener();
+    void channelWritabilityChanged() throws Http2Exception;
 
     /**
      * Implementations of this interface are used to progressively write chunks of the underlying
@@ -132,11 +142,20 @@ interface Listener {
         /**
          * Report the number of {@code writtenBytes} for a {@code stream}. Called after the
          * flow-controller has flushed bytes for the given stream.
-         *
+         * <p>
+         * This method should not throw. Any thrown exceptions are considered a programming error and are ignored.
          * @param stream that had bytes written.
          * @param writtenBytes the number of bytes written for a stream, can be 0 in the case of an
          *                     empty DATA frame.
          */
         void streamWritten(Http2Stream stream, int writtenBytes);
+
+        /**
+         * Notification that {@link Http2RemoteFlowController#isWritable(Http2Stream)} has changed for {@code stream}.
+         * <p>
+         * This method should not throw. Any thrown exceptions are considered a programming error and are ignored.
+         * @param stream The stream which writability has changed for.
+         */
+        void writabilityChanged(Http2Stream stream);
     }
 }

codec-http2/src/main/java/io/netty/handler/codec/http2/PriorityStreamByteDistributor.java

@@ -54,7 +54,7 @@ public void onStreamClosed(Http2Stream stream) {
             public void onPriorityTreeParentChanged(Http2Stream stream, Http2Stream oldParent) {
                 Http2Stream parent = stream.parent();
                 if (parent != null) {
-                    int delta = state(stream).unallocatedStreamableBytesForTree();
+                    long delta = state(stream).unallocatedStreamableBytesForTree();
                     if (delta != 0) {
                         state(parent).unallocatedStreamableBytesForTreeChanged(delta);
                     }
@@ -65,7 +65,7 @@ public void onPriorityTreeParentChanged(Http2Stream stream, Http2Stream oldParen
             public void onPriorityTreeParentChanging(Http2Stream stream, Http2Stream newParent) {
                 Http2Stream parent = stream.parent();
                 if (parent != null) {
-                    int delta = state(stream).unallocatedStreamableBytesForTree();
+                    long delta = state(stream).unallocatedStreamableBytesForTree();
                     if (delta != 0) {
                         state(parent).unallocatedStreamableBytesForTreeChanged(-delta);
                     }
@@ -103,7 +103,7 @@ int unallocatedStreamableBytes(Http2Stream stream) {
     /**
      * For testing only.
      */
-    int unallocatedStreamableBytesForTree(Http2Stream stream) {
+    long unallocatedStreamableBytesForTree(Http2Stream stream) {
         return state(stream).unallocatedStreamableBytesForTree();
     }
 
@@ -307,7 +307,7 @@ private final class PriorityState {
         boolean hasFrame;
         int streamableBytes;
         int allocated;
-        int unallocatedStreamableBytesForTree;
+        long unallocatedStreamableBytesForTree;
 
         PriorityState(Http2Stream stream) {
             this.stream = stream;
@@ -317,7 +317,7 @@ private final class PriorityState {
          * Recursively increments the {@link #unallocatedStreamableBytesForTree()} for this branch in
          * the priority tree starting at the current node.
          */
-        void unallocatedStreamableBytesForTreeChanged(int delta) {
+        void unallocatedStreamableBytesForTreeChanged(long delta) {
             unallocatedStreamableBytesForTree += delta;
             if (!stream.isRoot()) {
                 state(stream.parent()).unallocatedStreamableBytesForTreeChanged(delta);
@@ -371,7 +371,7 @@ int unallocatedStreamableBytes() {
             return streamableBytes - allocated;
         }
 
-        int unallocatedStreamableBytesForTree() {
+        long unallocatedStreamableBytesForTree() {
             return unallocatedStreamableBytesForTree;
         }
     }