DebouncingQueue.kt

/**
 * Executes long running tasks not faster then once in [delayMillis] and makes sure
 * at most one tasks executes at a time. The task itself may take more than [delayMillis]
 * to complete. Task may be asynchronous.
 */
private class DebouncingQueue(
    private val delayMillis: Int,
    parentDisposable: Disposable
) {
    private val alarm = Alarm(Alarm.ThreadToUse.POOLED_THREAD, parentDisposable)
    private var pendingTask: ((CompletionToken) -> Unit)? = null
    private var inFlightTask: CompletionToken? = null

    fun submit(task: (CompletionToken) -> Unit, immediately: Boolean) = onAlarmThread {
        alarm.cancelAllRequests()
        if (immediately) {
            schedule(task)
        } else {
            alarm.addRequest({ schedule(task) }, delayMillis)
        }
    }

    /**
     * Tasks should call [taskCompleted] after they finished (successfully or not), to
     * allow the queue to schedule the next task.
     */
    inner class CompletionToken() {
        fun taskCompleted() = onAlarmThread {
            LOG.assertTrue(inFlightTask == this)
            inFlightTask = null

            val task = pendingTask
            if (task != null) {
                pendingTask = null
                execute(task)
            }
        }
    }

    private fun onAlarmThread(work: () -> Unit) = alarm.addRequest(work, 0)

    private fun schedule(task: (CompletionToken) -> Unit) {
        if (inFlightTask == null) {
            execute(task)
        } else {
            pendingTask = task
        }
    }

    private fun execute(task: (CompletionToken) -> Unit) {
        LOG.assertTrue(inFlightTask == null)
        val token = CompletionToken()
        inFlightTask = token
        task(token)
    }
}