@Test public void testExternalActionFunctionError() throws InterruptedException { newReconcilerWithRegistrations(current -> Collections.emptyList(), "r1", "a"); try { reconciler.apply("r1", c -> { throw new RuntimeException("simulated error"); }).block(); fail("Expected error"); } catch (Exception e) { assertThat(e).isInstanceOf(RuntimeException.class); assertThat(e.getMessage()).isEqualTo("simulated error"); } assertThat(changesSubscriber.takeNext()).isNull(); }
@Override public Mono<DATA> apply(String id, Function<DATA, Mono<DATA>> action) { return Mono.defer(() -> { ReconcilerEngine<DATA> executor = executors.get(id); if (executor == null) { return Mono.error(new IllegalArgumentException("Reconciler not found for data item " + id)); } return executor.apply(action); }); }
DefaultManyReconciler implements ManyReconciler<DATA> { @Override public Mono<DATA> apply(String id, Function<DATA, Mono<DATA>> action) { return Mono.defer(() -> { ReconcilerEngine<DATA> executor = executors.get(id); if (executor == null) { return Mono.error(new IllegalArgumentException("Reconciler not found for data item " + id)); } return executor.apply(action); }); } }
DefaultManyReconciler implements ManyReconciler<DATA> { @Override public Mono<DATA> apply(String id, Function<DATA, Mono<DATA>> action) { return Mono.defer(() -> { ReconcilerEngine<DATA> executor = executors.get(id); if (executor == null) { return Mono.error(new IllegalArgumentException("Reconciler not found for data item " + id)); } return executor.apply(action); }); } DefaultManyReconciler(
            String name,
            Duration quickCycle,
            Duration longCycle,
            Function<DATA, List<Mono<Function<DATA, DATA>>>> reconcilerActionsProvider,
            CloseableReference<Scheduler> reconcilerSchedulerRef,
            CloseableReference<Scheduler> notificationSchedulerRef,
            TitusRuntime titusRuntime); }
DefaultManyReconciler implements ManyReconciler<DATA> { @Override public Mono<DATA> apply(String id, Function<DATA, Mono<DATA>> action) { return Mono.defer(() -> { ReconcilerEngine<DATA> executor = executors.get(id); if (executor == null) { return Mono.error(new IllegalArgumentException("Reconciler not found for data item " + id)); } return executor.apply(action); }); } DefaultManyReconciler(
            String name,
            Duration quickCycle,
            Duration longCycle,
            Function<DATA, List<Mono<Function<DATA, DATA>>>> reconcilerActionsProvider,
            CloseableReference<Scheduler> reconcilerSchedulerRef,
            CloseableReference<Scheduler> notificationSchedulerRef,
            TitusRuntime titusRuntime); @Override Mono<Void> add(String id, DATA initial); @Override Mono<Void> remove(String id); @Override Mono<Void> close(); @Override Map<String, DATA> getAll(); @Override Optional<DATA> findById(String id); @Override Mono<DATA> apply(String id, Function<DATA, Mono<DATA>> action); @Override Flux<List<SimpleReconcilerEvent<DATA>>> changes(); }
DefaultManyReconciler implements ManyReconciler<DATA> { @Override public Mono<DATA> apply(String id, Function<DATA, Mono<DATA>> action) { return Mono.defer(() -> { ReconcilerEngine<DATA> executor = executors.get(id); if (executor == null) { return Mono.error(new IllegalArgumentException("Reconciler not found for data item " + id)); } return executor.apply(action); }); } DefaultManyReconciler(
            String name,
            Duration quickCycle,
            Duration longCycle,
            Function<DATA, List<Mono<Function<DATA, DATA>>>> reconcilerActionsProvider,
            CloseableReference<Scheduler> reconcilerSchedulerRef,
            CloseableReference<Scheduler> notificationSchedulerRef,
            TitusRuntime titusRuntime); @Override Mono<Void> add(String id, DATA initial); @Override Mono<Void> remove(String id); @Override Mono<Void> close(); @Override Map<String, DATA> getAll(); @Override Optional<DATA> findById(String id); @Override Mono<DATA> apply(String id, Function<DATA, Mono<DATA>> action); @Override Flux<List<SimpleReconcilerEvent<DATA>>> changes();  }