@Test public void handleResponse() { AtomicBoolean responseHandled = new AtomicBoolean(false); ClientStreamIdentification client = new ClientStreamIdentification(Topology.DEFAULT_CONTEXT, "Client"); CommandInformation commandInformation = new CommandInformation("TheCommand", "Source", "Target", (r) -> responseHandled.set(true), client, "Component"); when(commandCache.remove(any(String.class))).thenReturn(commandInformation); commandDispatcher.handleResponse(new SerializedCommandResponse(CommandResponse.newBuilder().build()), false); assertTrue(responseHandled.get()); }
public void handleResponse(SerializedCommandResponse commandResponse, boolean proxied) { CommandInformation toPublisher = commandCache.remove(commandResponse.getRequestIdentifier()); if (toPublisher != null) { logger.debug("Sending response to: {}", toPublisher); if (!proxied) { metricRegistry.add(toPublisher.getRequestIdentifier(), toPublisher.getSourceClientId(), toPublisher.getTargetClientId(), toPublisher.getClientStreamIdentification().getContext(), System.currentTimeMillis() - toPublisher.getTimestamp()); } toPublisher.getResponseConsumer().accept(commandResponse); } else { logger.info("Could not find command request: {}", commandResponse.getRequestIdentifier()); } }
CommandDispatcher { public void handleResponse(SerializedCommandResponse commandResponse, boolean proxied) { CommandInformation toPublisher = commandCache.remove(commandResponse.getRequestIdentifier()); if (toPublisher != null) { logger.debug("Sending response to: {}", toPublisher); if (!proxied) { metricRegistry.add(toPublisher.getRequestIdentifier(), toPublisher.getSourceClientId(), toPublisher.getTargetClientId(), toPublisher.getClientStreamIdentification().getContext(), System.currentTimeMillis() - toPublisher.getTimestamp()); } toPublisher.getResponseConsumer().accept(commandResponse); } else { logger.info("Could not find command request: {}", commandResponse.getRequestIdentifier()); } } }
CommandDispatcher { public void handleResponse(SerializedCommandResponse commandResponse, boolean proxied) { CommandInformation toPublisher = commandCache.remove(commandResponse.getRequestIdentifier()); if (toPublisher != null) { logger.debug("Sending response to: {}", toPublisher); if (!proxied) { metricRegistry.add(toPublisher.getRequestIdentifier(), toPublisher.getSourceClientId(), toPublisher.getTargetClientId(), toPublisher.getClientStreamIdentification().getContext(), System.currentTimeMillis() - toPublisher.getTimestamp()); } toPublisher.getResponseConsumer().accept(commandResponse); } else { logger.info("Could not find command request: {}", commandResponse.getRequestIdentifier()); } } CommandDispatcher(CommandRegistrationCache registrations, CommandCache commandCache,
                             CommandMetricsRegistry metricRegistry,
                             MeterFactory meterFactory,
                             @Value("${axoniq.axonserver.command-queue-capacity-per-client:10000}") int queueCapacity); }
CommandDispatcher { public void handleResponse(SerializedCommandResponse commandResponse, boolean proxied) { CommandInformation toPublisher = commandCache.remove(commandResponse.getRequestIdentifier()); if (toPublisher != null) { logger.debug("Sending response to: {}", toPublisher); if (!proxied) { metricRegistry.add(toPublisher.getRequestIdentifier(), toPublisher.getSourceClientId(), toPublisher.getTargetClientId(), toPublisher.getClientStreamIdentification().getContext(), System.currentTimeMillis() - toPublisher.getTimestamp()); } toPublisher.getResponseConsumer().accept(commandResponse); } else { logger.info("Could not find command request: {}", commandResponse.getRequestIdentifier()); } } CommandDispatcher(CommandRegistrationCache registrations, CommandCache commandCache,
                             CommandMetricsRegistry metricRegistry,
                             MeterFactory meterFactory,
                             @Value("${axoniq.axonserver.command-queue-capacity-per-client:10000}") int queueCapacity); void dispatch(String context, SerializedCommand request,
                         Consumer<SerializedCommandResponse> responseObserver, boolean proxied); MeterFactory.RateMeter commandRate(String context); @EventListener void on(TopologyEvents.CommandHandlerDisconnected event); void handleResponse(SerializedCommandResponse commandResponse, boolean proxied); FlowControlQueues<WrappedCommand> getCommandQueues(); int activeCommandCount(); }
CommandDispatcher { public void handleResponse(SerializedCommandResponse commandResponse, boolean proxied) { CommandInformation toPublisher = commandCache.remove(commandResponse.getRequestIdentifier()); if (toPublisher != null) { logger.debug("Sending response to: {}", toPublisher); if (!proxied) { metricRegistry.add(toPublisher.getRequestIdentifier(), toPublisher.getSourceClientId(), toPublisher.getTargetClientId(), toPublisher.getClientStreamIdentification().getContext(), System.currentTimeMillis() - toPublisher.getTimestamp()); } toPublisher.getResponseConsumer().accept(commandResponse); } else { logger.info("Could not find command request: {}", commandResponse.getRequestIdentifier()); } } CommandDispatcher(CommandRegistrationCache registrations, CommandCache commandCache,
                             CommandMetricsRegistry metricRegistry,
                             MeterFactory meterFactory,
                             @Value("${axoniq.axonserver.command-queue-capacity-per-client:10000}") int queueCapacity); void dispatch(String context, SerializedCommand request,
                         Consumer<SerializedCommandResponse> responseObserver, boolean proxied); MeterFactory.RateMeter commandRate(String context); @EventListener void on(TopologyEvents.CommandHandlerDisconnected event); void handleResponse(SerializedCommandResponse commandResponse, boolean proxied); FlowControlQueues<WrappedCommand> getCommandQueues(); int activeCommandCount();  }