WorkerDispatcher implements Dispatcher { @Override public Optional<Message> dispatch(Message message) { Object payload = message.getPayload(); if (payload == null) { LOG.error("Can't dispatch message {} as its payload is null", message); return Optional.empty(); } Optional<Processor> processorCandidate = getProcessor(payload); if (processorCandidate.isPresent()) { try { processorCandidate.get().process(message); eventBus.fireAsync(new MessageProcessedEvent(message)); } catch (Exception e) { eventBus.fireAsync(new MessageNotProcessedEvent(message)); return Optional.of(message); } } else { LOG.warn("Skipping message {} as no corresponding processor exists", message); } return Optional.empty(); } @Autowired  WorkerDispatcher(ApplicationContext applicationContext, EventBus eventBus); @Override Optional<Message> dispatch(Message message);  }
@Test public void testEmptyMessage() { assertThat( dispatcher .dispatch(message(MOCK, null)) .isPresent(), is(false) ); }
@Test public void testUnknownProcessor() { assertThat( dispatcher .dispatch(message(MOCK, "unknown_payload")) .isPresent(), is(false) ); }
@Test public void testNoProcessor() { assertThat( dispatcher.dispatch(message(MOCK, 123d)).isPresent(), is(false) ); }
@Test public void testProcessingException() { Message message = message(MOCK, 42); assertThat( dispatcher.dispatch(message), equalTo(Optional.of(message)) ); }
@Test public void testProcessingSuccess() { assertThat( dispatcher.dispatch(message(MOCK, "payload")).isPresent(), is(false) ); assertThat(alwaysSucceedingProcessor.wasProcessed(), is(true)); }