@Test void testDispatchFailingCommandWithCallbackAndMessageMonitor() throws Exception { CommandMessage<Object> testCommandMessage = GenericCommandMessage.asCommandMessage("fail"); CommandCallback<Object, Object> mockCallback = mock(CommandCallback.class); testSubject.dispatch(testCommandMessage, mockCallback); verify(mockCommandRouter).findDestination(testCommandMessage); verify(mockConnector).send(eq(mockMember), eq(testCommandMessage), any(CommandCallback.class)); verify(mockMessageMonitor).onMessageIngested(any()); verify(mockMonitorCallback).reportFailure(isA(Exception.class)); ArgumentCaptor<CommandResultMessage<Object>> commandResultMessageCaptor = ArgumentCaptor.forClass(CommandResultMessage.class); verify(mockCallback).onResult(eq(testCommandMessage), commandResultMessageCaptor.capture()); assertEquals(Exception.class, commandResultMessageCaptor.getValue().exceptionResult().getClass()); }
@Override public <C> void dispatch(CommandMessage<C> command) { if (defaultCommandCallback != null) { dispatch(command, defaultCommandCallback); return; } LoggingCallback loggingCallback = LoggingCallback.INSTANCE; if (NoOpMessageMonitor.INSTANCE.equals(messageMonitor)) { CommandMessage<? extends C> interceptedCommand = intercept(command); Optional<Member> optionalDestination = commandRouter.findDestination(interceptedCommand); if (optionalDestination.isPresent()) { Member destination = optionalDestination.get(); try { connector.send(destination, interceptedCommand); } catch (Exception e) { destination.suspect(); loggingCallback.onResult(interceptedCommand, asCommandResultMessage( new CommandDispatchException(DISPATCH_ERROR_MESSAGE + ": " + e.getMessage(), e) )); } } else { loggingCallback.onResult(interceptedCommand, asCommandResultMessage(new NoHandlerForCommandException( format("No node known to accept [%s]", interceptedCommand.getCommandName()) ))); } } else { dispatch(command, loggingCallback); } }
DistributedCommandBus implements CommandBus, Distributed<CommandBus> { @Override public <C> void dispatch(CommandMessage<C> command) { if (defaultCommandCallback != null) { dispatch(command, defaultCommandCallback); return; } LoggingCallback loggingCallback = LoggingCallback.INSTANCE; if (NoOpMessageMonitor.INSTANCE.equals(messageMonitor)) { CommandMessage<? extends C> interceptedCommand = intercept(command); Optional<Member> optionalDestination = commandRouter.findDestination(interceptedCommand); if (optionalDestination.isPresent()) { Member destination = optionalDestination.get(); try { connector.send(destination, interceptedCommand); } catch (Exception e) { destination.suspect(); loggingCallback.onResult(interceptedCommand, asCommandResultMessage( new CommandDispatchException(DISPATCH_ERROR_MESSAGE + ": " + e.getMessage(), e) )); } } else { loggingCallback.onResult(interceptedCommand, asCommandResultMessage(new NoHandlerForCommandException( format("No node known to accept [%s]", interceptedCommand.getCommandName()) ))); } } else { dispatch(command, loggingCallback); } } }
DistributedCommandBus implements CommandBus, Distributed<CommandBus> { @Override public <C> void dispatch(CommandMessage<C> command) { if (defaultCommandCallback != null) { dispatch(command, defaultCommandCallback); return; } LoggingCallback loggingCallback = LoggingCallback.INSTANCE; if (NoOpMessageMonitor.INSTANCE.equals(messageMonitor)) { CommandMessage<? extends C> interceptedCommand = intercept(command); Optional<Member> optionalDestination = commandRouter.findDestination(interceptedCommand); if (optionalDestination.isPresent()) { Member destination = optionalDestination.get(); try { connector.send(destination, interceptedCommand); } catch (Exception e) { destination.suspect(); loggingCallback.onResult(interceptedCommand, asCommandResultMessage( new CommandDispatchException(DISPATCH_ERROR_MESSAGE + ": " + e.getMessage(), e) )); } } else { loggingCallback.onResult(interceptedCommand, asCommandResultMessage(new NoHandlerForCommandException( format("No node known to accept [%s]", interceptedCommand.getCommandName()) ))); } } else { dispatch(command, loggingCallback); } } protected  DistributedCommandBus(Builder builder); }
DistributedCommandBus implements CommandBus, Distributed<CommandBus> { @Override public <C> void dispatch(CommandMessage<C> command) { if (defaultCommandCallback != null) { dispatch(command, defaultCommandCallback); return; } LoggingCallback loggingCallback = LoggingCallback.INSTANCE; if (NoOpMessageMonitor.INSTANCE.equals(messageMonitor)) { CommandMessage<? extends C> interceptedCommand = intercept(command); Optional<Member> optionalDestination = commandRouter.findDestination(interceptedCommand); if (optionalDestination.isPresent()) { Member destination = optionalDestination.get(); try { connector.send(destination, interceptedCommand); } catch (Exception e) { destination.suspect(); loggingCallback.onResult(interceptedCommand, asCommandResultMessage( new CommandDispatchException(DISPATCH_ERROR_MESSAGE + ": " + e.getMessage(), e) )); } } else { loggingCallback.onResult(interceptedCommand, asCommandResultMessage(new NoHandlerForCommandException( format("No node known to accept [%s]", interceptedCommand.getCommandName()) ))); } } else { dispatch(command, loggingCallback); } } protected  DistributedCommandBus(Builder builder); static Builder builder(); @ShutdownHandler(phase = Phase.INBOUND_COMMAND_CONNECTOR) void disconnect(); @ShutdownHandler(phase = Phase.OUTBOUND_COMMAND_CONNECTORS) CompletableFuture<Void> shutdownDispatching(); @Override void dispatch(CommandMessage<C> command); @Override void dispatch(CommandMessage<C> command, CommandCallback<? super C, ? super R> callback); @Override Registration subscribe(String commandName, MessageHandler<? super CommandMessage<?>> handler); @Override CommandBus localSegment(); int getLoadFactor(); void updateLoadFactor(int loadFactor); Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super CommandMessage<?>> dispatchInterceptor); @Override Registration registerHandlerInterceptor(
            MessageHandlerInterceptor<? super CommandMessage<?>> handlerInterceptor); }
