@Test public void dispatchQueueFull() { commandDispatcher = new CommandDispatcher(registrations, commandCache, metricsRegistry, meterFactory, 0); FakeStreamObserver<SerializedCommandResponse> responseObserver = new FakeStreamObserver<>(); Command request = Command.newBuilder() .addProcessingInstructions(ProcessingInstructionHelper.routingKey("1234")) .setName("Command") .setMessageIdentifier("12") .build(); FakeStreamObserver<SerializedCommandProviderInbound> commandProviderInbound = new FakeStreamObserver<>(); ClientStreamIdentification client = new ClientStreamIdentification(Topology.DEFAULT_CONTEXT, "client"); DirectCommandHandler result = new DirectCommandHandler(commandProviderInbound, client, "client", "component"); when(registrations.getHandlerForCommand(any(), anyObject(), anyObject())).thenReturn(result); commandDispatcher.dispatch(Topology.DEFAULT_CONTEXT, new SerializedCommand(request), response -> { responseObserver.onNext(response); responseObserver.onCompleted(); }, false); assertEquals(1, responseObserver.values().size()); assertNotEquals("", responseObserver.values().get(0).getErrorCode()); }
public void dispatch(String context, SerializedCommand request, Consumer<SerializedCommandResponse> responseObserver, boolean proxied) { if (proxied) { String clientStreamId = request.getClientStreamId(); ClientStreamIdentification clientIdentification = new ClientStreamIdentification(context, clientStreamId); CommandHandler<?> handler = registrations.findByClientAndCommand(clientIdentification, request.getCommand()); dispatchToCommandHandler(request, handler, responseObserver, ErrorCode.CLIENT_DISCONNECTED, String.format("Client %s not found while processing: %s" , clientStreamId, request.getCommand())); } else { commandRate(context).mark(); CommandHandler<?> commandHandler = registrations.getHandlerForCommand(context, request.wrapped(), request.getRoutingKey()); dispatchToCommandHandler(request, commandHandler, responseObserver, ErrorCode.NO_HANDLER_FOR_COMMAND, "No Handler for command: " + request.getCommand() ); } }
CommandDispatcher { public void dispatch(String context, SerializedCommand request, Consumer<SerializedCommandResponse> responseObserver, boolean proxied) { if (proxied) { String clientStreamId = request.getClientStreamId(); ClientStreamIdentification clientIdentification = new ClientStreamIdentification(context, clientStreamId); CommandHandler<?> handler = registrations.findByClientAndCommand(clientIdentification, request.getCommand()); dispatchToCommandHandler(request, handler, responseObserver, ErrorCode.CLIENT_DISCONNECTED, String.format("Client %s not found while processing: %s" , clientStreamId, request.getCommand())); } else { commandRate(context).mark(); CommandHandler<?> commandHandler = registrations.getHandlerForCommand(context, request.wrapped(), request.getRoutingKey()); dispatchToCommandHandler(request, commandHandler, responseObserver, ErrorCode.NO_HANDLER_FOR_COMMAND, "No Handler for command: " + request.getCommand() ); } } }
CommandDispatcher { public void dispatch(String context, SerializedCommand request, Consumer<SerializedCommandResponse> responseObserver, boolean proxied) { if (proxied) { String clientStreamId = request.getClientStreamId(); ClientStreamIdentification clientIdentification = new ClientStreamIdentification(context, clientStreamId); CommandHandler<?> handler = registrations.findByClientAndCommand(clientIdentification, request.getCommand()); dispatchToCommandHandler(request, handler, responseObserver, ErrorCode.CLIENT_DISCONNECTED, String.format("Client %s not found while processing: %s" , clientStreamId, request.getCommand())); } else { commandRate(context).mark(); CommandHandler<?> commandHandler = registrations.getHandlerForCommand(context, request.wrapped(), request.getRoutingKey()); dispatchToCommandHandler(request, commandHandler, responseObserver, ErrorCode.NO_HANDLER_FOR_COMMAND, "No Handler for command: " + request.getCommand() ); } } CommandDispatcher(CommandRegistrationCache registrations, CommandCache commandCache,
                             CommandMetricsRegistry metricRegistry,
                             MeterFactory meterFactory,
                             @Value("${axoniq.axonserver.command-queue-capacity-per-client:10000}") int queueCapacity); }
CommandDispatcher { public void dispatch(String context, SerializedCommand request, Consumer<SerializedCommandResponse> responseObserver, boolean proxied) { if (proxied) { String clientStreamId = request.getClientStreamId(); ClientStreamIdentification clientIdentification = new ClientStreamIdentification(context, clientStreamId); CommandHandler<?> handler = registrations.findByClientAndCommand(clientIdentification, request.getCommand()); dispatchToCommandHandler(request, handler, responseObserver, ErrorCode.CLIENT_DISCONNECTED, String.format("Client %s not found while processing: %s" , clientStreamId, request.getCommand())); } else { commandRate(context).mark(); CommandHandler<?> commandHandler = registrations.getHandlerForCommand(context, request.wrapped(), request.getRoutingKey()); dispatchToCommandHandler(request, commandHandler, responseObserver, ErrorCode.NO_HANDLER_FOR_COMMAND, "No Handler for command: " + request.getCommand() ); } } 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 dispatch(String context, SerializedCommand request, Consumer<SerializedCommandResponse> responseObserver, boolean proxied) { if (proxied) { String clientStreamId = request.getClientStreamId(); ClientStreamIdentification clientIdentification = new ClientStreamIdentification(context, clientStreamId); CommandHandler<?> handler = registrations.findByClientAndCommand(clientIdentification, request.getCommand()); dispatchToCommandHandler(request, handler, responseObserver, ErrorCode.CLIENT_DISCONNECTED, String.format("Client %s not found while processing: %s" , clientStreamId, request.getCommand())); } else { commandRate(context).mark(); CommandHandler<?> commandHandler = registrations.getHandlerForCommand(context, request.wrapped(), request.getRoutingKey()); dispatchToCommandHandler(request, commandHandler, responseObserver, ErrorCode.NO_HANDLER_FOR_COMMAND, "No Handler for command: " + request.getCommand() ); } } 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();  }