@Test void testScatterGatherReturnsEmptyStreamWhenNoHandlersAvailable() { QueryMessage<String, String> testQueryMessage = new GenericQueryMessage<>("Hello, World", singleStringResponse); Set<Object> allResults = testSubject.scatterGather(testQueryMessage, 0, TimeUnit.SECONDS).collect(toSet()); assertEquals(0, allResults.size()); verify(messageMonitor).onMessageIngested(any()); verify(monitorCallback).reportIgnored(); }
@Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); if (handlers.isEmpty()) { monitorCallback.reportIgnored(); return Stream.empty(); } long deadline = System.currentTimeMillis() + unit.toMillis(timeout); return handlers.stream() .map(handler -> { long leftTimeout = getRemainingOfDeadline(deadline); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(DefaultUnitOfWork.startAndGet(interceptedQuery), handler); QueryResponseMessage<R> response = null; if (resultMessage.isExceptional()) { monitorCallback.reportFailure(resultMessage.exceptionResult()); errorHandler.onError(resultMessage.exceptionResult(), interceptedQuery, handler); } else { try { response = resultMessage.getPayload().get(leftTimeout, TimeUnit.MILLISECONDS); monitorCallback.reportSuccess(); } catch (Exception e) { monitorCallback.reportFailure(e); errorHandler.onError(e, interceptedQuery, handler); } } return response; }).filter(Objects::nonNull); }
SimpleQueryBus implements QueryBus { @Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); if (handlers.isEmpty()) { monitorCallback.reportIgnored(); return Stream.empty(); } long deadline = System.currentTimeMillis() + unit.toMillis(timeout); return handlers.stream() .map(handler -> { long leftTimeout = getRemainingOfDeadline(deadline); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(DefaultUnitOfWork.startAndGet(interceptedQuery), handler); QueryResponseMessage<R> response = null; if (resultMessage.isExceptional()) { monitorCallback.reportFailure(resultMessage.exceptionResult()); errorHandler.onError(resultMessage.exceptionResult(), interceptedQuery, handler); } else { try { response = resultMessage.getPayload().get(leftTimeout, TimeUnit.MILLISECONDS); monitorCallback.reportSuccess(); } catch (Exception e) { monitorCallback.reportFailure(e); errorHandler.onError(e, interceptedQuery, handler); } } return response; }).filter(Objects::nonNull); } }
SimpleQueryBus implements QueryBus { @Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); if (handlers.isEmpty()) { monitorCallback.reportIgnored(); return Stream.empty(); } long deadline = System.currentTimeMillis() + unit.toMillis(timeout); return handlers.stream() .map(handler -> { long leftTimeout = getRemainingOfDeadline(deadline); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(DefaultUnitOfWork.startAndGet(interceptedQuery), handler); QueryResponseMessage<R> response = null; if (resultMessage.isExceptional()) { monitorCallback.reportFailure(resultMessage.exceptionResult()); errorHandler.onError(resultMessage.exceptionResult(), interceptedQuery, handler); } else { try { response = resultMessage.getPayload().get(leftTimeout, TimeUnit.MILLISECONDS); monitorCallback.reportSuccess(); } catch (Exception e) { monitorCallback.reportFailure(e); errorHandler.onError(e, interceptedQuery, handler); } } return response; }).filter(Objects::nonNull); } protected  SimpleQueryBus(Builder builder); }
SimpleQueryBus implements QueryBus { @Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); if (handlers.isEmpty()) { monitorCallback.reportIgnored(); return Stream.empty(); } long deadline = System.currentTimeMillis() + unit.toMillis(timeout); return handlers.stream() .map(handler -> { long leftTimeout = getRemainingOfDeadline(deadline); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(DefaultUnitOfWork.startAndGet(interceptedQuery), handler); QueryResponseMessage<R> response = null; if (resultMessage.isExceptional()) { monitorCallback.reportFailure(resultMessage.exceptionResult()); errorHandler.onError(resultMessage.exceptionResult(), interceptedQuery, handler); } else { try { response = resultMessage.getPayload().get(leftTimeout, TimeUnit.MILLISECONDS); monitorCallback.reportSuccess(); } catch (Exception e) { monitorCallback.reportFailure(e); errorHandler.onError(e, interceptedQuery, handler); } } return response; }).filter(Objects::nonNull); } protected  SimpleQueryBus(Builder builder); static Builder builder(); @Override Registration subscribe(String queryName,
                                      Type responseType,
                                      MessageHandler<? super QueryMessage<?, R>> handler); @Override CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query); @Override Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit); @SuppressWarnings("unchecked") @Override SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> subscriptionQuery(
            SubscriptionQueryMessage<Q, I, U> query,
            SubscriptionQueryBackpressure backpressure,
            int updateBufferSize); @Override QueryUpdateEmitter queryUpdateEmitter(); @Override Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?, ?>> interceptor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> interceptor); }
SimpleQueryBus implements QueryBus { @Override public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); if (handlers.isEmpty()) { monitorCallback.reportIgnored(); return Stream.empty(); } long deadline = System.currentTimeMillis() + unit.toMillis(timeout); return handlers.stream() .map(handler -> { long leftTimeout = getRemainingOfDeadline(deadline); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(DefaultUnitOfWork.startAndGet(interceptedQuery), handler); QueryResponseMessage<R> response = null; if (resultMessage.isExceptional()) { monitorCallback.reportFailure(resultMessage.exceptionResult()); errorHandler.onError(resultMessage.exceptionResult(), interceptedQuery, handler); } else { try { response = resultMessage.getPayload().get(leftTimeout, TimeUnit.MILLISECONDS); monitorCallback.reportSuccess(); } catch (Exception e) { monitorCallback.reportFailure(e); errorHandler.onError(e, interceptedQuery, handler); } } return response; }).filter(Objects::nonNull); } protected  SimpleQueryBus(Builder builder); static Builder builder(); @Override Registration subscribe(String queryName,
                                      Type responseType,
                                      MessageHandler<? super QueryMessage<?, R>> handler); @Override CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query); @Override Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit); @SuppressWarnings("unchecked") @Override SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> subscriptionQuery(
            SubscriptionQueryMessage<Q, I, U> query,
            SubscriptionQueryBackpressure backpressure,
            int updateBufferSize); @Override QueryUpdateEmitter queryUpdateEmitter(); @Override Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?, ?>> interceptor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> interceptor);  }