IO: add byte channel readRemaining test

Author: Sergey Mashkov

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/ByteBufferChannel.kt

@@ -231,7 +231,7 @@ internal class ByteBufferChannel(
         }
     }
 
-    private fun readAsMuchAsPossible(dst: ByteBuffer, consumed0: Int = 0): Int {
+    private tailrec fun readAsMuchAsPossible(dst: ByteBuffer, consumed0: Int = 0): Int {
         var consumed = 0
 
         val rc = reading {
@@ -259,7 +259,7 @@ internal class ByteBufferChannel(
         else consumed + consumed0
     }
 
-    private fun readAsMuchAsPossible(dst: ByteArray, offset: Int, length: Int, consumed0: Int = 0): Int {
+    private tailrec fun readAsMuchAsPossible(dst: ByteArray, offset: Int, length: Int, consumed0: Int = 0): Int {
         var consumed = 0
 
         val rc = reading {

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/ByteReadChannel.kt

@@ -159,7 +159,7 @@ suspend fun ByteReadChannel.copyAndClose(dst: ByteWriteChannel): Long {
  * Reads all the bytes from receiver channel and builds a packet that is returned unless the specified [limit] exceeded.
  * It will simply stop reading and return packet of size [limit] in this case
  */
-suspend fun ByteReadChannel.readAll(limit: Int = Int.MAX_VALUE): ByteReadPacket {
+suspend fun ByteReadChannel.readRemaining(limit: Int = Int.MAX_VALUE): ByteReadPacket {
     val buffer = BufferPool.borrow()
     val packet = WritePacket()
 

core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/CopyAndCloseTest.kt

@@ -1,18 +1,21 @@
 package kotlinx.coroutines.experimental.io
 
-import kotlinx.coroutines.experimental.TestBase
-import kotlinx.coroutines.experimental.launch
-import kotlinx.coroutines.experimental.runBlocking
-import kotlinx.coroutines.experimental.yield
-import org.junit.Test
-import java.io.IOException
-import kotlin.test.assertEquals
-import kotlin.test.fail
+import kotlinx.coroutines.experimental.*
+import kotlinx.coroutines.experimental.io.internal.*
+import org.junit.*
+import java.io.*
+import kotlin.test.*
 
 class CopyAndCloseTest : TestBase() {
     private val from = ByteChannel(true)
     private val to = ByteChannel(true)
 
+    @After
+    fun tearDown() {
+        from.close(CancellationException())
+        to.close(CancellationException())
+    }
+
     @Test
     fun smokeTest() = runBlocking {
         expect(1)
@@ -101,4 +104,50 @@ class CopyAndCloseTest : TestBase() {
 
         finish(6)
     }
+
+    @Test
+    fun readRemaining() = runBlocking {
+        expect(1)
+
+        launch(coroutineContext) {
+            expect(2)
+            from.writeFully("123".toByteArray())
+
+            yield()
+            expect(3)
+            from.writeFully("456".toByteArray().asByteBuffer())
+
+            yield()
+            expect(4)
+            from.close()
+        }
+
+        yield()
+        assertEquals("123456", from.readRemaining().readText().toString())
+
+        yield()
+
+        finish(5)
+    }
+
+    @Test
+    fun readRemainingLimitFailed() = runBlocking {
+        expect(1)
+
+        launch(coroutineContext) {
+            expect(2)
+            from.writeFully("123".toByteArray())
+
+            yield()
+            expect(3)
+            from.writeFully("456".toByteArray().asByteBuffer())
+        }
+
+        yield()
+        assertEquals("12345", from.readRemaining(5).readText().toString())
+
+        finish(4)
+    }
+
+
 }