@Test void runHooksShouldExecuteHooksSequentially() { ReentrantLock reentrantLock = new ReentrantLock(); Answer<Publisher<Void>> lockAndSleepAnswer = invocationOnMock -> { reentrantLock.lock(); Thread.sleep(Duration.ofMillis(100).toMillis()); reentrantLock.unlock(); return Mono.empty(); }; Answer<Publisher<Void>> throwIfLockedAnswer = invocationOnMock -> { if (reentrantLock.isLocked()) { throw new RuntimeException("This task is running while the previous one is waiting"); } return Mono.empty(); }; when(hook1.notifyDelete(any())).thenAnswer(lockAndSleepAnswer); when(hook2.notifyDelete(any())).thenAnswer(throwIfLockedAnswer); assertThatCode(() -> testee.runHooks(DELETE_OPERATION).block()) .describedAs("RunHook does not throw if hooks are executed in a sequential manner") .doesNotThrowAnyException(); }
public Mono<Void> runHooks(PreDeletionHook.DeleteOperation deleteOperation) { return Flux.fromIterable(hooks) .publishOn(Schedulers.elastic()) .flatMap(hook -> metricFactory.map(factory -> publishMetric(deleteOperation, hook, factory)) .orElse(Mono.empty()), CONCURRENCY) .then(); }
PreDeletionHooks { public Mono<Void> runHooks(PreDeletionHook.DeleteOperation deleteOperation) { return Flux.fromIterable(hooks) .publishOn(Schedulers.elastic()) .flatMap(hook -> metricFactory.map(factory -> publishMetric(deleteOperation, hook, factory)) .orElse(Mono.empty()), CONCURRENCY) .then(); } }
PreDeletionHooks { public Mono<Void> runHooks(PreDeletionHook.DeleteOperation deleteOperation) { return Flux.fromIterable(hooks) .publishOn(Schedulers.elastic()) .flatMap(hook -> metricFactory.map(factory -> publishMetric(deleteOperation, hook, factory)) .orElse(Mono.empty()), CONCURRENCY) .then(); } @Inject  PreDeletionHooks(Set<PreDeletionHook> hooks, MetricFactory metricFactory); private  PreDeletionHooks(Set<PreDeletionHook> hooks, Optional<MetricFactory> metricFactory); }
PreDeletionHooks { public Mono<Void> runHooks(PreDeletionHook.DeleteOperation deleteOperation) { return Flux.fromIterable(hooks) .publishOn(Schedulers.elastic()) .flatMap(hook -> metricFactory.map(factory -> publishMetric(deleteOperation, hook, factory)) .orElse(Mono.empty()), CONCURRENCY) .then(); } @Inject  PreDeletionHooks(Set<PreDeletionHook> hooks, MetricFactory metricFactory); private  PreDeletionHooks(Set<PreDeletionHook> hooks, Optional<MetricFactory> metricFactory); Mono<Void> runHooks(PreDeletionHook.DeleteOperation deleteOperation); }
PreDeletionHooks { public Mono<Void> runHooks(PreDeletionHook.DeleteOperation deleteOperation) { return Flux.fromIterable(hooks) .publishOn(Schedulers.elastic()) .flatMap(hook -> metricFactory.map(factory -> publishMetric(deleteOperation, hook, factory)) .orElse(Mono.empty()), CONCURRENCY) .then(); } @Inject  PreDeletionHooks(Set<PreDeletionHook> hooks, MetricFactory metricFactory); private  PreDeletionHooks(Set<PreDeletionHook> hooks, Optional<MetricFactory> metricFactory); Mono<Void> runHooks(PreDeletionHook.DeleteOperation deleteOperation); static final PreDeletionHooks NO_PRE_DELETION_HOOK; }