DefaultLocalScheduler implements LocalScheduler { @Override public ScheduleReference scheduleMono(ScheduleDescriptor scheduleDescriptor, Function<ExecutionContext, Mono<Void>> actionProducer, Scheduler scheduler) { return scheduleInternal(scheduleDescriptor, actionProducer, scheduler, DO_NOTHING); } DefaultLocalScheduler(Duration internalLoopInterval, Scheduler scheduler, Clock clock, Registry registry); void shutdown(); @Override List<Schedule> getActiveSchedules(); @Override List<Schedule> getArchivedSchedules(); @Override Optional<Schedule> findSchedule(String scheduleId); @Override Flux<LocalSchedulerEvent> events(); @Override ScheduleReference scheduleMono(ScheduleDescriptor scheduleDescriptor, Function<ExecutionContext, Mono<Void>> actionProducer, Scheduler scheduler); @Override ScheduleReference schedule(ScheduleDescriptor scheduleDescriptor, Consumer<ExecutionContext> action, boolean isolated); @Override ScheduleReference schedule(ScheduleDescriptor scheduleDescriptor, Consumer<ExecutionContext> action, ExecutorService executorService); @Override Mono<Void> cancel(String scheduleId);  }
@Test(timeout = 60_000) public void testScheduleMono() throws Exception { AtomicLong tickCounter = new AtomicLong(); ScheduleReference reference = localScheduler.scheduleMono( scheduleDescriptor.toBuilder().withName("testScheduleMono").build(), tick -> Mono.delay(Duration.ofMillis(1)).flatMap(t -> { tickCounter.incrementAndGet(); return Mono.empty(); }), Schedulers.parallel() ); testExecutionLifecycle(reference, tickCounter); }
@Test public void testTimeout() throws Exception { ScheduleReference reference = localScheduler.scheduleMono( scheduleDescriptor.toBuilder().withName("testTimeout").build(), tick -> Mono.never(), Schedulers.parallel() ); expectScheduleAdded(reference); expectScheduleUpdateEvent(SchedulingState.Running); expectScheduleUpdateEvent(SchedulingState.Failed); expectScheduleUpdateEvent(SchedulingState.Waiting); assertThat(reference.getSchedule().getCompletedActions()).hasSize(1); ScheduledAction failedAction = reference.getSchedule().getCompletedActions().get(0); assertThat(failedAction.getStatus().getState()).isEqualTo(SchedulingState.Failed); assertThat(failedAction.getStatus().getError().get()).isInstanceOf(TimeoutException.class); }
@Test public void testRetries() throws InterruptedException { AtomicInteger counter = new AtomicInteger(); ScheduleReference reference = localScheduler.scheduleMono( scheduleDescriptor.toBuilder().withName("testRetries").build(), tick -> Mono.defer(() -> counter.incrementAndGet() % 2 == 0 ? Mono.empty() : Mono.error(new RuntimeException("Simulated error at iteration " + counter.get()))), Schedulers.parallel() ); expectScheduleAdded(reference); expectScheduleUpdateEvent(SchedulingState.Running); expectScheduleUpdateEvent(SchedulingState.Failed); expectScheduleUpdateEvent(SchedulingState.Waiting); expectScheduleUpdateEvent(SchedulingState.Running); expectScheduleUpdateEvent(SchedulingState.Succeeded); }