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; }
@Test void runHooksShouldCallAllHooks() { when(hook1.notifyDelete(any())).thenReturn(Mono.empty()); when(hook2.notifyDelete(any())).thenReturn(Mono.empty()); testee.runHooks(DELETE_OPERATION).block(); verify(hook1).notifyDelete(DELETE_OPERATION); verify(hook2).notifyDelete(DELETE_OPERATION); verifyNoMoreInteractions(hook1); verifyNoMoreInteractions(hook2); }
@Test void runHooksShouldThrowWhenOneHookThrows() { when(hook1.notifyDelete(any())).thenThrow(new RuntimeException()); when(hook2.notifyDelete(any())).thenReturn(Mono.empty()); assertThatThrownBy(() -> testee.runHooks(DELETE_OPERATION).block()).isInstanceOf(RuntimeException.class); }
@Test void runHooksShouldNotRunHooksAfterAHookThrows() { when(hook1.notifyDelete(any())).thenThrow(new RuntimeException()); when(hook2.notifyDelete(any())).thenReturn(Mono.empty()); try { testee.runHooks(DELETE_OPERATION).block(); } catch (Exception e) { } verifyZeroInteractions(hook2); }
@Test void runHooksShouldThrowWhenOneHookReturnsErrorMono() { when(hook1.notifyDelete(any())).thenReturn(Mono.error(new RuntimeException())); when(hook2.notifyDelete(any())).thenReturn(Mono.empty()); assertThatThrownBy(() -> testee.runHooks(DELETE_OPERATION).block()).isInstanceOf(RuntimeException.class); }
@Test void runHooksShouldNotRunHooksAfterAHookReturnsErrorMono() { when(hook1.notifyDelete(any())).thenReturn(Mono.error(new RuntimeException())); when(hook2.notifyDelete(any())).thenReturn(Mono.empty()); try { testee.runHooks(DELETE_OPERATION).block(); } catch (Exception e) { } verifyZeroInteractions(hook2); }
@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(); }
@Test void runHooksShouldPublishTimerMetrics() { Duration sleepDuration = Duration.ofSeconds(1); Mono<Void> notifyDeleteAnswer = Mono.fromCallable(() -> { Thread.sleep(sleepDuration.toMillis()); return Mono.empty(); }).then(); when(hook1.notifyDelete(any())).thenReturn(notifyDeleteAnswer); when(hook2.notifyDelete(any())).thenReturn(notifyDeleteAnswer); testee.runHooks(DELETE_OPERATION).block(); assertThat(metricFactory.executionTimesFor(PRE_DELETION_HOOK_METRIC_NAME)) .hasSize(2) .allSatisfy(duration -> assertThat(duration).isGreaterThanOrEqualTo(sleepDuration)); }