IO: add more byte packet tests and fix few bugs

Sergey Mashkov · Sergey Mashkov · commit 5feaf6973c23 · 2017-09-06T21:50:51.000+03:00
diff --git a/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacket.kt b/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacket.kt
@@ -1,6 +1,6 @@
 package kotlinx.coroutines.experimental.io.packet
 
-import java.io.InputStream
+import java.io.*
 import java.nio.ByteBuffer
 import kotlin.experimental.and
 
@@ -28,11 +28,14 @@ interface ByteReadPacket {
     fun readFloat(): Float
 
     fun skip(n: Int): Int
-    fun skipExact(n: Int)
+    fun skipExact(n: Int) {
+        if (skip(n) != n) throw EOFException("Unable to skip $n bytes due to end of packet")
+    }
 
     fun release()
 
     fun readUTF8LineTo(out: Appendable, limit: Int = Int.MAX_VALUE): Boolean
 
     fun inputStream(): InputStream
+    fun readerUTF8(): Reader
 }
diff --git a/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacketEmpty.kt b/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacketEmpty.kt
@@ -1,8 +1,7 @@
 package kotlinx.coroutines.experimental.io.packet
 
-import java.io.EOFException
-import java.io.InputStream
-import java.nio.ByteBuffer
+import java.io.*
+import java.nio.*
 
 object ByteReadPacketEmpty : ByteReadPacket {
     override val remaining: Int
@@ -44,22 +43,17 @@ object ByteReadPacketEmpty : ByteReadPacket {
         throw EOFException("Couldn't read float from empty packet")
     }
 
-    override fun skip(n: Int): Int {
-        return 0
-    }
-
-    override fun skipExact(n: Int) {
-        if (n != 0) throw EOFException("Couldn't skip $n bytes in empty packet")
-    }
+    override fun skip(n: Int) = 0
 
     override fun release() {
     }
 
     override fun readUTF8LineTo(out: Appendable, limit: Int) = false
 
-    override fun inputStream() = EmptyInputStream
+    override fun inputStream(): InputStream = EmptyInputStream
+    override fun readerUTF8(): Reader = EmptyReader
 
-    private val EmptyInputStream = object : InputStream() {
+    private object EmptyInputStream : InputStream() {
         override fun available() = 0
 
         override fun read(): Int = -1
@@ -78,4 +72,13 @@ object ByteReadPacketEmpty : ByteReadPacket {
         override fun close() {
         }
     }
+
+    private object EmptyReader : Reader() {
+        override fun close() {
+        }
+
+        override fun read(cbuf: CharArray?, off: Int, len: Int) = -1
+        override fun read() = -1
+        override fun read(target: CharBuffer?) = -1
+    }
 }
diff --git a/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacketImpl.kt b/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacketImpl.kt
@@ -4,11 +4,10 @@ package kotlinx.coroutines.experimental.io.packet
 
 import kotlinx.coroutines.experimental.io.internal.ObjectPool
 import kotlinx.coroutines.experimental.io.internal.decodeUTF8
-import java.io.EOFException
-import java.io.InputStream
+import java.io.*
 import java.nio.BufferOverflowException
 import java.nio.ByteBuffer
-import java.nio.charset.MalformedInputException
+import java.nio.charset.*
 import java.util.*
 
 internal class ByteReadPacketImpl(private val packets: ArrayDeque<ByteBuffer>, internal val pool: ObjectPool<ByteBuffer>) : ByteReadPacket {
@@ -141,7 +140,7 @@ internal class ByteReadPacketImpl(private val packets: ArrayDeque<ByteBuffer>, i
             !end && size == 0
         }
 
-        if (!rc && size != 0) throw MalformedInputException(0)
+        if (rc && size > 0) throw MalformedInputException(0)
 
         return rc
     }
@@ -160,10 +159,6 @@ internal class ByteReadPacketImpl(private val packets: ArrayDeque<ByteBuffer>, i
         return skipped
     }
 
-    override fun skipExact(n: Int) {
-        if (skip(n) != n) throw EOFException("Unable to skip $n bytes due to end of packet")
-    }
-
     override fun inputStream(): InputStream {
         return object : InputStream() {
             override fun read(): Int {
@@ -182,6 +177,38 @@ internal class ByteReadPacketImpl(private val packets: ArrayDeque<ByteBuffer>, i
         }
     }
 
+    override fun readerUTF8(): Reader {
+        return object : Reader() {
+            override fun close() {
+                release()
+            }
+
+            override fun read(cbuf: CharArray, off: Int, len: Int): Int {
+                var decoded = 0
+                var size = 1
+
+                val rc = reading(size) { bb ->
+                    size = bb.decodeUTF8 { ch ->
+                        if (decoded == len) false
+                        else {
+                            cbuf[off + decoded] = ch
+                            decoded++
+                            true
+                        }
+                    }
+
+                    size == 0
+                }
+
+                return when {
+                    rc && size > 0 -> throw CharacterCodingException()
+                    rc -> decoded
+                    else -> -1
+                }
+            }
+        }
+    }
+
     override fun release() {
         while (packets.isNotEmpty()) {
             recycle(packets.remove())
@@ -227,6 +254,8 @@ internal class ByteReadPacketImpl(private val packets: ArrayDeque<ByteBuffer>, i
         val extraBytes = size - buffer.remaining()
         val next = packets.peekFirst()
 
+        if (extraBytes > next.remaining()) return  false
+
         buffer.compact()
         repeat(extraBytes) {
             buffer.put(next.get())
diff --git a/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacketSingle.kt b/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacketSingle.kt
@@ -2,8 +2,7 @@ package kotlinx.coroutines.experimental.io.packet
 
 import kotlinx.coroutines.experimental.io.internal.ObjectPool
 import kotlinx.coroutines.experimental.io.internal.decodeUTF8
-import java.io.EOFException
-import java.io.InputStream
+import java.io.*
 import java.nio.BufferOverflowException
 import java.nio.ByteBuffer
 import java.nio.charset.MalformedInputException
@@ -151,10 +150,6 @@ internal class ByteReadPacketSingle(private var buffer: ByteBuffer?, internal va
         return skipped
     }
 
-    override fun skipExact(n: Int) {
-        if (skip(n) != n) throw EOFException("Unable to skip $n bytes due to end of packet")
-    }
-
     override fun inputStream(): InputStream {
         return object : InputStream() {
             override fun read(): Int {
@@ -170,6 +165,39 @@ internal class ByteReadPacketSingle(private var buffer: ByteBuffer?, internal va
             }
 
             override fun available() = remaining
+
+            override fun close() {
+                release()
+            }
+        }
+    }
+
+    override fun readerUTF8(): Reader {
+        return object : Reader() {
+            override fun close() {
+                release()
+            }
+
+            override fun read(cbuf: CharArray, off: Int, len: Int): Int {
+                var decoded = 0
+
+                val rc = reading { bb ->
+                    if (len == 0) return@reading
+
+                    val rc = bb.decodeUTF8 { ch ->
+                        if (decoded == len) false
+                        else {
+                            cbuf[off + decoded] =ch
+                            decoded++
+                            true
+                        }
+                    }
+
+                    if (rc > 0) throw MalformedInputException(0)
+                }
+
+                return if (!rc) -1 else decoded
+            }
         }
     }
 
diff --git a/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteWritePacket.kt b/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteWritePacket.kt
@@ -1,6 +1,6 @@
 package kotlinx.coroutines.experimental.io.packet
 
-import java.io.OutputStream
+import java.io.*
 import java.nio.ByteBuffer
 import java.nio.CharBuffer
 
@@ -26,6 +26,7 @@ interface ByteWritePacket : Appendable {
     fun build(): ByteReadPacket
 
     fun outputStream(): OutputStream
+    fun writerUTF8(): Writer
 
     override fun append(csq: CharSequence): ByteWritePacket {
         append(csq, 0, csq.length)
diff --git a/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteWritePacketImpl.kt b/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteWritePacketImpl.kt
@@ -1,7 +1,7 @@
 package kotlinx.coroutines.experimental.io.packet
 
 import kotlinx.coroutines.experimental.io.internal.ObjectPool
-import java.io.OutputStream
+import java.io.*
 import java.nio.ByteBuffer
 import java.nio.CharBuffer
 import java.util.*
@@ -105,7 +105,52 @@ internal class ByteWritePacketImpl(private val pool: ObjectPool<ByteBuffer>) : B
 
     // expects at least one byte remaining in [bb]
     private tailrec fun appendUTF8(csq: CharSequence, start: Int, end: Int, bb: ByteBuffer) {
-        for (i in start .. end) {
+        val limitedEnd = minOf(end, start + bb.remaining())
+
+        for (i in start until limitedEnd) {
+            val chi = csq[i].toInt() and 0xffff
+            val requiredSize = when {
+                chi <= 0x7f -> 1
+                chi > 0x7ff -> 3
+                else -> 2
+            }
+
+            if (bb.remaining() < requiredSize) {
+                return appendUTF8(csq, i, end, appendNewBuffer())
+            }
+
+            size += bb.putUtf8Char(chi)
+        }
+
+        if (limitedEnd < end) {
+            return appendUTF8(csq, limitedEnd, end, appendNewBuffer())
+        }
+    }
+
+    private tailrec fun appendASCII(csq: CharArray, start: Int, end: Int) {
+        val bb = ensure()
+        val limitedEnd = minOf(end, start + bb.remaining())
+
+        for (i in start until limitedEnd) {
+            val chi = csq[i].toInt() and 0xffff
+            if (chi >= 0x80) {
+                appendUTF8(csq, i, end, bb)
+                return
+            }
+
+            bb.put(chi.toByte())
+            size++
+        }
+
+        if (limitedEnd < end) {
+            return appendASCII(csq, limitedEnd, end)
+        }
+    }
+
+    // expects at least one byte remaining in [bb]
+    private tailrec fun appendUTF8(csq: CharArray, start: Int, end: Int, bb: ByteBuffer) {
+        val limitedEnd = minOf(end, start + bb.remaining())
+        for (i in start until limitedEnd) {
             val chi = csq[i].toInt() and 0xffff
             val requiredSize = when {
                 chi <= 0x7f -> 1
@@ -119,6 +164,10 @@ internal class ByteWritePacketImpl(private val pool: ObjectPool<ByteBuffer>) : B
 
             size += bb.putUtf8Char(chi)
         }
+
+        if (limitedEnd < end) {
+            return appendUTF8(csq, limitedEnd, end, appendNewBuffer())
+        }
     }
 
     override fun writeStringUtf8(s: String) {
@@ -164,6 +213,20 @@ internal class ByteWritePacketImpl(private val pool: ObjectPool<ByteBuffer>) : B
         }
     }
 
+    override fun writerUTF8(): Writer {
+        return object : Writer() {
+            override fun write(cbuf: CharArray, off: Int, len: Int) {
+                appendASCII(cbuf, off, len)
+            }
+
+            override fun flush() {
+            }
+
+            override fun close() {
+            }
+        }
+    }
+
     override fun build(): ByteReadPacket {
         val bs = buffers ?: return ByteReadPacketEmpty
         buffers = null
@@ -202,7 +265,7 @@ internal class ByteWritePacketImpl(private val pool: ObjectPool<ByteBuffer>) : B
         val buffer = last()?.takeIf { it.remaining() >= size }
 
         if (buffer == null) {
-            block(ensure())
+            block(appendNewBuffer())
         } else {
             block(buffer)
         }
diff --git a/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/Packets.kt b/core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/Packets.kt
@@ -5,7 +5,7 @@ import kotlinx.coroutines.experimental.io.internal.ObjectPoolImpl
 import kotlinx.coroutines.experimental.io.internal.getIOIntProperty
 import java.nio.ByteBuffer
 
-private val PACKET_BUFFER_SIZE = getIOIntProperty("PacketBufferSize", 4096)
+internal val PACKET_BUFFER_SIZE = getIOIntProperty("PacketBufferSize", 4096)
 private val PACKET_BUFFER_POOL_SIZE = getIOIntProperty("PacketBufferPoolSize", 128)
 
 private val PacketBufferPool: ObjectPool<ByteBuffer> =
diff --git a/core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/BytePacketBuildTest.kt b/core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/BytePacketBuildTest.kt
diff --git a/core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/BytePacketReaderWriterTest.kt b/core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/BytePacketReaderWriterTest.kt
