SequentialExecutorService: java.util.concurrent.ExecutorService implementation which delegates task execution on other ExecutorService sequentially.

SequentialExecutorService.kt

import java.util.concurrent.*
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock

class SequentialExecutorService(executor: ExecutorService = ForkJoinPool.commonPool(), queue: BlockingQueue<Runnable> = LinkedBlockingQueue<Runnable>()) : AbstractExecutorService() {
    val tasks = queue
    val backingExecutor = executor
    var count = 0
    var terminating = false
    val terminated = CountDownLatch(1)
    val lock = ReentrantLock()

    override fun execute(task: Runnable) {
        if (terminating) {
            throw RejectedExecutionException()
        }

        tasks.put(task)
        lock.withLock {
            if (terminating) {
                throw RejectedExecutionException()
            }

            count++
            if (count == 1) {
                next()
            }
        }
    }

    private fun next() {
        if (count == 0) {
            if (terminating) {
                terminated.countDown()
            }
            return
        }

        val task = tasks.peek()
        backingExecutor.submit {
            try {
                task.run()
            } finally {
                lock.withLock {
                    tasks.remove()
                    count--
                    next()
                }
            }
        }
    }

    override fun isTerminated(): Boolean {
        lock.withLock {
            return terminating && count == 0
        }
    }

    override fun shutdown() {
        lock.withLock {
            terminating = true
            if (count == 0) {
                terminated.countDown()
            }
        }
    }

    override fun shutdownNow(): MutableList<Runnable> {
        lock.withLock {
            shutdown()
            val remaining = mutableListOf<Runnable>()
            tasks.drainTo(remaining)
            remaining.forEach { (it as? RunnableFuture<*>)?.cancel(false) }
            return remaining
        }
    }

    override fun isShutdown(): Boolean {
        return terminating
    }

    override fun awaitTermination(timeout: Long, unit: TimeUnit): Boolean {
        return terminated.await(timeout, unit)
    }
}