IO: implement byte packet copy

Author: Sergey Mashkov

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

@@ -33,6 +33,7 @@ interface ByteReadPacket {
     }
 
     fun release()
+    fun copy(): ByteReadPacket
 
     fun readUTF8LineTo(out: Appendable, limit: Int = Int.MAX_VALUE): Boolean
 

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

@@ -48,6 +48,8 @@ object ByteReadPacketEmpty : ByteReadPacket {
     override fun release() {
     }
 
+    override fun copy() = this
+
     override fun readUTF8LineTo(out: Appendable, limit: Int) = false
 
     override fun inputStream(): InputStream = EmptyInputStream

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

@@ -50,6 +50,7 @@ internal class ByteReadPacketImpl(private val packets: ArrayDeque<ByteBuffer>, i
 
     override fun readFully(dst: ByteArray, offset: Int, length: Int) {
         val rc = readAvailable(dst, offset, length)
+        if (rc == -1 && length == 0) return
         if (rc < length) throw EOFException("Not enough bytes in the packet")
     }
 
@@ -232,6 +233,17 @@ internal class ByteReadPacketImpl(private val packets: ArrayDeque<ByteBuffer>, i
         }
     }
 
+    override fun copy(): ByteReadPacket {
+        if (packets.isEmpty()) return ByteReadPacketEmpty
+        val copyDeque = ArrayDeque<ByteBuffer>(packets.size)
+
+        for (p in packets) {
+            copyDeque.add(pool.borrow().also { it.put(p.duplicate()); it.flip() })
+        }
+
+        return ByteReadPacketImpl(copyDeque, pool)
+    }
+
     private inline fun reading(size: Int, block: (ByteBuffer) -> Boolean): Boolean {
         if (packets.isEmpty()) return false
 

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

@@ -1,5 +1,6 @@
 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 java.io.*
@@ -225,6 +226,14 @@ internal class ByteReadPacketSingle(private var buffer: ByteBuffer?, internal va
         buffer = null
     }
 
+    override fun copy(): ByteReadPacket {
+        val bb = buffer?.takeIf { it.hasRemaining() } ?: return ByteReadPacketEmpty
+        val copyBuffer = pool.borrow()
+        copyBuffer.put(bb.duplicate())
+        copyBuffer.flip()
+        return ByteReadPacketSingle(copyBuffer, pool)
+    }
+
     private inline fun reading(block: (ByteBuffer) -> Unit): Boolean {
         val buffer = buffer ?: return false
 

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

@@ -274,4 +274,35 @@ class BytePacketReaderWriterTest {
 
         assertTrue { bb.contentEquals(p.readBytes()) }
     }
+
+    @Test
+    fun testCopySingleBufferPacket() {
+        val bb = ByteArray(100)
+        Random().nextBytes(bb)
+
+        val p = buildPacket {
+            writeFully(bb)
+        }
+
+        val copy = p.copy()
+        assertEquals(p.remaining, p.remaining)
+        assertTrue { p.readBytes().contentEquals(copy.readBytes()) }
+    }
+
+    @Test
+    fun testCopyMultipleBufferPacket() {
+        val bb = ByteArray(1000000)
+        Random().nextBytes(bb)
+
+        val p = buildPacket {
+            writeFully(bb)
+        }
+
+        val copy = p.copy()
+        assertEquals(p.remaining, p.remaining)
+        val bytes = p.readBytes()
+        val copied = copy.readBytes()
+
+        assertTrue { bytes.contentEquals(copied) }
+    }
 }