@Test void testScatterGatherQuery() { long expectedTimeout = 1L; TimeUnit expectedTimeUnit = TimeUnit.SECONDS; when(mockBus.scatterGather(anyMessage(String.class, String.class), anyLong(), any())) .thenReturn(Stream.of(answer)); Stream<String> queryResponse = testSubject.scatterGather("scatterGather", instanceOf(String.class), expectedTimeout, expectedTimeUnit); Optional<String> firstResult = queryResponse.findFirst(); assertTrue(firstResult.isPresent()); assertEquals("answer", firstResult.get()); ArgumentCaptor<QueryMessage<String, String>> queryMessageCaptor = ArgumentCaptor.forClass(QueryMessage.class); verify(mockBus).scatterGather(queryMessageCaptor.capture(), eq(expectedTimeout), eq(expectedTimeUnit)); QueryMessage<String, String> result = queryMessageCaptor.getValue(); assertEquals("scatterGather", result.getPayload()); assertEquals(String.class, result.getPayloadType()); assertEquals(String.class.getName(), result.getQueryName()); assertTrue(InstanceResponseType.class.isAssignableFrom(result.getResponseType().getClass())); assertEquals(String.class, result.getResponseType().getExpectedResponseType()); assertEquals(MetaData.emptyInstance(), result.getMetaData()); }
@Override public <R, Q> Stream<R> scatterGather(String queryName, Q query, ResponseType<R> responseType, long timeout, TimeUnit timeUnit) { GenericQueryMessage<?, R> queryMessage = new GenericQueryMessage<>(asMessage(query), queryName, responseType); return queryBus.scatterGather(processInterceptors(queryMessage), timeout, timeUnit) .map(QueryResponseMessage::getPayload); }
DefaultQueryGateway implements QueryGateway { @Override public <R, Q> Stream<R> scatterGather(String queryName, Q query, ResponseType<R> responseType, long timeout, TimeUnit timeUnit) { GenericQueryMessage<?, R> queryMessage = new GenericQueryMessage<>(asMessage(query), queryName, responseType); return queryBus.scatterGather(processInterceptors(queryMessage), timeout, timeUnit) .map(QueryResponseMessage::getPayload); } }
DefaultQueryGateway implements QueryGateway { @Override public <R, Q> Stream<R> scatterGather(String queryName, Q query, ResponseType<R> responseType, long timeout, TimeUnit timeUnit) { GenericQueryMessage<?, R> queryMessage = new GenericQueryMessage<>(asMessage(query), queryName, responseType); return queryBus.scatterGather(processInterceptors(queryMessage), timeout, timeUnit) .map(QueryResponseMessage::getPayload); } protected  DefaultQueryGateway(Builder builder); }
DefaultQueryGateway implements QueryGateway { @Override public <R, Q> Stream<R> scatterGather(String queryName, Q query, ResponseType<R> responseType, long timeout, TimeUnit timeUnit) { GenericQueryMessage<?, R> queryMessage = new GenericQueryMessage<>(asMessage(query), queryName, responseType); return queryBus.scatterGather(processInterceptors(queryMessage), timeout, timeUnit) .map(QueryResponseMessage::getPayload); } protected  DefaultQueryGateway(Builder builder); static Builder builder(); @Override CompletableFuture<R> query(String queryName, Q query, ResponseType<R> responseType); @Override Stream<R> scatterGather(String queryName,
                                          Q query,
                                          ResponseType<R> responseType,
                                          long timeout,
                                          TimeUnit timeUnit); @Override SubscriptionQueryResult<I, U> subscriptionQuery(String queryName,
                                                                     Q query,
                                                                     ResponseType<I> initialResponseType,
                                                                     ResponseType<U> updateResponseType,
                                                                     SubscriptionQueryBackpressure backpressure,
                                                                     int updateBufferSize); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> interceptor); }
DefaultQueryGateway implements QueryGateway { @Override public <R, Q> Stream<R> scatterGather(String queryName, Q query, ResponseType<R> responseType, long timeout, TimeUnit timeUnit) { GenericQueryMessage<?, R> queryMessage = new GenericQueryMessage<>(asMessage(query), queryName, responseType); return queryBus.scatterGather(processInterceptors(queryMessage), timeout, timeUnit) .map(QueryResponseMessage::getPayload); } protected  DefaultQueryGateway(Builder builder); static Builder builder(); @Override CompletableFuture<R> query(String queryName, Q query, ResponseType<R> responseType); @Override Stream<R> scatterGather(String queryName,
                                          Q query,
                                          ResponseType<R> responseType,
                                          long timeout,
                                          TimeUnit timeUnit); @Override SubscriptionQueryResult<I, U> subscriptionQuery(String queryName,
                                                                     Q query,
                                                                     ResponseType<I> initialResponseType,
                                                                     ResponseType<U> updateResponseType,
                                                                     SubscriptionQueryBackpressure backpressure,
                                                                     int updateBufferSize); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> interceptor);  }