MessageDispatcher implements Consumer<M>, Disposable { @Override public void accept(final M message) { if (disposed) { return; } runner.post( () -> { try { consumer.accept(message); } catch (Throwable throwable) { MobiusHooks.handleError( new RuntimeException( "Consumer threw an exception when accepting message: " + message, throwable)); } }); } MessageDispatcher(WorkRunner runner, Consumer<M> consumer); @Override void accept(final M message); @Override void dispose();  }
@Test public void shouldForwardMessagesToConsumer() throws Exception { new MessageDispatcher<String>(WorkRunners.immediate(), messages::add).accept("hey hello"); assertThat(messages).containsExactly("hey hello"); }
@Test public void shouldSendErrorsFromConsumerToMobiusHooks() throws Exception { TestErrorHandler errorHandler = new TestErrorHandler(); MobiusHooks.setErrorHandler(errorHandler); final RuntimeException expected = new RuntimeException("boo"); new MessageDispatcher<String>( WorkRunners.immediate(), s -> { throw expected; }) .accept("here's an event that should be reported as the cause of failure"); assertThat(errorHandler.handledErrors).extracting(Throwable::getCause).contains(expected); assertThat(errorHandler.handledErrors) .extracting(Throwable::getMessage) .has( matching( containsString("here's an event that should be reported as the cause of failure")), atIndex(0)); }