DistributedCommandBus implements CommandBus, Distributed<CommandBus> { @Override public <C> void dispatch(CommandMessage<C> command) { if (defaultCommandCallback != null) { dispatch(command, defaultCommandCallback); return; } LoggingCallback loggingCallback = LoggingCallback.INSTANCE; if (NoOpMessageMonitor.INSTANCE.equals(messageMonitor)) { CommandMessage<? extends C> interceptedCommand = intercept(command); Optional<Member> optionalDestination = commandRouter.findDestination(interceptedCommand); if (optionalDestination.isPresent()) { Member destination = optionalDestination.get(); try { connector.send(destination, interceptedCommand); } catch (Exception e) { destination.suspect(); loggingCallback.onResult(interceptedCommand, asCommandResultMessage( new CommandDispatchException(DISPATCH_ERROR_MESSAGE + ": " + e.getMessage(), e) )); } } else { loggingCallback.onResult(interceptedCommand, asCommandResultMessage(new NoHandlerForCommandException( format("No node known to accept [%s]", interceptedCommand.getCommandName()) ))); } } else { dispatch(command, loggingCallback); } } protected  DistributedCommandBus(Builder builder); static Builder builder(); @ShutdownHandler(phase = Phase.INBOUND_COMMAND_CONNECTOR) void disconnect(); @ShutdownHandler(phase = Phase.OUTBOUND_COMMAND_CONNECTORS) CompletableFuture<Void> shutdownDispatching(); @Override void dispatch(CommandMessage<C> command); @Override void dispatch(CommandMessage<C> command, CommandCallback<? super C, ? super R> callback); @Override Registration subscribe(String commandName, MessageHandler<? super CommandMessage<?>> handler); @Override CommandBus localSegment(); int getLoadFactor(); void updateLoadFactor(int loadFactor); Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super CommandMessage<?>> dispatchInterceptor); @Override Registration registerHandlerInterceptor(
            MessageHandlerInterceptor<? super CommandMessage<?>> handlerInterceptor); static final int INITIAL_LOAD_FACTOR; }