IO: add writeDirect to WritePacket (useful for better NIO interop)

Author: Sergey Mashkov

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

@@ -3,6 +3,7 @@ package kotlinx.coroutines.experimental.io.buffers
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.experimental.io.*
 import kotlinx.coroutines.experimental.io.internal.*
+import kotlinx.coroutines.experimental.io.packet.*
 
 internal class BufferView private constructor(private var content: ByteBuffer,
                                                  private val origin: BufferView?,
@@ -32,8 +33,8 @@ internal class BufferView private constructor(private var content: ByteBuffer,
             field = value
         }
 
-    val readRemaining: Int get() = writePosition - readPosition
-    val writeRemaining: Int get() = limit - writePosition
+    inline val readRemaining: Int get() = writePosition - readPosition
+    inline val writeRemaining: Int get() = limit - writePosition
 
     val startGap: Int get() = readPosition
     val endGap: Int get() = limit - writePosition
@@ -268,11 +269,23 @@ internal class BufferView private constructor(private var content: ByteBuffer,
                 unlink()
                 origin.release()
             } else {
-                pool?.recycle(this)
+                pool?.recycle(this) ?: unlink()
             }
         }
     }
 
+    internal fun writeDirect(size: Int, block: (ByteBuffer) -> Unit) {
+        val rem = writeRemaining
+        require (size <= rem) { "size $size is greater than buffer's remaining capacity $rem" }
+        val buffer = writeDuplicated(rem)
+        val positionBefore = buffer.position()
+        block(buffer)
+        val delta = buffer.position() - positionBefore
+        if (delta < 0 || delta > rem) throw IllegalStateException("Wrong buffer position change: $delta (position should be moved forward only by at most size bytes (size =  $size)")
+
+        writePosition += delta
+    }
+
     private fun readDuplicated(limit: Int = Int.MAX_VALUE): ByteBuffer {
         val position = readPosition
         val upLimit = (position.toLong() + limit).coerceAtMost(Int.MAX_VALUE.toLong()).toInt()
@@ -323,8 +336,8 @@ internal class BufferView private constructor(private var content: ByteBuffer,
         }
     }
 
-    internal object Pool : ObjectPoolImpl<BufferView>(BufferViewCapacity) {
-        override fun produceInstance() = BufferView(ByteBuffer.allocateDirect(BufferViewContentSize), this)
+    internal object Pool : ObjectPoolImpl<BufferView>(PACKET_BUFFER_POOL_SIZE) {
+        override fun produceInstance() = BufferView(ByteBuffer.allocateDirect(PACKET_BUFFER_SIZE), this)
         override fun clearInstance(instance: BufferView): BufferView {
             return instance.apply {
                 next = null
@@ -344,9 +357,6 @@ internal class BufferView private constructor(private var content: ByteBuffer,
     }
 
     companion object {
-        private val BufferViewContentSize = 4096
-        private val BufferViewCapacity = 1024
-
         private val EmptyBuffer = ByteBuffer.allocate(0)!!
 
         val Empty = BufferView(EmptyBuffer)

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

@@ -34,6 +34,8 @@ interface ByteWritePacket : Appendable {
     @Deprecated("Use copy() instead", level = DeprecationLevel.ERROR, replaceWith = ReplaceWith("writePacket(p.copy())"))
     fun writePacketUnconsumed(p: ByteReadPacket) = writePacket(p.copy())
 
+    fun writeDirect(size: Int = 1, block: (ByteBuffer) -> Unit)
+
     override fun append(csq: CharSequence): ByteWritePacket {
         append(csq, 0, csq.length)
         return this

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

@@ -81,6 +81,12 @@ internal class ByteWritePacketImpl(private var headerSizeHint: Int, private val
         size += 4
     }
 
+    override fun writeDirect(size: Int, block: (ByteBuffer) -> Unit) {
+        write(size) {
+            it.writeDirect(size, block)
+        }
+    }
+
     override fun append(c: Char): ByteWritePacket {
         write(3) {
             size += it.putUtf8Char(c.toInt() and 0xffff)

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

@@ -4,7 +4,7 @@ import kotlinx.coroutines.experimental.io.buffers.*
 import kotlinx.coroutines.experimental.io.internal.*
 
 internal val PACKET_BUFFER_SIZE = getIOIntProperty("PacketBufferSize", 4096)
-private val PACKET_BUFFER_POOL_SIZE = getIOIntProperty("PacketBufferPoolSize", 128)
+internal val PACKET_BUFFER_POOL_SIZE = getIOIntProperty("PacketBufferPoolSize", 128)
 internal val PACKET_MAX_COPY_SIZE = getIOIntProperty("PacketMaxCopySize", 500)
 
 internal inline fun buildPacket(pool: ObjectPool<BufferView> = BufferView.Pool, headerSizeHint: Int, block: ByteWritePacket.() -> Unit): ByteReadPacket {

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

@@ -433,5 +433,16 @@ class BytePacketReaderWriterTest {
         assertEquals(100000, inner.remaining)
     }
 
+    @Test
+    fun testWriteDirect() {
+        val packet = buildPacket {
+            writeDirect { bb ->
+                bb.putLong(0x1234567812345678L)
+            }
+        }
+
+        assertEquals(0x1234567812345678L, packet.readLong())
+    }
+
     private inline fun buildPacket(block: ByteWritePacket.() -> Unit) = buildPacket(pool, 0, block)
 }