Merge pull request #5537 from square/jwilson.1006.cancel_close

If RST STREAM and END OF STREAM race, send RST STREAM (4.2.x branch)

Author: squarejesse

okhttp/src/main/java/okhttp3/internal/http2/Http2Stream.kt

@@ -534,6 +534,7 @@ class Http2Stream internal constructor(
     @Throws(IOException::class)
     private fun emitFrame(outFinishedOnLastFrame: Boolean) {
       val toWrite: Long
+      val outFinished: Boolean
       synchronized(this@Http2Stream) {
         writeTimeout.enter()
         try {
@@ -550,11 +551,11 @@ class Http2Stream internal constructor(
         checkOutNotClosed() // Kick out if the stream was reset or closed while waiting.
         toWrite = minOf(writeBytesMaximum - writeBytesTotal, sendBuffer.size)
         writeBytesTotal += toWrite
+        outFinished = outFinishedOnLastFrame && toWrite == sendBuffer.size && errorCode == null
       }
 
       writeTimeout.enter()
       try {
-        val outFinished = outFinishedOnLastFrame && toWrite == sendBuffer.size
         connection.writeData(id, outFinished, sendBuffer, toWrite)
       } finally {
         writeTimeout.exitAndThrowIfTimedOut()
@@ -578,8 +579,11 @@ class Http2Stream internal constructor(
     @Throws(IOException::class)
     override fun close() {
       assert(!Thread.holdsLock(this@Http2Stream))
+
+      val outFinished: Boolean
       synchronized(this@Http2Stream) {
         if (closed) return
+        outFinished = errorCode == null
       }
       if (!sink.finished) {
         // We have 0 or more frames of data, and 0 or more frames of trailers. We need to send at
@@ -592,7 +596,7 @@ class Http2Stream internal constructor(
             while (sendBuffer.size > 0L) {
               emitFrame(false)
             }
-            connection.writeHeaders(id, true, trailers!!.toHeaderList())
+            connection.writeHeaders(id, outFinished, trailers!!.toHeaderList())
           }
 
           hasData -> {
@@ -601,7 +605,7 @@ class Http2Stream internal constructor(
             }
           }
 
-          else -> {
+          outFinished -> {
             connection.writeData(id, true, null, 0L)
           }
         }

okhttp/src/test/java/okhttp3/internal/http2/HttpOverHttp2Test.java

@@ -30,8 +30,10 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import javax.annotation.Nullable;
 import okhttp3.Cache;
 import okhttp3.Call;
 import okhttp3.Callback;
@@ -1620,4 +1622,30 @@ public void shutdownAfterLateCoalescing() throws Exception {
 
     latch.await();
   }
+
+  @Test public void cancelWhileWritingRequestBodySendsCancelToServer() throws Exception {
+    server.enqueue(new MockResponse());
+
+    AtomicReference<Call> callReference = new AtomicReference<>();
+    Call call = client.newCall(new Request.Builder()
+        .url(server.url("/"))
+        .post(new RequestBody() {
+          @Override public @Nullable MediaType contentType() {
+            return MediaType.get("text/plain; charset=utf-8");
+          }
+
+          @Override public void writeTo(BufferedSink sink) {
+            callReference.get().cancel();
+          }
+        })
+        .build());
+    callReference.set(call);
+
+    try {
+      call.execute();
+      fail();
+    } catch (IOException expected) {
+      assertThat(call.isCanceled()).isTrue();
+    }
+  }
 }