Okio based Kotlin suspension extension functions for Java asynchronous NIO channels

OkioAsyncChannels.kt

import kotlinx.coroutines.experimental.nio.aRead
import kotlinx.coroutines.experimental.nio.aWrite
import okio.Buffer
import java.nio.ByteBuffer
import java.nio.channels.AsynchronousFileChannel
import java.nio.channels.AsynchronousSocketChannel
import java.util.concurrent.TimeUnit

suspend fun AsynchronousSocketChannel.aRead(
    buffer: Buffer,
    byteCount: Long,
    timeout: Long = 0L,
    timeUnit: TimeUnit = TimeUnit.MILLISECONDS,
    cursor: Buffer.UnsafeCursor = Buffer.UnsafeCursor()
): Long {
  buffer.readAndWriteUnsafe(cursor).use {
    val oldSize = buffer.size
    val length = minOf(8192, byteCount.toInt())

    cursor.expandBuffer(length)
    // provided by kotlinx-coroutines-nio
    val read = aRead(ByteBuffer.wrap(cursor.data, cursor.start, length), timeout, timeUnit)

    if (read == -1) cursor.resizeBuffer(oldSize)
    else cursor.resizeBuffer(oldSize + read)
    return read.toLong()
  }
}

suspend fun AsynchronousSocketChannel.aWrite(
    buffer: Buffer,
    byteCount: Long,
    timeout: Long = 0L,
    timeUnit: TimeUnit = TimeUnit.MILLISECONDS,
    cursor: Buffer.UnsafeCursor = Buffer.UnsafeCursor()
) {
  var remaining = byteCount
  while (remaining > 0) {
    buffer.readAndWriteUnsafe(cursor).use {
      cursor.seek(0)

      val length = minOf(cursor.end - cursor.start, remaining.toInt())
      // provided by kotlinx-coroutines-nio
      val written = aWrite(ByteBuffer.wrap(cursor.data, cursor.start, length), timeout, timeUnit)

      remaining -= written
      buffer.skip(written.toLong())
    }
  }
}

suspend fun AsynchronousFileChannel.aRead(
    buffer: Buffer,
    position: Long,
    byteCount: Long,
    cursor: Buffer.UnsafeCursor = Buffer.UnsafeCursor()
): Long {
  buffer.readAndWriteUnsafe(cursor).use {
    val oldSize = buffer.size
    val length = minOf(8192, byteCount.toInt())

    cursor.expandBuffer(length)
    // provided by kotlinx-coroutines-nio
    val read = aRead(ByteBuffer.wrap(cursor.data, cursor.start, length), position)

    if (read == -1) cursor.resizeBuffer(oldSize)
    else cursor.resizeBuffer(oldSize + read)
    return read.toLong()
  }
}

suspend fun AsynchronousFileChannel.aWrite(
    buffer: Buffer,
    position: Long,
    byteCount: Long,
    cursor: Buffer.UnsafeCursor = Buffer.UnsafeCursor()
) {
  var remaining = byteCount
  while (remaining > 0) {
    buffer.readAndWriteUnsafe(cursor).use {
      cursor.seek(0)

      val length = minOf(cursor.end - cursor.start, remaining.toInt())
      // provided by kotlinx-coroutines-nio
      val written = aWrite(ByteBuffer.wrap(cursor.data, cursor.start, length), position)

      remaining -= written
      buffer.skip(written.toLong())
    }
  }
}