NonblockingOutputStream.kt

package biz.radioteria.flow.stream

import java.io.IOException
import java.io.OutputStream
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit

class NonblockingOutputStream(private val outputStream: OutputStream) : OutputStream() {

    val QUEUE_SIZE = 128

    val threadPoolExecutor: ThreadPoolExecutor =
            ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, LinkedBlockingQueue(QUEUE_SIZE))

    override fun write(b: Int) {
        blockingAction {
            outputStream.write(b)
        }
    }

    override fun write(b: ByteArray) {
        blockingAction {
            outputStream.write(b)
        }
    }

    override fun write(b: ByteArray, off: Int, len: Int) {
        blockingAction {
            outputStream.write(b, off, len)
        }
    }

    override fun flush() {
        blockingAction {
            outputStream.flush()
        }
    }

    override fun close() {
        blockingAction {
            outputStream.close()
        }
    }

    fun isBlocked(): Boolean {
        return threadPoolExecutor.queue.remainingCapacity() == 0
    }

    private fun blockingAction(block: () -> Unit) {
        if (isBlocked()) {
            throw IOException("Queue size exceeded limit ($QUEUE_SIZE)")
        }
        threadPoolExecutor.submit {
            block.invoke()
        }
    }
}