EventDispatcher { public CompletableFuture<?> dispatch(DispatchedEvent<Event> de) { EventHandler eventHandler = eventTypesAndHandlers.get(de.getEventType()); if (eventHandler != null) { CompletableFuture<Object> result = new CompletableFuture<>(); dispatchEvent(de, eventHandler, result, Optional.empty()); return result; } else { RuntimeException ex = new RuntimeException("No handler for event - subscriberId: " + subscriberId + ", " + de.getEventType()); logger.error("dispatching failure", ex); CompletableFuture completableFuture = new CompletableFuture(); completableFuture.completeExceptionally(ex); return completableFuture; } } EventDispatcher(String subscriberId, Map<Class<?>, EventHandler> eventTypesAndHandlers, EventDeliveryExceptionHandlerManager eventDeliveryExceptionHandlerManager); CompletableFuture<?> dispatch(DispatchedEvent<Event> de);  }
@Test public void shouldDispatchEventFailAndFail() throws Exception { whenEventHandlerDispatchReturning(CompletableFutureUtil.failedFuture(exception)); EventDeliveryExceptionHandlerWithState eventDeliveryExceptionHandlerWithState = mock(EventDeliveryExceptionHandlerWithState.class); when(exceptionHandlerManager.getEventHandler(exception)).thenReturn(eventDeliveryExceptionHandlerWithState); CompletableFuture<?> result = eventDispatcher.dispatch(de); ArgumentCaptor<Consumer> failArg = ArgumentCaptor.forClass(Consumer.class); verify(eventDeliveryExceptionHandlerWithState).handle(eq(exception), any(Runnable.class), failArg.capture(), any(Runnable.class)); verify(exceptionHandlerManager).getEventHandler(exception); failArg.getValue().accept(exception); try { result.get(1, TimeUnit.SECONDS); } catch (ExecutionException e) { assertEquals(exception, e.getCause()); } verify(eventHandler).dispatch(de); }
@Test public void shouldDispatchEventSuccessfully() throws InterruptedException, ExecutionException, TimeoutException { whenEventHandlerDispatchReturning(CompletableFuture.completedFuture(null)); eventDispatcher.dispatch(de).get(1, TimeUnit.SECONDS); verify(eventHandler).dispatch(de); verifyZeroInteractions(exceptionHandlerManager); }
@Test public void shouldDispatchEventFailAndRetry() throws Exception { whenEventHandlerDispatchReturning(CompletableFutureUtil.failedFuture(exception), CompletableFuture.completedFuture(null)); EventDeliveryExceptionHandlerWithState eventDeliveryExceptionHandlerWithState = mock(EventDeliveryExceptionHandlerWithState.class); when(exceptionHandlerManager.getEventHandler(exception)).thenReturn(eventDeliveryExceptionHandlerWithState); CompletableFuture<?> result = eventDispatcher.dispatch(de); ArgumentCaptor<Runnable> redoArg = ArgumentCaptor.forClass(Runnable.class); verify(eventDeliveryExceptionHandlerWithState).handle(eq(exception), redoArg.capture(), any(Consumer.class), any(Runnable.class)); verify(exceptionHandlerManager).getEventHandler(exception); redoArg.getValue().run(); result.get(1, TimeUnit.SECONDS); verify(eventHandler, times(2)).dispatch(de); }
@Test public void shouldDispatchEventFailAndIgnore() throws Exception { whenEventHandlerDispatchReturning(CompletableFutureUtil.failedFuture(exception), CompletableFuture.completedFuture(null)); EventDeliveryExceptionHandlerWithState eventDeliveryExceptionHandlerWithState = mock(EventDeliveryExceptionHandlerWithState.class); when(exceptionHandlerManager.getEventHandler(exception)).thenReturn(eventDeliveryExceptionHandlerWithState); CompletableFuture<?> result = eventDispatcher.dispatch(de); ArgumentCaptor<Runnable> ignoreArg = ArgumentCaptor.forClass(Runnable.class); verify(eventDeliveryExceptionHandlerWithState).handle(eq(exception), any(Runnable.class), any(Consumer.class), ignoreArg.capture()); verify(exceptionHandlerManager).getEventHandler(exception); ignoreArg.getValue().run(); result.get(1, TimeUnit.SECONDS); verify(eventHandler).dispatch(de); }