Command queue: core logic

CommandQueue.kt

class CommandQueue(...) {
    // ...
    
    override fun schedule(command: CommandSpecification) {
        // add task to queue, log details, emit metrics
        schedule(ScheduledTask(
                command,
                queueName,
                clock,
                // this is the important part for deduplication to work
                scheduledExecutionDate = executionPolicy.computeNextExecutionDate(clock, tries = 0)
        ), command.deduplicate)
    }

    // ...

    override fun processCommands(): Boolean {
        val circuitBreaker = circuitBreaker()
        if (circuitBreaker.isOpen()) {
            return false
        }

        val task = taskRepository.tryLockingTaskWithEarliestScheduleOlderThan(queueName, LocalDateTime.now(clock))
                ?: return false

        val command = commandRegistry.get(task.commandName)

        // each case: decides what to do with task, log details, emit metrics
        val executionResult = when {
            command == null -> commandNotFound(task)
            violatesRateLimit(task.weight) -> rateLimited(task)
            !running -> aborted(task)
            else -> executeCommand(command, task)
        }

        registerCall(circuitBreaker, executionResult)
        // remove task, or move it to quarantine, or update number of tries
        handleExecutionResult(executionResult)
        return executionResult.commandExecuted
    }
    
    // ...
}