Handle source close while reading on Pipe

square · Jan 10, 2025 · 4e009a0 · 4e009a0
1 parent d784119
commit 4e009a0
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 0 deletions.
diff --git a/okio/src/jvmMain/kotlin/okio/Pipe.kt b/okio/src/jvmMain/kotlin/okio/Pipe.kt
@@ -140,6 +140,7 @@ class Pipe(internal val maxBufferSize: Long) {
           if (sinkClosed) return -1L
           timeout.awaitSignal(condition) // Wait until the sink fills the buffer.
           if (canceled) throw IOException("canceled")
+          if (sourceClosed) throw IOException("closed")
         }
 
         val result = buffer.read(sink, byteCount)

diff --git a/okio/src/jvmTest/kotlin/okio/PipeTest.kt b/okio/src/jvmTest/kotlin/okio/PipeTest.kt
@@ -18,6 +18,7 @@ package okio
 import java.io.IOException
 import java.io.InterruptedIOException
 import java.util.Random
+import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import okio.ByteString.Companion.decodeHex
 import okio.HashingSink.Companion.sha1
@@ -280,6 +281,24 @@ class PipeTest {
     pipe.source.close()
   }
 
+  @Test
+  fun sourceCloseWhileReadingThrows() {
+    val pipe = Pipe(100L)
+    val latch = CountDownLatch(1)
+    executorService.execute {
+      try {
+        pipe.source.read(Buffer(), 1)
+        fail()
+      } catch (expected: IOException) {
+        assertEquals("closed", expected.message)
+        latch.countDown()
+      }
+    }
+    Thread.sleep(1000)
+    pipe.source.close()
+    latch.await()
+  }
+
   @Test
   fun sourceReadUnblockedByClosedSink() {
     val pipe = Pipe(3L)