@Test void testPointToPointQuery() throws Exception { when(mockBus.query(anyMessage(String.class, String.class))).thenReturn(completedFuture(answer)); CompletableFuture<String> queryResponse = testSubject.query("query", String.class); assertEquals("answer", queryResponse.get()); ArgumentCaptor<QueryMessage<String, String>> queryMessageCaptor = ArgumentCaptor.forClass(QueryMessage.class); verify(mockBus).query(queryMessageCaptor.capture()); QueryMessage<String, String> result = queryMessageCaptor.getValue(); assertEquals("query", 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> CompletableFuture<R> query(String queryName, Q query, ResponseType<R> responseType) { CompletableFuture<QueryResponseMessage<R>> queryResponse = queryBus .query(processInterceptors(new GenericQueryMessage<>(asMessage(query), queryName, responseType))); CompletableFuture<R> result = new CompletableFuture<>(); queryResponse.exceptionally(cause -> asResponseMessage(responseType.responseMessagePayloadType(), cause)) .thenAccept(queryResponseMessage -> { try { if (queryResponseMessage.isExceptional()) { result.completeExceptionally(queryResponseMessage.exceptionResult()); } else { result.complete(queryResponseMessage.getPayload()); } } catch (Exception e) { result.completeExceptionally(e); } }); return result; }
DefaultQueryGateway implements QueryGateway { @Override public <R, Q> CompletableFuture<R> query(String queryName, Q query, ResponseType<R> responseType) { CompletableFuture<QueryResponseMessage<R>> queryResponse = queryBus .query(processInterceptors(new GenericQueryMessage<>(asMessage(query), queryName, responseType))); CompletableFuture<R> result = new CompletableFuture<>(); queryResponse.exceptionally(cause -> asResponseMessage(responseType.responseMessagePayloadType(), cause)) .thenAccept(queryResponseMessage -> { try { if (queryResponseMessage.isExceptional()) { result.completeExceptionally(queryResponseMessage.exceptionResult()); } else { result.complete(queryResponseMessage.getPayload()); } } catch (Exception e) { result.completeExceptionally(e); } }); return result; } }
DefaultQueryGateway implements QueryGateway { @Override public <R, Q> CompletableFuture<R> query(String queryName, Q query, ResponseType<R> responseType) { CompletableFuture<QueryResponseMessage<R>> queryResponse = queryBus .query(processInterceptors(new GenericQueryMessage<>(asMessage(query), queryName, responseType))); CompletableFuture<R> result = new CompletableFuture<>(); queryResponse.exceptionally(cause -> asResponseMessage(responseType.responseMessagePayloadType(), cause)) .thenAccept(queryResponseMessage -> { try { if (queryResponseMessage.isExceptional()) { result.completeExceptionally(queryResponseMessage.exceptionResult()); } else { result.complete(queryResponseMessage.getPayload()); } } catch (Exception e) { result.completeExceptionally(e); } }); return result; } protected  DefaultQueryGateway(Builder builder); }
DefaultQueryGateway implements QueryGateway { @Override public <R, Q> CompletableFuture<R> query(String queryName, Q query, ResponseType<R> responseType) { CompletableFuture<QueryResponseMessage<R>> queryResponse = queryBus .query(processInterceptors(new GenericQueryMessage<>(asMessage(query), queryName, responseType))); CompletableFuture<R> result = new CompletableFuture<>(); queryResponse.exceptionally(cause -> asResponseMessage(responseType.responseMessagePayloadType(), cause)) .thenAccept(queryResponseMessage -> { try { if (queryResponseMessage.isExceptional()) { result.completeExceptionally(queryResponseMessage.exceptionResult()); } else { result.complete(queryResponseMessage.getPayload()); } } catch (Exception e) { result.completeExceptionally(e); } }); return result; } 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> CompletableFuture<R> query(String queryName, Q query, ResponseType<R> responseType) { CompletableFuture<QueryResponseMessage<R>> queryResponse = queryBus .query(processInterceptors(new GenericQueryMessage<>(asMessage(query), queryName, responseType))); CompletableFuture<R> result = new CompletableFuture<>(); queryResponse.exceptionally(cause -> asResponseMessage(responseType.responseMessagePayloadType(), cause)) .thenAccept(queryResponseMessage -> { try { if (queryResponseMessage.isExceptional()) { result.completeExceptionally(queryResponseMessage.exceptionResult()); } else { result.complete(queryResponseMessage.getPayload()); } } catch (Exception e) { result.completeExceptionally(e); } }); return result; } 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);  }