@Test void testQueryWithHandlersResultsInException() throws Exception { QueryMessage<String, String> testQueryMessage = new GenericQueryMessage<>("query", "query", singleStringResponse); CompletableFuture<QueryResponseMessage<String>> result = testSubject.query(testQueryMessage); assertTrue(result.isDone()); assertTrue(result.isCompletedExceptionally()); assertEquals("NoHandlerForQueryException", result.thenApply(QueryResponseMessage::getPayload) .exceptionally(e -> e.getCause().getClass().getSimpleName()) .get()); }
@Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); CompletableFuture<QueryResponseMessage<R>> result = new CompletableFuture<>(); try { if (handlers.isEmpty()) { throw new NoHandlerForQueryException( format("No handler found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } Iterator<MessageHandler<? super QueryMessage<?, ?>>> handlerIterator = handlers.iterator(); boolean invocationSuccess = false; while (!invocationSuccess && handlerIterator.hasNext()) { DefaultUnitOfWork<QueryMessage<Q, R>> uow = DefaultUnitOfWork.startAndGet(interceptedQuery); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(uow, handlerIterator.next()); if (resultMessage.isExceptional()) { if (!(resultMessage.exceptionResult() instanceof NoHandlerForQueryException)) { result.complete(new GenericQueryResponseMessage<>( interceptedQuery.getResponseType().responseMessagePayloadType(), resultMessage.exceptionResult())); monitorCallback.reportFailure(resultMessage.exceptionResult()); return result; } } else { result = resultMessage.getPayload(); invocationSuccess = true; } } if (!invocationSuccess) { throw new NoHandlerForQueryException( format("No suitable handler was found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } monitorCallback.reportSuccess(); } catch (Exception e) { result.completeExceptionally(e); monitorCallback.reportFailure(e); } return result; }
SimpleQueryBus implements QueryBus { @Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); CompletableFuture<QueryResponseMessage<R>> result = new CompletableFuture<>(); try { if (handlers.isEmpty()) { throw new NoHandlerForQueryException( format("No handler found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } Iterator<MessageHandler<? super QueryMessage<?, ?>>> handlerIterator = handlers.iterator(); boolean invocationSuccess = false; while (!invocationSuccess && handlerIterator.hasNext()) { DefaultUnitOfWork<QueryMessage<Q, R>> uow = DefaultUnitOfWork.startAndGet(interceptedQuery); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(uow, handlerIterator.next()); if (resultMessage.isExceptional()) { if (!(resultMessage.exceptionResult() instanceof NoHandlerForQueryException)) { result.complete(new GenericQueryResponseMessage<>( interceptedQuery.getResponseType().responseMessagePayloadType(), resultMessage.exceptionResult())); monitorCallback.reportFailure(resultMessage.exceptionResult()); return result; } } else { result = resultMessage.getPayload(); invocationSuccess = true; } } if (!invocationSuccess) { throw new NoHandlerForQueryException( format("No suitable handler was found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } monitorCallback.reportSuccess(); } catch (Exception e) { result.completeExceptionally(e); monitorCallback.reportFailure(e); } return result; } }
SimpleQueryBus implements QueryBus { @Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); CompletableFuture<QueryResponseMessage<R>> result = new CompletableFuture<>(); try { if (handlers.isEmpty()) { throw new NoHandlerForQueryException( format("No handler found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } Iterator<MessageHandler<? super QueryMessage<?, ?>>> handlerIterator = handlers.iterator(); boolean invocationSuccess = false; while (!invocationSuccess && handlerIterator.hasNext()) { DefaultUnitOfWork<QueryMessage<Q, R>> uow = DefaultUnitOfWork.startAndGet(interceptedQuery); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(uow, handlerIterator.next()); if (resultMessage.isExceptional()) { if (!(resultMessage.exceptionResult() instanceof NoHandlerForQueryException)) { result.complete(new GenericQueryResponseMessage<>( interceptedQuery.getResponseType().responseMessagePayloadType(), resultMessage.exceptionResult())); monitorCallback.reportFailure(resultMessage.exceptionResult()); return result; } } else { result = resultMessage.getPayload(); invocationSuccess = true; } } if (!invocationSuccess) { throw new NoHandlerForQueryException( format("No suitable handler was found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } monitorCallback.reportSuccess(); } catch (Exception e) { result.completeExceptionally(e); monitorCallback.reportFailure(e); } return result; } protected  SimpleQueryBus(Builder builder); }
SimpleQueryBus implements QueryBus { @Override public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); CompletableFuture<QueryResponseMessage<R>> result = new CompletableFuture<>(); try { if (handlers.isEmpty()) { throw new NoHandlerForQueryException( format("No handler found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } Iterator<MessageHandler<? super QueryMessage<?, ?>>> handlerIterator = handlers.iterator(); boolean invocationSuccess = false; while (!invocationSuccess && handlerIterator.hasNext()) { DefaultUnitOfWork<QueryMessage<Q, R>> uow = DefaultUnitOfWork.startAndGet(interceptedQuery); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(uow, handlerIterator.next()); if (resultMessage.isExceptional()) { if (!(resultMessage.exceptionResult() instanceof NoHandlerForQueryException)) { result.complete(new GenericQueryResponseMessage<>( interceptedQuery.getResponseType().responseMessagePayloadType(), resultMessage.exceptionResult())); monitorCallback.reportFailure(resultMessage.exceptionResult()); return result; } } else { result = resultMessage.getPayload(); invocationSuccess = true; } } if (!invocationSuccess) { throw new NoHandlerForQueryException( format("No suitable handler was found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } monitorCallback.reportSuccess(); } catch (Exception e) { result.completeExceptionally(e); monitorCallback.reportFailure(e); } return result; } 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> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query) { MessageMonitor.MonitorCallback monitorCallback = messageMonitor.onMessageIngested(query); QueryMessage<Q, R> interceptedQuery = intercept(query); List<MessageHandler<? super QueryMessage<?, ?>>> handlers = getHandlersForMessage(interceptedQuery); CompletableFuture<QueryResponseMessage<R>> result = new CompletableFuture<>(); try { if (handlers.isEmpty()) { throw new NoHandlerForQueryException( format("No handler found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } Iterator<MessageHandler<? super QueryMessage<?, ?>>> handlerIterator = handlers.iterator(); boolean invocationSuccess = false; while (!invocationSuccess && handlerIterator.hasNext()) { DefaultUnitOfWork<QueryMessage<Q, R>> uow = DefaultUnitOfWork.startAndGet(interceptedQuery); ResultMessage<CompletableFuture<QueryResponseMessage<R>>> resultMessage = interceptAndInvoke(uow, handlerIterator.next()); if (resultMessage.isExceptional()) { if (!(resultMessage.exceptionResult() instanceof NoHandlerForQueryException)) { result.complete(new GenericQueryResponseMessage<>( interceptedQuery.getResponseType().responseMessagePayloadType(), resultMessage.exceptionResult())); monitorCallback.reportFailure(resultMessage.exceptionResult()); return result; } } else { result = resultMessage.getPayload(); invocationSuccess = true; } } if (!invocationSuccess) { throw new NoHandlerForQueryException( format("No suitable handler was found for [%s] with response type [%s]", interceptedQuery.getQueryName(), interceptedQuery.getResponseType()) ); } monitorCallback.reportSuccess(); } catch (Exception e) { result.completeExceptionally(e); monitorCallback.reportFailure(e); } return result; } 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);  }