@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); }
@Override public ScheduleReference scheduleMono(ScheduleDescriptor scheduleDescriptor, Function<ExecutionContext, Mono<Void>> actionProducer, Scheduler scheduler) { return scheduleInternal(scheduleDescriptor, actionProducer, scheduler, DO_NOTHING); }
DefaultLocalScheduler implements LocalScheduler { @Override public ScheduleReference scheduleMono(ScheduleDescriptor scheduleDescriptor, Function<ExecutionContext, Mono<Void>> actionProducer, Scheduler scheduler) { return scheduleInternal(scheduleDescriptor, actionProducer, scheduler, DO_NOTHING); } }
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); }
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); }
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);  }