EventSourced-AbortStageHandler.kt

@Component
class AbortStageHandler(
  override val queue: Queue,
  override val eventRepository: ExecutionEventRepository,
  @Qualifier("queueEventPublisher") private val publisher: ApplicationEventPublisher,
  private val clock: Clock
) : OrcaMessageHandler<AbortStage> {

  override fun handle(message: AbortStage) {
    message.withStage { stage ->
      if (stage.status in setOf(RUNNING, NOT_STARTED)) {
        stage.status = TERMINAL
        stage.endTime = clock.millis()
        eventReposotory.record(StageAborted(stage, message))
        queue.push(CancelStage(message))
        if (stage.parentStageId == null) {
          queue.push(CompleteExecution(message))
        } else {
          queue.push(CompleteStage(stage.parent()))
        }
        publisher.publishEvent(StageComplete(this, stage))
      }
    }
  }

  override val messageType = AbortStage::class.java
}

Today-AbortStageHandler.kt

@Component
class AbortStageHandler(
  override val queue: Queue,
  override val repository: ExecutionRepository,
  @Qualifier("queueEventPublisher") private val publisher: ApplicationEventPublisher,
  private val clock: Clock
) : OrcaMessageHandler<AbortStage> {

  override fun handle(message: AbortStage) {
    message.withStage { stage ->
      if (stage.status in setOf(RUNNING, NOT_STARTED)) {
        stage.status = TERMINAL
        stage.endTime = clock.millis()
        repository.storeStage(stage)
        queue.push(CancelStage(message))
        if (stage.parentStageId == null) {
          queue.push(CompleteExecution(message))
        } else {
          queue.push(CompleteStage(stage.parent()))
        }
        publisher.publishEvent(StageComplete(this, stage))
      }
    }
  }

  override val messageType = AbortStage::class.java
}