IO: add ByteWritePacket.writePacketUnconsumed

Author: Sergey Mashkov

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

@@ -8,7 +8,7 @@ import java.nio.*
 import java.nio.charset.*
 import java.util.*
 
-internal class ByteReadPacketImpl(private val packets: ArrayDeque<ByteBuffer>, internal val pool: ObjectPool<ByteBuffer>) : ByteReadPacket {
+internal class ByteReadPacketImpl(internal val packets: ArrayDeque<ByteBuffer>, internal val pool: ObjectPool<ByteBuffer>) : ByteReadPacket {
     override val remaining: Int
         get() = packets.sumBy { it.remaining() }
 

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

@@ -1,14 +1,11 @@
 package kotlinx.coroutines.experimental.io.packet
 
-import kotlinx.coroutines.experimental.io.*
-import kotlinx.coroutines.experimental.io.internal.ObjectPool
-import kotlinx.coroutines.experimental.io.internal.decodeUTF8
+import kotlinx.coroutines.experimental.io.internal.*
 import java.io.*
-import java.nio.BufferOverflowException
-import java.nio.ByteBuffer
+import java.nio.*
 import java.nio.charset.*
 
-internal class ByteReadPacketSingle(private var buffer: ByteBuffer?, internal val pool: ObjectPool<ByteBuffer>) : ByteReadPacket {
+internal class ByteReadPacketSingle(internal var buffer: ByteBuffer?, internal val pool: ObjectPool<ByteBuffer>) : ByteReadPacket {
     override val remaining: Int
         get() = buffer?.remaining() ?: 0
 

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

@@ -29,6 +29,7 @@ interface ByteWritePacket : Appendable {
     fun writerUTF8(): Writer
 
     fun writePacket(p: ByteReadPacket)
+    fun writePacketUnconsumed(p: ByteReadPacket)
 
     override fun append(csq: CharSequence): ByteWritePacket {
         append(csq, 0, csq.length)

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

@@ -96,6 +96,24 @@ internal class ByteWritePacketImpl(private val pool: ObjectPool<ByteBuffer>) : B
                     last(p.steal().also { it.compact() })
                 }
             }
+            else -> {
+                writeFully(p.readBytes())
+            }
+        }
+    }
+
+    override fun writePacketUnconsumed(p: ByteReadPacket) {
+        when (p) {
+            is ByteReadPacketEmpty -> {}
+            is ByteReadPacketSingle -> {
+                p.buffer?.duplicate()?.let { writeFully(it) }
+            }
+            is ByteReadPacketImpl -> {
+                for (buffer in p.packets) {
+                    writeFully(buffer.duplicate())
+                }
+            }
+            else -> throw UnsupportedOperationException()
         }
     }
 

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

@@ -319,6 +319,7 @@ class BytePacketReaderWriterTest {
         }
 
         assertEquals("123ABC.", outer.readText().toString())
+        assertEquals(0, inner.remaining)
     }
 
     @Test
@@ -334,5 +335,38 @@ class BytePacketReaderWriterTest {
         }
 
         assertEquals("123" + "o".repeat(100000) + ".", outer.readText().toString())
+        assertEquals(0, inner.remaining)
+    }
+
+    @Test
+    fun testWritePacketSingleUnconsumed() {
+        val inner = buildPacket {
+            append("ABC")
+        }
+
+        val outer = buildPacket {
+            append("123")
+            writePacketUnconsumed(inner)
+            append(".")
+        }
+
+        assertEquals("123ABC.", outer.readText().toString())
+        assertEquals(3, inner.remaining)
+    }
+
+    @Test
+    fun testWritePacketMultipleUnconsumed() {
+        val inner = buildPacket {
+            append("o".repeat(100000))
+        }
+
+        val outer = buildPacket {
+            append("123")
+            writePacketUnconsumed(inner)
+            append(".")
+        }
+
+        assertEquals("123" + "o".repeat(100000) + ".", outer.readText().toString())
+        assertEquals(100000, inner.remaining)
     }
 }