@Test void scatterGather() { QueryMessage<String, String> testQuery = new GenericQueryMessage<>("Hello, World", instanceOf(String.class)); when(mockQueryChannel.query(any())).thenReturn(new StubResultStream(stubResponse("<string>1</string>"), stubResponse("<string>2</string>"), stubResponse("<string>3</string>"))); assertEquals(3, testSubject.scatterGather(testQuery, 12, TimeUnit.SECONDS).count()); verify(targetContextResolver).resolveContext(testQuery); verify(mockQueryChannel).query(argThat( r -> r.getPayload().getData().toStringUtf8().equals("<string>Hello, World</string>") && -1 == ProcessingInstructionHelper.numberOfResults(r.getProcessingInstructionsList()))); }
@Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage, long timeout, TimeUnit timeUnit) { shutdownLatch.ifShuttingDown(String.format( "Cannot dispatch new %s as this bus is being shut down", "scatter-gather queries" )); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, SCATTER_GATHER_NUMBER_OF_RESULTS, timeUnit.toMillis(timeout), priorityCalculator.determinePriority(interceptedQuery)); ResultStream<QueryResponse> queryResult = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); return StreamSupport.stream( new QueryResponseSpliterator<>(queryMessage, queryResult, deadline, serializer), false ); } catch (Exception e) { logger.debug("There was a problem issuing a scatter-gather query {}.", interceptedQuery, e); queryInTransit.end(); throw e; } }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage, long timeout, TimeUnit timeUnit) { shutdownLatch.ifShuttingDown(String.format( "Cannot dispatch new %s as this bus is being shut down", "scatter-gather queries" )); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, SCATTER_GATHER_NUMBER_OF_RESULTS, timeUnit.toMillis(timeout), priorityCalculator.determinePriority(interceptedQuery)); ResultStream<QueryResponse> queryResult = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); return StreamSupport.stream( new QueryResponseSpliterator<>(queryMessage, queryResult, deadline, serializer), false ); } catch (Exception e) { logger.debug("There was a problem issuing a scatter-gather query {}.", interceptedQuery, e); queryInTransit.end(); throw e; } } }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage, long timeout, TimeUnit timeUnit) { shutdownLatch.ifShuttingDown(String.format( "Cannot dispatch new %s as this bus is being shut down", "scatter-gather queries" )); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, SCATTER_GATHER_NUMBER_OF_RESULTS, timeUnit.toMillis(timeout), priorityCalculator.determinePriority(interceptedQuery)); ResultStream<QueryResponse> queryResult = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); return StreamSupport.stream( new QueryResponseSpliterator<>(queryMessage, queryResult, deadline, serializer), false ); } catch (Exception e) { logger.debug("There was a problem issuing a scatter-gather query {}.", interceptedQuery, e); queryInTransit.end(); throw e; } } AxonServerQueryBus(Builder builder); }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage, long timeout, TimeUnit timeUnit) { shutdownLatch.ifShuttingDown(String.format( "Cannot dispatch new %s as this bus is being shut down", "scatter-gather queries" )); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, SCATTER_GATHER_NUMBER_OF_RESULTS, timeUnit.toMillis(timeout), priorityCalculator.determinePriority(interceptedQuery)); ResultStream<QueryResponse> queryResult = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); return StreamSupport.stream( new QueryResponseSpliterator<>(queryMessage, queryResult, deadline, serializer), false ); } catch (Exception e) { logger.debug("There was a problem issuing a scatter-gather query {}.", interceptedQuery, e); queryInTransit.end(); throw e; } } 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> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage, long timeout, TimeUnit timeUnit) { shutdownLatch.ifShuttingDown(String.format( "Cannot dispatch new %s as this bus is being shut down", "scatter-gather queries" )); QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage); ShutdownLatch.ActivityHandle queryInTransit = shutdownLatch.registerActivity(); long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { String targetContext = targetContextResolver.resolveContext(interceptedQuery); QueryRequest queryRequest = serializer.serializeRequest(interceptedQuery, SCATTER_GATHER_NUMBER_OF_RESULTS, timeUnit.toMillis(timeout), priorityCalculator.determinePriority(interceptedQuery)); ResultStream<QueryResponse> queryResult = axonServerConnectionManager.getConnection(targetContext) .queryChannel() .query(queryRequest); return StreamSupport.stream( new QueryResponseSpliterator<>(queryMessage, queryResult, deadline, serializer), false ); } catch (Exception e) { logger.debug("There was a problem issuing a scatter-gather query {}.", interceptedQuery, e); queryInTransit.end(); throw e; } } 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();  }