asyncfilereader.kt

class AsyncFileReader(
    file: File,
    override val coroutineContext: CoroutineContext = Dispatchers.IO
) : CoroutineScope, Closeable {
    private val fileChannel = file.inputStream().buffered()
    private val bufferChannel: Channel<ByteBuffer> = Channel(capacity = 2)
    private var job: Job? = null

    fun start() {
        job = launch {
            while (isActive) {
                val byteBuffer = ByteBufferPool.borrow()
                when (val read = fileChannel.read(byteBuffer.array())) {
                    -1 -> {
                        close()
                    }
                    else -> {
                        byteBuffer.limit(read)
                        bufferChannel.send(byteBuffer)
                    }
                }
            }
        }
    }

    suspend fun <T> read(block: suspend (ByteBuffer?) -> T): T {
        return block(bufferChannel.receiveOrNull())
    }

    override fun close() {
        job?.cancel()
        fileChannel.close()
        bufferChannel.close()
    }
}