@Test public void queryFound() { QueryRequest request = QueryRequest.newBuilder() .setQuery("test") .setMessageIdentifier("1234") .build(); FakeStreamObserver<QueryResponse> responseObserver = new FakeStreamObserver<>(); Set<QueryHandler> handlers = new HashSet<>(); FakeStreamObserver<QueryProviderInbound> dispatchStreamObserver = new FakeStreamObserver<>(); handlers.add(new DirectQueryHandler(dispatchStreamObserver, new ClientStreamIdentification(Topology.DEFAULT_CONTEXT, "client"), "componentName", "client")); when(registrationCache.find(any(), any())).thenReturn(handlers); testSubject.query(new SerializedQuery(Topology.DEFAULT_CONTEXT, request), responseObserver::onNext, client -> responseObserver.onCompleted()); assertEquals(0, responseObserver.values().size()); }
public void query(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { queryRate(serializedQuery.context()).mark(); QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); Set<? extends QueryHandler> handlers = registrationCache.find(serializedQuery.context(), query); if (handlers.isEmpty()) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.NO_HANDLER_FOR_QUERY.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build("No handler for query: " + query.getQuery())) .build()); onCompleted.accept("NoClient"); } else { QueryDefinition queryDefinition = new QueryDefinition(serializedQuery.context(), query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } QueryInformation queryInformation = new QueryInformation(query.getMessageIdentifier(), query.getClientId(), queryDefinition, handlers.stream() .map(QueryHandler::getClientStreamId) .collect(Collectors.toSet()), expectedResults, callback, onCompleted); try { queryCache.put(query.getMessageIdentifier(), queryInformation); handlers.forEach(h -> dispatchOne(h, serializedQuery, timeout)); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.QUERY_DISPATCH_ERROR.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build(insufficientCacheCapacityException .getMessage())) .build()); onCompleted.accept("NoCapacity"); } } }
QueryDispatcher { public void query(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { queryRate(serializedQuery.context()).mark(); QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); Set<? extends QueryHandler> handlers = registrationCache.find(serializedQuery.context(), query); if (handlers.isEmpty()) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.NO_HANDLER_FOR_QUERY.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build("No handler for query: " + query.getQuery())) .build()); onCompleted.accept("NoClient"); } else { QueryDefinition queryDefinition = new QueryDefinition(serializedQuery.context(), query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } QueryInformation queryInformation = new QueryInformation(query.getMessageIdentifier(), query.getClientId(), queryDefinition, handlers.stream() .map(QueryHandler::getClientStreamId) .collect(Collectors.toSet()), expectedResults, callback, onCompleted); try { queryCache.put(query.getMessageIdentifier(), queryInformation); handlers.forEach(h -> dispatchOne(h, serializedQuery, timeout)); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.QUERY_DISPATCH_ERROR.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build(insufficientCacheCapacityException .getMessage())) .build()); onCompleted.accept("NoCapacity"); } } } }
QueryDispatcher { public void query(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { queryRate(serializedQuery.context()).mark(); QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); Set<? extends QueryHandler> handlers = registrationCache.find(serializedQuery.context(), query); if (handlers.isEmpty()) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.NO_HANDLER_FOR_QUERY.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build("No handler for query: " + query.getQuery())) .build()); onCompleted.accept("NoClient"); } else { QueryDefinition queryDefinition = new QueryDefinition(serializedQuery.context(), query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } QueryInformation queryInformation = new QueryInformation(query.getMessageIdentifier(), query.getClientId(), queryDefinition, handlers.stream() .map(QueryHandler::getClientStreamId) .collect(Collectors.toSet()), expectedResults, callback, onCompleted); try { queryCache.put(query.getMessageIdentifier(), queryInformation); handlers.forEach(h -> dispatchOne(h, serializedQuery, timeout)); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.QUERY_DISPATCH_ERROR.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build(insufficientCacheCapacityException .getMessage())) .build()); onCompleted.accept("NoCapacity"); } } } QueryDispatcher(QueryRegistrationCache registrationCache, QueryCache queryCache,
                           QueryMetricsRegistry queryMetricsRegistry,
                           MeterFactory meterFactory,
                           @Value("${axoniq.axonserver.query-queue-capacity-per-client:10000}") int queueCapacity); }
QueryDispatcher { public void query(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { queryRate(serializedQuery.context()).mark(); QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); Set<? extends QueryHandler> handlers = registrationCache.find(serializedQuery.context(), query); if (handlers.isEmpty()) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.NO_HANDLER_FOR_QUERY.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build("No handler for query: " + query.getQuery())) .build()); onCompleted.accept("NoClient"); } else { QueryDefinition queryDefinition = new QueryDefinition(serializedQuery.context(), query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } QueryInformation queryInformation = new QueryInformation(query.getMessageIdentifier(), query.getClientId(), queryDefinition, handlers.stream() .map(QueryHandler::getClientStreamId) .collect(Collectors.toSet()), expectedResults, callback, onCompleted); try { queryCache.put(query.getMessageIdentifier(), queryInformation); handlers.forEach(h -> dispatchOne(h, serializedQuery, timeout)); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.QUERY_DISPATCH_ERROR.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build(insufficientCacheCapacityException .getMessage())) .build()); onCompleted.accept("NoCapacity"); } } } 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 query(SerializedQuery serializedQuery, Consumer<QueryResponse> callback, Consumer<String> onCompleted) { queryRate(serializedQuery.context()).mark(); QueryRequest query = serializedQuery.query(); long timeout = System.currentTimeMillis() + ProcessingInstructionHelper.timeout(query.getProcessingInstructionsList()); Set<? extends QueryHandler> handlers = registrationCache.find(serializedQuery.context(), query); if (handlers.isEmpty()) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.NO_HANDLER_FOR_QUERY.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build("No handler for query: " + query.getQuery())) .build()); onCompleted.accept("NoClient"); } else { QueryDefinition queryDefinition = new QueryDefinition(serializedQuery.context(), query.getQuery()); int expectedResults = Integer.MAX_VALUE; int nrOfResults = ProcessingInstructionHelper.numberOfResults(query.getProcessingInstructionsList()); if (nrOfResults > 0) { expectedResults = nrOfResults; } QueryInformation queryInformation = new QueryInformation(query.getMessageIdentifier(), query.getClientId(), queryDefinition, handlers.stream() .map(QueryHandler::getClientStreamId) .collect(Collectors.toSet()), expectedResults, callback, onCompleted); try { queryCache.put(query.getMessageIdentifier(), queryInformation); handlers.forEach(h -> dispatchOne(h, serializedQuery, timeout)); } catch (InsufficientCacheCapacityException insufficientCacheCapacityException) { callback.accept(QueryResponse.newBuilder() .setErrorCode(ErrorCode.QUERY_DISPATCH_ERROR.getCode()) .setMessageIdentifier(query.getMessageIdentifier()) .setErrorMessage(ErrorMessageFactory .build(insufficientCacheCapacityException .getMessage())) .build()); onCompleted.accept("NoCapacity"); } } } 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);  }