@Test void testExceptionInInitialResultOfSubscriptionQueryReportedInMono() { when(mockBus.subscriptionQuery(anySubscriptionMessage(String.class, String.class), any(), anyInt())) .thenReturn(new DefaultSubscriptionQueryResult<>( Mono.just(new GenericQueryResponseMessage<>(String.class, new MockException())), Flux.empty(), () -> true )); SubscriptionQueryResult<String, String> actual = testSubject.subscriptionQuery("Test", instanceOf(String.class), instanceOf(String.class)); assertEquals( MockException.class, actual.initialResult().map(i -> null).onErrorResume(e -> Mono.just(e.getClass())).block() ); }
@Override public <Q, I, U> SubscriptionQueryResult<I, U> subscriptionQuery(String queryName, Q query, ResponseType<I> initialResponseType, ResponseType<U> updateResponseType, SubscriptionQueryBackpressure backpressure, int updateBufferSize) { SubscriptionQueryMessage<?, I, U> subscriptionQueryMessage = new GenericSubscriptionQueryMessage<>( asMessage(query), queryName, initialResponseType, updateResponseType ); SubscriptionQueryMessage<?, I, U> interceptedQuery = processInterceptors(subscriptionQueryMessage); SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> result = queryBus.subscriptionQuery(interceptedQuery, backpressure, updateBufferSize); return new DefaultSubscriptionQueryResult<>( result.initialResult() .filter(initialResult -> Objects.nonNull(initialResult.getPayload())) .map(Message::getPayload) .onErrorMap(e -> e instanceof IllegalPayloadAccessException ? e.getCause() : e), result.updates() .filter(update -> Objects.nonNull(update.getPayload())) .map(SubscriptionQueryUpdateMessage::getPayload), result ); }
DefaultQueryGateway implements QueryGateway { @Override public <Q, I, U> SubscriptionQueryResult<I, U> subscriptionQuery(String queryName, Q query, ResponseType<I> initialResponseType, ResponseType<U> updateResponseType, SubscriptionQueryBackpressure backpressure, int updateBufferSize) { SubscriptionQueryMessage<?, I, U> subscriptionQueryMessage = new GenericSubscriptionQueryMessage<>( asMessage(query), queryName, initialResponseType, updateResponseType ); SubscriptionQueryMessage<?, I, U> interceptedQuery = processInterceptors(subscriptionQueryMessage); SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> result = queryBus.subscriptionQuery(interceptedQuery, backpressure, updateBufferSize); return new DefaultSubscriptionQueryResult<>( result.initialResult() .filter(initialResult -> Objects.nonNull(initialResult.getPayload())) .map(Message::getPayload) .onErrorMap(e -> e instanceof IllegalPayloadAccessException ? e.getCause() : e), result.updates() .filter(update -> Objects.nonNull(update.getPayload())) .map(SubscriptionQueryUpdateMessage::getPayload), result ); } }
DefaultQueryGateway implements QueryGateway { @Override public <Q, I, U> SubscriptionQueryResult<I, U> subscriptionQuery(String queryName, Q query, ResponseType<I> initialResponseType, ResponseType<U> updateResponseType, SubscriptionQueryBackpressure backpressure, int updateBufferSize) { SubscriptionQueryMessage<?, I, U> subscriptionQueryMessage = new GenericSubscriptionQueryMessage<>( asMessage(query), queryName, initialResponseType, updateResponseType ); SubscriptionQueryMessage<?, I, U> interceptedQuery = processInterceptors(subscriptionQueryMessage); SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> result = queryBus.subscriptionQuery(interceptedQuery, backpressure, updateBufferSize); return new DefaultSubscriptionQueryResult<>( result.initialResult() .filter(initialResult -> Objects.nonNull(initialResult.getPayload())) .map(Message::getPayload) .onErrorMap(e -> e instanceof IllegalPayloadAccessException ? e.getCause() : e), result.updates() .filter(update -> Objects.nonNull(update.getPayload())) .map(SubscriptionQueryUpdateMessage::getPayload), result ); } protected  DefaultQueryGateway(Builder builder); }
DefaultQueryGateway implements QueryGateway { @Override public <Q, I, U> SubscriptionQueryResult<I, U> subscriptionQuery(String queryName, Q query, ResponseType<I> initialResponseType, ResponseType<U> updateResponseType, SubscriptionQueryBackpressure backpressure, int updateBufferSize) { SubscriptionQueryMessage<?, I, U> subscriptionQueryMessage = new GenericSubscriptionQueryMessage<>( asMessage(query), queryName, initialResponseType, updateResponseType ); SubscriptionQueryMessage<?, I, U> interceptedQuery = processInterceptors(subscriptionQueryMessage); SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> result = queryBus.subscriptionQuery(interceptedQuery, backpressure, updateBufferSize); return new DefaultSubscriptionQueryResult<>( result.initialResult() .filter(initialResult -> Objects.nonNull(initialResult.getPayload())) .map(Message::getPayload) .onErrorMap(e -> e instanceof IllegalPayloadAccessException ? e.getCause() : e), result.updates() .filter(update -> Objects.nonNull(update.getPayload())) .map(SubscriptionQueryUpdateMessage::getPayload), 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 <Q, I, U> SubscriptionQueryResult<I, U> subscriptionQuery(String queryName, Q query, ResponseType<I> initialResponseType, ResponseType<U> updateResponseType, SubscriptionQueryBackpressure backpressure, int updateBufferSize) { SubscriptionQueryMessage<?, I, U> subscriptionQueryMessage = new GenericSubscriptionQueryMessage<>( asMessage(query), queryName, initialResponseType, updateResponseType ); SubscriptionQueryMessage<?, I, U> interceptedQuery = processInterceptors(subscriptionQueryMessage); SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> result = queryBus.subscriptionQuery(interceptedQuery, backpressure, updateBufferSize); return new DefaultSubscriptionQueryResult<>( result.initialResult() .filter(initialResult -> Objects.nonNull(initialResult.getPayload())) .map(Message::getPayload) .onErrorMap(e -> e instanceof IllegalPayloadAccessException ? e.getCause() : e), result.updates() .filter(update -> Objects.nonNull(update.getPayload())) .map(SubscriptionQueryUpdateMessage::getPayload), 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);  }