@Test void queryReportsDispatchException() throws Exception { StubResultStream t = new StubResultStream(new RuntimeException("Faking problems")); when(mockQueryChannel.query(any())).thenReturn(t); QueryMessage<String, String> testQuery = new GenericQueryMessage<>("Hello, World", instanceOf(String.class)); CompletableFuture<QueryResponseMessage<String>> result = testSubject.query(testQuery); try { result.get(); fail("Expected exception"); } catch (ExecutionException e) { assertTrue(e.getCause() instanceof AxonServerQueryDispatchException); assertEquals("Faking problems", e.getCause().getMessage()); } verify(targetContextResolver).resolveContext(testQuery); }
@Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage) { shutdownLatch.ifShuttingDown(String.format("Cannot dispatch new %s as this bus is being shut down", "queries")); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); CompletableFuture<QueryResponseMessage<R>> queryTransaction = new CompletableFuture<>(); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); int priority = priorityCalculator.determinePriority(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, DIRECT_QUERY_NUMBER_OF_RESULTS, DIRECT_QUERY_TIMEOUT_MS, priority); ResultStream<QueryResponse> result = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); ResponseProcessingTask<R> responseProcessingTask = new ResponseProcessingTask<>( result, serializer, queryTransaction, priority, queryMessage.getResponseType() ); result.onAvailable(() -> queryExecutor.submit(responseProcessingTask)); } catch (Exception e) { logger.debug("There was a problem issuing a query {}.", interceptedQuery, e); AxonException exception = ErrorCode.QUERY_DISPATCH_ERROR.convert(configuration.getClientId(), e); queryTransaction.completeExceptionally(exception); queryInTransit.end(); } return queryTransaction; }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage) { shutdownLatch.ifShuttingDown(String.format("Cannot dispatch new %s as this bus is being shut down", "queries")); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); CompletableFuture<QueryResponseMessage<R>> queryTransaction = new CompletableFuture<>(); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); int priority = priorityCalculator.determinePriority(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, DIRECT_QUERY_NUMBER_OF_RESULTS, DIRECT_QUERY_TIMEOUT_MS, priority); ResultStream<QueryResponse> result = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); ResponseProcessingTask<R> responseProcessingTask = new ResponseProcessingTask<>( result, serializer, queryTransaction, priority, queryMessage.getResponseType() ); result.onAvailable(() -> queryExecutor.submit(responseProcessingTask)); } catch (Exception e) { logger.debug("There was a problem issuing a query {}.", interceptedQuery, e); AxonException exception = ErrorCode.QUERY_DISPATCH_ERROR.convert(configuration.getClientId(), e); queryTransaction.completeExceptionally(exception); queryInTransit.end(); } return queryTransaction; } }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage) { shutdownLatch.ifShuttingDown(String.format("Cannot dispatch new %s as this bus is being shut down", "queries")); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); CompletableFuture<QueryResponseMessage<R>> queryTransaction = new CompletableFuture<>(); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); int priority = priorityCalculator.determinePriority(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, DIRECT_QUERY_NUMBER_OF_RESULTS, DIRECT_QUERY_TIMEOUT_MS, priority); ResultStream<QueryResponse> result = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); ResponseProcessingTask<R> responseProcessingTask = new ResponseProcessingTask<>( result, serializer, queryTransaction, priority, queryMessage.getResponseType() ); result.onAvailable(() -> queryExecutor.submit(responseProcessingTask)); } catch (Exception e) { logger.debug("There was a problem issuing a query {}.", interceptedQuery, e); AxonException exception = ErrorCode.QUERY_DISPATCH_ERROR.convert(configuration.getClientId(), e); queryTransaction.completeExceptionally(exception); queryInTransit.end(); } return queryTransaction; } AxonServerQueryBus(Builder builder); }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage) { shutdownLatch.ifShuttingDown(String.format("Cannot dispatch new %s as this bus is being shut down", "queries")); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); CompletableFuture<QueryResponseMessage<R>> queryTransaction = new CompletableFuture<>(); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); int priority = priorityCalculator.determinePriority(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, DIRECT_QUERY_NUMBER_OF_RESULTS, DIRECT_QUERY_TIMEOUT_MS, priority); ResultStream<QueryResponse> result = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); ResponseProcessingTask<R> responseProcessingTask = new ResponseProcessingTask<>( result, serializer, queryTransaction, priority, queryMessage.getResponseType() ); result.onAvailable(() -> queryExecutor.submit(responseProcessingTask)); } catch (Exception e) { logger.debug("There was a problem issuing a query {}.", interceptedQuery, e); AxonException exception = ErrorCode.QUERY_DISPATCH_ERROR.convert(configuration.getClientId(), e); queryTransaction.completeExceptionally(exception); queryInTransit.end(); } return queryTransaction; } AxonServerQueryBus(Builder builder); static Builder builder(); @StartHandler(phase = Phase.INBOUND_QUERY_CONNECTOR) void start(); @Override Registration subscribe(String queryName,
                                      Type responseType,
                                      MessageHandler<? super QueryMessage<?, R>> handler); @Override CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage); @Override Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage,
                                                                long timeout,
                                                                TimeUnit timeUnit); @Override SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> subscriptionQuery(
            SubscriptionQueryMessage<Q, I, U> query,
            SubscriptionQueryBackpressure backPressure,
            int updateBufferSize
    ); @Override QueryUpdateEmitter queryUpdateEmitter(); @Override QueryBus localSegment(); @Override Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?, ?>> interceptor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> dispatchInterceptor); @ShutdownHandler(phase = Phase.INBOUND_QUERY_CONNECTOR) void disconnect(); @ShutdownHandler(phase = Phase.OUTBOUND_QUERY_CONNECTORS) CompletableFuture<Void> shutdownDispatching(); }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage) { shutdownLatch.ifShuttingDown(String.format("Cannot dispatch new %s as this bus is being shut down", "queries")); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); CompletableFuture<QueryResponseMessage<R>> queryTransaction = new CompletableFuture<>(); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); int priority = priorityCalculator.determinePriority(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, DIRECT_QUERY_NUMBER_OF_RESULTS, DIRECT_QUERY_TIMEOUT_MS, priority); ResultStream<QueryResponse> result = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); ResponseProcessingTask<R> responseProcessingTask = new ResponseProcessingTask<>( result, serializer, queryTransaction, priority, queryMessage.getResponseType() ); result.onAvailable(() -> queryExecutor.submit(responseProcessingTask)); } catch (Exception e) { logger.debug("There was a problem issuing a query {}.", interceptedQuery, e); AxonException exception = ErrorCode.QUERY_DISPATCH_ERROR.convert(configuration.getClientId(), e); queryTransaction.completeExceptionally(exception); queryInTransit.end(); } return queryTransaction; } AxonServerQueryBus(Builder builder); static Builder builder(); @StartHandler(phase = Phase.INBOUND_QUERY_CONNECTOR) void start(); @Override Registration subscribe(String queryName,
                                      Type responseType,
                                      MessageHandler<? super QueryMessage<?, R>> handler); @Override CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage); @Override Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage,
                                                                long timeout,
                                                                TimeUnit timeUnit); @Override SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> subscriptionQuery(
            SubscriptionQueryMessage<Q, I, U> query,
            SubscriptionQueryBackpressure backPressure,
            int updateBufferSize
    ); @Override QueryUpdateEmitter queryUpdateEmitter(); @Override QueryBus localSegment(); @Override Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?, ?>> interceptor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> dispatchInterceptor); @ShutdownHandler(phase = Phase.INBOUND_QUERY_CONNECTOR) void disconnect(); @ShutdownHandler(phase = Phase.OUTBOUND_QUERY_CONNECTORS) CompletableFuture<Void> shutdownDispatching();  }