Always checking MAX_CONCURRENT_STREAMS in

BufferingHttp2ConnectionEncoder.

Currently we don't check this setting when handling a streamClosed()
event. If the setting has lowered prior to this event, the stream
creation could fail.

netty/src/main/java/io/grpc/transport/netty/BufferingHttp2ConnectionEncoder.java

@@ -87,7 +87,7 @@ class BufferingHttp2ConnectionEncoder extends DecoratingHttp2ConnectionEncoder {
 
       @Override
       public void onStreamClosed(Http2Stream stream) {
-        createNextPendingStream();
+        tryCreatePendingStreams();
       }
     });
   }
@@ -178,20 +178,13 @@ class BufferingHttp2ConnectionEncoder extends DecoratingHttp2ConnectionEncoder {
 
   private void tryCreatePendingStreams() {
     while (!pendingStreams.isEmpty() && connection().local().canCreateStream()) {
-      createNextPendingStream();
+      Map.Entry<Integer, PendingStream> entry = pendingStreams.pollFirstEntry();
+      PendingStream pendingStream = entry.getValue();
+      pendingStream.sendFrames();
+      largestCreatedStreamId = pendingStream.streamId;
     }
   }
 
-  private void createNextPendingStream() {
-    Map.Entry<Integer, PendingStream> entry = pendingStreams.pollFirstEntry();
-    if (entry == null) {
-      return;
-    }
-    PendingStream pendingStream = entry.getValue();
-    pendingStream.sendFrames();
-    largestCreatedStreamId = pendingStream.streamId;
-  }
-
   private void cancelPendingStreams() {
     Exception e = new Exception("Connection closed.");
     while (!pendingStreams.isEmpty()) {

netty/src/test/java/io/grpc/transport/netty/BufferingHttp2ConnectionEncoderTest.java

@@ -143,6 +143,28 @@ public class BufferingHttp2ConnectionEncoderTest {
             .writeData(eq(ctx), eq(3), any(ByteBuf.class), eq(0), eq(false), eq(promise));
   }
 
+  @Test
+  public void ensureCanCreateNextStreamWhenStreamCloses() {
+    connection.local().maxActiveStreams(1);
+
+    encoderWriteHeaders(3, promise);
+    // This one gets buffered.
+    encoderWriteHeaders(5, promise);
+    assertEquals(1, connection.numActiveStreams());
+
+    // Now prevent us from creating another stream.
+    connection.local().maxActiveStreams(0);
+
+    // Close the previous stream.
+    connection.stream(3).close();
+
+    // Ensure that no streams are currently active and that only the HEADERS from the first
+    // stream were written.
+    writeVerifyWriteHeaders(times(1), 3, promise);
+    writeVerifyWriteHeaders(never(), 5, promise);
+    assertEquals(0, connection.numActiveStreams());
+  }
+
   @Test
   public void alternatingWritesToActiveAndBufferedStreams() {
     connection.local().maxActiveStreams(1);