@Test public void queryResponse() { AtomicInteger dispatchCalled = new AtomicInteger(0); AtomicBoolean doneCalled = new AtomicBoolean(false); queryCache.put("1234", new QueryInformation("1234", "Source", new QueryDefinition("c", "q"), Collections.singleton("client"), 2, r -> dispatchCalled.incrementAndGet(), (client) -> doneCalled.set(true))); testSubject.handleResponse(QueryResponse.newBuilder() .setMessageIdentifier("12345") .setRequestIdentifier("1234") .build(), "client", "clientId",false); assertEquals(1, dispatchCalled.get()); assertFalse(doneCalled.get()); testSubject.handleResponse(QueryResponse.newBuilder() .setMessageIdentifier("1234") .setRequestIdentifier("1234") .build(), "client", "clientId",false); assertEquals(2, dispatchCalled.get()); assertTrue(doneCalled.get()); }
public void handleResponse(QueryResponse queryResponse, String clientStreamId, String clientId, boolean proxied) { String requestIdentifier = queryResponse.getRequestIdentifier(); QueryInformation queryInformation = getQueryInformation(clientStreamId, requestIdentifier); if (queryInformation != null) { ClientStreamIdentification clientStream = new ClientStreamIdentification(queryInformation.getContext(), clientStreamId); if (queryInformation.forward(clientStreamId, queryResponse) <= 0) { queryCache.remove(queryInformation.getKey()); if (!proxied) { queryMetricsRegistry.add(queryInformation.getQuery(), queryInformation.getSourceClientId(), clientId, clientStream.getContext(), System.currentTimeMillis() - queryInformation.getTimestamp()); } } } else { logger.debug("No (more) information for {}", queryResponse.getRequestIdentifier()); } }
QueryDispatcher { public void handleResponse(QueryResponse queryResponse, String clientStreamId, String clientId, boolean proxied) { String requestIdentifier = queryResponse.getRequestIdentifier(); QueryInformation queryInformation = getQueryInformation(clientStreamId, requestIdentifier); if (queryInformation != null) { ClientStreamIdentification clientStream = new ClientStreamIdentification(queryInformation.getContext(), clientStreamId); if (queryInformation.forward(clientStreamId, queryResponse) <= 0) { queryCache.remove(queryInformation.getKey()); if (!proxied) { queryMetricsRegistry.add(queryInformation.getQuery(), queryInformation.getSourceClientId(), clientId, clientStream.getContext(), System.currentTimeMillis() - queryInformation.getTimestamp()); } } } else { logger.debug("No (more) information for {}", queryResponse.getRequestIdentifier()); } } }
QueryDispatcher { public void handleResponse(QueryResponse queryResponse, String clientStreamId, String clientId, boolean proxied) { String requestIdentifier = queryResponse.getRequestIdentifier(); QueryInformation queryInformation = getQueryInformation(clientStreamId, requestIdentifier); if (queryInformation != null) { ClientStreamIdentification clientStream = new ClientStreamIdentification(queryInformation.getContext(), clientStreamId); if (queryInformation.forward(clientStreamId, queryResponse) <= 0) { queryCache.remove(queryInformation.getKey()); if (!proxied) { queryMetricsRegistry.add(queryInformation.getQuery(), queryInformation.getSourceClientId(), clientId, clientStream.getContext(), System.currentTimeMillis() - queryInformation.getTimestamp()); } } } else { logger.debug("No (more) information for {}", queryResponse.getRequestIdentifier()); } } QueryDispatcher(QueryRegistrationCache registrationCache, QueryCache queryCache,
                           QueryMetricsRegistry queryMetricsRegistry,
                           MeterFactory meterFactory,
                           @Value("${axoniq.axonserver.query-queue-capacity-per-client:10000}") int queueCapacity); }
QueryDispatcher { public void handleResponse(QueryResponse queryResponse, String clientStreamId, String clientId, boolean proxied) { String requestIdentifier = queryResponse.getRequestIdentifier(); QueryInformation queryInformation = getQueryInformation(clientStreamId, requestIdentifier); if (queryInformation != null) { ClientStreamIdentification clientStream = new ClientStreamIdentification(queryInformation.getContext(), clientStreamId); if (queryInformation.forward(clientStreamId, queryResponse) <= 0) { queryCache.remove(queryInformation.getKey()); if (!proxied) { queryMetricsRegistry.add(queryInformation.getQuery(), queryInformation.getSourceClientId(), clientId, clientStream.getContext(), System.currentTimeMillis() - queryInformation.getTimestamp()); } } } else { logger.debug("No (more) information for {}", queryResponse.getRequestIdentifier()); } } 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 handleResponse(QueryResponse queryResponse, String clientStreamId, String clientId, boolean proxied) { String requestIdentifier = queryResponse.getRequestIdentifier(); QueryInformation queryInformation = getQueryInformation(clientStreamId, requestIdentifier); if (queryInformation != null) { ClientStreamIdentification clientStream = new ClientStreamIdentification(queryInformation.getContext(), clientStreamId); if (queryInformation.forward(clientStreamId, queryResponse) <= 0) { queryCache.remove(queryInformation.getKey()); if (!proxied) { queryMetricsRegistry.add(queryInformation.getQuery(), queryInformation.getSourceClientId(), clientId, clientStream.getContext(), System.currentTimeMillis() - queryInformation.getTimestamp()); } } } else { logger.debug("No (more) information for {}", queryResponse.getRequestIdentifier()); } } 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);  }