@Test public void dispatchProxied() { QueryRequest request = QueryRequest.newBuilder() .setQuery("test") .setMessageIdentifier("1234") .build(); FakeStreamObserver<QueryProviderInbound> FakeStreamObserver = new FakeStreamObserver<>(); SerializedQuery forwardedQuery = new SerializedQuery(Topology.DEFAULT_CONTEXT, "client", request); QueryHandler handler = new DirectQueryHandler(FakeStreamObserver, new ClientStreamIdentification(Topology.DEFAULT_CONTEXT, "client"), "componentName", "client"); when(registrationCache.find(any(), anyObject(), anyObject())).thenReturn(handler); testSubject.dispatchProxied(forwardedQuery, r -> { }, s -> { }); }
public void dispatchProxied(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); String context = serializedQuery.context(); String clientId = serializedQuery.clientStreamId(); QueryHandler queryHandler = registrationCache.find(context, query, clientId); if (queryHandler == null) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.CLIENT_DISCONNECTED.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage( ErrorMessageFactory .build(String.format("Client %s not found while processing: %s" , clientId, query.getQuery()))) .build()); onCompleted.accept(clientId); } else { QueryDefinition queryDefinition = new QueryDefinition(context, query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } String key = query.getMessageIdentifier() + "/" + serializedQuery.clientStreamId(); QueryInformation queryInformation = new QueryInformation(key, serializedQuery.query().getClientId(), queryDefinition, singleton(queryHandler.getClientStreamId()), expectedResults, callback, onCompleted); try { queryCache.put(key, queryInformation); dispatchOne(queryHandler, serializedQuery, timeout); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { queryInformation.completeWithError(queryHandler.getClientId(), ErrorCode.QUERY_DISPATCH_ERROR, insufficientCacheCapacityException.getMessage()); } } }
QueryDispatcher { public void dispatchProxied(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); String context = serializedQuery.context(); String clientId = serializedQuery.clientStreamId(); QueryHandler queryHandler = registrationCache.find(context, query, clientId); if (queryHandler == null) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.CLIENT_DISCONNECTED.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage( ErrorMessageFactory .build(String.format("Client %s not found while processing: %s" , clientId, query.getQuery()))) .build()); onCompleted.accept(clientId); } else { QueryDefinition queryDefinition = new QueryDefinition(context, query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } String key = query.getMessageIdentifier() + "/" + serializedQuery.clientStreamId(); QueryInformation queryInformation = new QueryInformation(key, serializedQuery.query().getClientId(), queryDefinition, singleton(queryHandler.getClientStreamId()), expectedResults, callback, onCompleted); try { queryCache.put(key, queryInformation); dispatchOne(queryHandler, serializedQuery, timeout); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { queryInformation.completeWithError(queryHandler.getClientId(), ErrorCode.QUERY_DISPATCH_ERROR, insufficientCacheCapacityException.getMessage()); } } } }
QueryDispatcher { public void dispatchProxied(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); String context = serializedQuery.context(); String clientId = serializedQuery.clientStreamId(); QueryHandler queryHandler = registrationCache.find(context, query, clientId); if (queryHandler == null) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.CLIENT_DISCONNECTED.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage( ErrorMessageFactory .build(String.format("Client %s not found while processing: %s" , clientId, query.getQuery()))) .build()); onCompleted.accept(clientId); } else { QueryDefinition queryDefinition = new QueryDefinition(context, query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } String key = query.getMessageIdentifier() + "/" + serializedQuery.clientStreamId(); QueryInformation queryInformation = new QueryInformation(key, serializedQuery.query().getClientId(), queryDefinition, singleton(queryHandler.getClientStreamId()), expectedResults, callback, onCompleted); try { queryCache.put(key, queryInformation); dispatchOne(queryHandler, serializedQuery, timeout); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { queryInformation.completeWithError(queryHandler.getClientId(), ErrorCode.QUERY_DISPATCH_ERROR, insufficientCacheCapacityException.getMessage()); } } } QueryDispatcher(QueryRegistrationCache registrationCache, QueryCache queryCache,
                           QueryMetricsRegistry queryMetricsRegistry,
                           MeterFactory meterFactory,
                           @Value("${axoniq.axonserver.query-queue-capacity-per-client:10000}") int queueCapacity); }
