IO: fix joining and continuous writing byte array interference

Author: Sergey Mashkov

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

@@ -1428,7 +1428,10 @@ internal class ByteBufferChannel(
     private tailrec suspend fun writeFullySuspend(src: ByteArray, offset: Int, length: Int) {
         if (length == 0) return
         writeSuspendUnit(src, offset, length)
-        val copied = writeSuspendResult
+        val copied = writeSuspendResult.takeIf { it != -1 } ?:
+                joining?.let { resolveDelegation(this, it) }?.writeSuspendResult?.takeIf { it != -1 } ?:
+                error("-1 should be only set in case of joining")
+
         return writeFullySuspend(src, offset + copied, length - copied)
     }
 
@@ -1446,7 +1449,12 @@ internal class ByteBufferChannel(
         while (true) {
             tryWriteSuspend(1)
 
-            joining?.let { resolveDelegation(this, it)?.let { return it.writeSuspendUnit(src, offset, length) } }
+            joining?.let {
+                resolveDelegation(this, it)?.let {
+                    writeSuspendResult = -1;
+                    return it.writeSuspendUnit(src, offset, length)
+                }
+            }
 
             val size = writeAsMuchAsPossible(src, offset, length)
             if (size > 0) {

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

@@ -1618,6 +1618,33 @@ class ByteBufferChannelTest : TestBase() {
         assertEquals(0x22, next.readByte())
     }
 
+    @Test
+    fun testJoiningDuringWriteFully() = runTest {
+        val bb = ByteArray(65536)
+        Random().nextBytes(bb)
+        val dest = ByteChannel()
+
+        launch(coroutineContext) {
+            expect(1)
+            ch.writeFully(bb)
+            ch.flush()
+        }
+        yield()
+        expect(2)
+
+        launch {
+            expect(3)
+            ch.joinTo(dest, false)
+        }
+        yield()
+
+        val result = ByteArray(bb.size)
+        dest.readFully(result)
+
+        Assert.assertArrayEquals(bb, result)
+        finish(4)
+    }
+
     private inline fun buildPacket(block: ByteWritePacket.() -> Unit): ByteReadPacket {
         val builder = BytePacketBuilder(0, pktPool)
         try {