SerialTaskManagerWorker implements TaskManagerWorker { @Override public Mono<Task.Result> executeTask(TaskWithId taskWithId) { if (!cancelledTasks.remove(taskWithId.getId())) { Mono<Task.Result> taskMono = Mono.fromCallable(() -> runWithMdc(taskWithId, listener)).subscribeOn(taskExecutor); CompletableFuture<Task.Result> future = taskMono.toFuture(); runningTask.set(Tuples.of(taskWithId.getId(), future)); return Mono.using( () -> pollAdditionalInformation(taskWithId).subscribe(), ignored -> Mono.fromFuture(future) .onErrorResume(exception -> Mono.from(handleExecutionError(taskWithId, listener, exception)) .thenReturn(Task.Result.PARTIAL)), Disposable::dispose); } else { return Mono.from(listener.cancelled(taskWithId.getId(), taskWithId.getTask().details())) .then(Mono.empty()); } } SerialTaskManagerWorker(Listener listener, Duration pollingInterval); @Override Mono<Task.Result> executeTask(TaskWithId taskWithId); @Override void cancelTask(TaskId taskId); @Override Publisher<Void> fail(TaskId taskId, Optional<TaskExecutionDetails.AdditionalInformation> additionalInformation, String errorMessage, Throwable reason); @Override void close(); static final boolean MAY_INTERRUPT_IF_RUNNING; }
@Test void aSuccessfullTaskShouldCompleteSuccessfully() { TaskWithId taskWithId = new TaskWithId(TaskId.generateTaskId(), this.successfulTask); Mono<Task.Result> result = worker.executeTask(taskWithId); assertThat(result.block()).isEqualTo(Task.Result.COMPLETED); verify(listener, atLeastOnce()).completed(taskWithId.getId(), Task.Result.COMPLETED, Optional.empty()); }
@Test void aRunningTaskShouldProvideInformationUpdatesDuringExecution() throws InterruptedException { TaskWithId taskWithId = new TaskWithId(TaskId.generateTaskId(), new MemoryReferenceWithCounterTask((counter) -> Mono.fromCallable(counter::incrementAndGet) .delayElement(Duration.ofSeconds(2)) .repeat(10) .then(Mono.just(Task.Result.COMPLETED)) .block())); worker.executeTask(taskWithId).subscribe(); TimeUnit.SECONDS.sleep(2); verify(listener, atLeastOnce()).updated(eq(taskWithId.getId()), notNull()); }
@Test void aRunningTaskShouldHaveAFiniteNumberOfInformation() { TaskWithId taskWithId = new TaskWithId(TaskId.generateTaskId(), new MemoryReferenceWithCounterTask((counter) -> Mono.fromCallable(counter::incrementAndGet) .delayElement(Duration.ofSeconds(1)) .repeat(3) .then(Mono.just(Task.Result.COMPLETED)) .block())); worker.executeTask(taskWithId).block(); verify(listener, atMost(4)).updated(eq(taskWithId.getId()), notNull()); }
@Test void aRunningTaskShouldEmitAtMostOneInformationPerSecond() { TaskWithId taskWithId = new TaskWithId(TaskId.generateTaskId(), new MemoryReferenceWithCounterTask((counter) -> Mono.fromCallable(counter::incrementAndGet) .delayElement(Duration.ofMillis(10)) .repeat(200) .then(Mono.just(Task.Result.COMPLETED)) .block())); worker.executeTask(taskWithId).block(); verify(listener, atMost(3)).updated(eq(taskWithId.getId()), notNull()); }
@Test void aFailedTaskShouldCompleteWithFailedStatus() { TaskWithId taskWithId = new TaskWithId(TaskId.generateTaskId(), failedTask); Mono<Task.Result> result = worker.executeTask(taskWithId); assertThat(result.block()).isEqualTo(Task.Result.PARTIAL); verify(listener, atLeastOnce()).failed(taskWithId.getId(), Optional.empty()); }
@Test void aThrowingTaskShouldCompleteWithFailedStatus() { TaskWithId taskWithId = new TaskWithId(TaskId.generateTaskId(), throwingTask); Mono<Task.Result> result = worker.executeTask(taskWithId); assertThat(result.block()).isEqualTo(Task.Result.PARTIAL); verify(listener, atLeastOnce()).failed(eq(taskWithId.getId()), eq(Optional.empty()), any(RuntimeException.class)); }
@Test void theWorkerShouldReportThatATaskIsInProgress() throws InterruptedException { TaskId id = TaskId.generateTaskId(); CountDownLatch latch = new CountDownLatch(1); CountDownLatch taskLaunched = new CountDownLatch(1); Task inProgressTask = new MemoryReferenceTask(() -> { taskLaunched.countDown(); await(latch); return Task.Result.COMPLETED; }); TaskWithId taskWithId = new TaskWithId(id, inProgressTask); worker.executeTask(taskWithId).subscribe(); await(taskLaunched); verify(listener, atLeastOnce()).started(id); verifyNoMoreInteractions(listener); latch.countDown(); }