QueryDispatcher { public void dispatchProxied(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); String context = serializedQuery.context(); String clientId = serializedQuery.clientStreamId(); QueryHandler queryHandler = registrationCache.find(context, query, clientId); if (queryHandler == null) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.CLIENT_DISCONNECTED.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage( ErrorMessageFactory .build(String.format("Client %s not found while processing: %s" , clientId, query.getQuery()))) .build()); onCompleted.accept(clientId); } else { QueryDefinition queryDefinition = new QueryDefinition(context, query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } String key = query.getMessageIdentifier() + "/" + serializedQuery.clientStreamId(); QueryInformation queryInformation = new QueryInformation(key, serializedQuery.query().getClientId(), queryDefinition, singleton(queryHandler.getClientStreamId()), expectedResults, callback, onCompleted); try { queryCache.put(key, queryInformation); dispatchOne(queryHandler, serializedQuery, timeout); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { queryInformation.completeWithError(queryHandler.getClientId(), ErrorCode.QUERY_DISPATCH_ERROR, insufficientCacheCapacityException.getMessage()); } } } QueryDispatcher(QueryRegistrationCache registrationCache, QueryCache queryCache,
                           QueryMetricsRegistry queryMetricsRegistry,
                           MeterFactory meterFactory,
                           @Value("${axoniq.axonserver.query-queue-capacity-per-client:10000}") int queueCapacity); void handleResponse(QueryResponse queryResponse,
                               String clientStreamId,
                               String clientId,
                               boolean proxied); void handleComplete(String requestId, String clientStreamId, String clientId, boolean proxied); @EventListener void on(TopologyEvents.QueryHandlerDisconnected event); void removeFromCache(String client, String messageId); FlowControlQueues<WrappedQuery> getQueryQueue(); void query(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted); MeterFactory.RateMeter queryRate(String context); void dispatchProxied(SerializedQuery serializedQuery, Consumer<QueryResponse> callback,
                                Consumer<String> onCompleted); }
QueryDispatcher { public void dispatchProxied(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); String context = serializedQuery.context(); String clientId = serializedQuery.clientStreamId(); QueryHandler queryHandler = registrationCache.find(context, query, clientId); if (queryHandler == null) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.CLIENT_DISCONNECTED.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage( ErrorMessageFactory .build(String.format("Client %s not found while processing: %s" , clientId, query.getQuery()))) .build()); onCompleted.accept(clientId); } else { QueryDefinition queryDefinition = new QueryDefinition(context, query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } String key = query.getMessageIdentifier() + "/" + serializedQuery.clientStreamId(); QueryInformation queryInformation = new QueryInformation(key, serializedQuery.query().getClientId(), queryDefinition, singleton(queryHandler.getClientStreamId()), expectedResults, callback, onCompleted); try { queryCache.put(key, queryInformation); dispatchOne(queryHandler, serializedQuery, timeout); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { queryInformation.completeWithError(queryHandler.getClientId(), ErrorCode.QUERY_DISPATCH_ERROR, insufficientCacheCapacityException.getMessage()); } } } QueryDispatcher(QueryRegistrationCache registrationCache, QueryCache queryCache,
                           QueryMetricsRegistry queryMetricsRegistry,
                           MeterFactory meterFactory,
                           @Value("${axoniq.axonserver.query-queue-capacity-per-client:10000}") int queueCapacity); void handleResponse(QueryResponse queryResponse,
                               String clientStreamId,
                               String clientId,
                               boolean proxied); void handleComplete(String requestId, String clientStreamId, String clientId, boolean proxied); @EventListener void on(TopologyEvents.QueryHandlerDisconnected event); void removeFromCache(String client, String messageId); FlowControlQueues<WrappedQuery> getQueryQueue(); void query(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted); MeterFactory.RateMeter queryRate(String context); void dispatchProxied(SerializedQuery serializedQuery, Consumer<QueryResponse> callback,
                                Consumer<String> onCompleted);  }