HTTP/2 UniformStreamByteDistributor negative window shouldn't write

Motivation:
If the stream window is negative UniformStreamByteDistributor may write data. This is prohibited by the RFC https://tools.ietf.org/html/rfc7540#section-6.9.2.

Modifications:
- UniformStreamByteDistributor should use StreamState.isWriteAllowed()

Result:
UniformStreamByteDistributor is more complaint with HTTP/2 RFC.
Fixes https://github.com/netty/netty/issues/4545

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

@@ -97,10 +97,13 @@ public final class UniformStreamByteDistributor implements StreamByteDistributor
         State state = queue.pollFirst();
         do {
             state.enqueued = false;
-            if (state.streamableBytes > 0 && maxBytes == 0) {
+            if (state.windowNegative) {
-                // Stop at the first state that can't send. Add this state back to the head of
+                continue;
-                // the queue. Note that empty frames at the head of the queue will always be
+            }
-                // written.
+            if (maxBytes == 0 && state.streamableBytes > 0) {
+                // Stop at the first state that can't send. Add this state back to the head of the queue. Note
+                // that empty frames at the head of the queue will always be written, assuming the stream window
+                // is not negative.
                 queue.addFirst(state);
                 state.enqueued = true;
                 break;
@@ -134,6 +137,7 @@ public final class UniformStreamByteDistributor implements StreamByteDistributor
     private final class State {
         final Http2Stream stream;
         int streamableBytes;
+        boolean windowNegative;
         boolean enqueued;
         boolean writing;
 
@@ -149,12 +153,15 @@ public final class UniformStreamByteDistributor implements StreamByteDistributor
                 streamableBytes = newStreamableBytes;
                 totalStreamableBytes += delta;
             }
-            // We should queue this state if there is a frame. We don't want to queue this frame if the window
+            // In addition to only enqueuing state when they have frames we enforce the following restrictions:
-            // size is <= 0 and we are writing this state. The rational being we already gave this state the chance to
+            // 1. If the window has gone negative. We never want to queue a state. However we also don't want to
-            // write, and if there were empty frames the expectation is they would have been sent. At this point there
+            //    Immediately remove the item if it is already queued because removal from dequeue is O(n). So
-            // must be a call to updateStreamableBytes for this state to be able to write again.
+            //    we allow it to stay queued and rely on the distribution loop to remove this state.
-            if (hasFrame && (!writing || windowSize > 0)) {
+            // 2. If the window is zero we only want to queue if we are not writing. If we are writing that means
-                // It's not in the queue but has data to send, add it.
+            //    we gave the state a chance to write zero length frames. We wait until updateStreamableBytes is
+            //    called again before this state is allowed to write.
+            windowNegative = windowSize < 0;
+            if (hasFrame && (windowSize > 0 || (windowSize == 0 && !writing))) {
                 addToQueue();
             }
         }

codec-http2/src/test/java/io/netty/handler/codec/http2/UniformStreamByteDistributorTest.java

@@ -211,6 +211,19 @@ public class UniformStreamByteDistributorTest {
         verifyNoMoreInteractions(writer);
     }
 
+    @Test
+    public void streamWindowExhaustedDoesNotWrite() throws Http2Exception {
+        updateStream(STREAM_A, 0, true, false);
+        updateStream(STREAM_B, 0, true);
+        updateStream(STREAM_C, 0, true);
+        updateStream(STREAM_D, 0, true, false);
+
+        assertFalse(write(10));
+        verifyWrite(STREAM_B, 0);
+        verifyWrite(STREAM_C, 0);
+        verifyNoMoreInteractions(writer);
+    }
+
     private Http2Stream stream(int streamId) {
         return connection.stream(streamId);
     }