@Test(timeout = 30_000) public void testExternalActionCancel() throws Exception { newReconciler(current -> Collections.emptyList()); CountDownLatch latch = new CountDownLatch(1); AtomicBoolean canceled = new AtomicBoolean(); Mono<String> action = Mono.<String>never() .doOnSubscribe(s -> latch.countDown()) .doOnCancel(() -> canceled.set(true)); Disposable disposable = reconciler.apply(c -> action).subscribe(); latch.await(); assertThat(disposable.isDisposed()).isFalse(); assertThat(canceled.get()).isFalse(); disposable.dispose(); await().until(canceled::get); assertThat(changesSubscriber.takeNext()).isNull(); assertThat(reconciler.apply(anything -> Mono.just("Regular")).block(TIMEOUT)).isEqualTo("Regular"); }
@Override public Mono<DATA> apply(Function<DATA, Mono<DATA>> action) { return executor.apply(action); }
DefaultOneOffReconciler implements OneOffReconciler<DATA> { @Override public Mono<DATA> apply(Function<DATA, Mono<DATA>> action) { return executor.apply(action); } }
DefaultOneOffReconciler implements OneOffReconciler<DATA> { @Override public Mono<DATA> apply(Function<DATA, Mono<DATA>> action) { return executor.apply(action); } DefaultOneOffReconciler(String id,
                                   DATA initial,
                                   Duration quickCycle,
                                   Duration longCycle,
                                   Function<DATA, List<Mono<Function<DATA, DATA>>>> reconcilerActionsProvider,
                                   Scheduler scheduler,
                                   TitusRuntime titusRuntime); }
DefaultOneOffReconciler implements OneOffReconciler<DATA> { @Override public Mono<DATA> apply(Function<DATA, Mono<DATA>> action) { return executor.apply(action); } DefaultOneOffReconciler(String id,
                                   DATA initial,
                                   Duration quickCycle,
                                   Duration longCycle,
                                   Function<DATA, List<Mono<Function<DATA, DATA>>>> reconcilerActionsProvider,
                                   Scheduler scheduler,
                                   TitusRuntime titusRuntime); @Override Mono<Void> close(); @Override DATA getCurrent(); @Override Mono<DATA> apply(Function<DATA, Mono<DATA>> action); @Override Flux<DATA> changes(); }
DefaultOneOffReconciler implements OneOffReconciler<DATA> { @Override public Mono<DATA> apply(Function<DATA, Mono<DATA>> action) { return executor.apply(action); } DefaultOneOffReconciler(String id,
                                   DATA initial,
                                   Duration quickCycle,
                                   Duration longCycle,
                                   Function<DATA, List<Mono<Function<DATA, DATA>>>> reconcilerActionsProvider,
                                   Scheduler scheduler,
                                   TitusRuntime titusRuntime); @Override Mono<Void> close(); @Override DATA getCurrent(); @Override Mono<DATA> apply(Function<DATA, Mono<DATA>> action); @Override Flux<DATA> changes();  }