@SuppressWarnings("unchecked") @Test public void dispatch() { doAnswer(invocationOnMock -> { Consumer<QueryResponse> callback = (Consumer<QueryResponse>) invocationOnMock.getArguments()[1]; callback.accept(QueryResponse.newBuilder().build()); return null; }).when(queryDispatcher).query(isA(SerializedQuery.class), isA(Consumer.class), any()); FakeStreamObserver<QueryResponse> responseObserver = new FakeStreamObserver<>(); testSubject.query(QueryRequest.newBuilder().build(), responseObserver); assertEquals(1, responseObserver.values().size()); }
@Override public void query(QueryRequest request, StreamObserver<QueryResponse> responseObserver) { if (logger.isTraceEnabled()) { logger.trace("{}: Received query: {}", request.getClientId(), request.getQuery()); } GrpcQueryResponseConsumer responseConsumer = new GrpcQueryResponseConsumer(responseObserver); queryDispatcher.query(new SerializedQuery(contextProvider.getContext(), request), responseConsumer::onNext, result -> responseConsumer.onCompleted()); }
QueryService extends QueryServiceGrpc.QueryServiceImplBase implements AxonServerClientService { @Override public void query(QueryRequest request, StreamObserver<QueryResponse> responseObserver) { if (logger.isTraceEnabled()) { logger.trace("{}: Received query: {}", request.getClientId(), request.getQuery()); } GrpcQueryResponseConsumer responseConsumer = new GrpcQueryResponseConsumer(responseObserver); queryDispatcher.query(new SerializedQuery(contextProvider.getContext(), request), responseConsumer::onNext, result -> responseConsumer.onCompleted()); } }
QueryService extends QueryServiceGrpc.QueryServiceImplBase implements AxonServerClientService { @Override public void query(QueryRequest request, StreamObserver<QueryResponse> responseObserver) { if (logger.isTraceEnabled()) { logger.trace("{}: Received query: {}", request.getClientId(), request.getQuery()); } GrpcQueryResponseConsumer responseConsumer = new GrpcQueryResponseConsumer(responseObserver); queryDispatcher.query(new SerializedQuery(contextProvider.getContext(), request), responseConsumer::onNext, result -> responseConsumer.onCompleted()); } QueryService(Topology topology, QueryDispatcher queryDispatcher, ContextProvider contextProvider,
                        ClientIdRegistry clientIdRegistry,
                        ApplicationEventPublisher eventPublisher,
                        @Qualifier("queryInstructionAckSource")
                                InstructionAckSource<QueryProviderInbound> instructionAckSource); }
QueryService extends QueryServiceGrpc.QueryServiceImplBase implements AxonServerClientService { @Override public void query(QueryRequest request, StreamObserver<QueryResponse> responseObserver) { if (logger.isTraceEnabled()) { logger.trace("{}: Received query: {}", request.getClientId(), request.getQuery()); } GrpcQueryResponseConsumer responseConsumer = new GrpcQueryResponseConsumer(responseObserver); queryDispatcher.query(new SerializedQuery(contextProvider.getContext(), request), responseConsumer::onNext, result -> responseConsumer.onCompleted()); } QueryService(Topology topology, QueryDispatcher queryDispatcher, ContextProvider contextProvider,
                        ClientIdRegistry clientIdRegistry,
                        ApplicationEventPublisher eventPublisher,
                        @Qualifier("queryInstructionAckSource")
                                InstructionAckSource<QueryProviderInbound> instructionAckSource); @PreDestroy void cleanup(); @Override StreamObserver<QueryProviderOutbound> openStream(
            StreamObserver<QueryProviderInbound> inboundStreamObserver); @Override void query(QueryRequest request, StreamObserver<QueryResponse> responseObserver); @Override StreamObserver<SubscriptionQueryRequest> subscription(
            StreamObserver<SubscriptionQueryResponse> responseObserver); Set<GrpcQueryDispatcherListener> listeners(); void completeStream(String clientId, ClientStreamIdentification clientStreamIdentification); }
QueryService extends QueryServiceGrpc.QueryServiceImplBase implements AxonServerClientService { @Override public void query(QueryRequest request, StreamObserver<QueryResponse> responseObserver) { if (logger.isTraceEnabled()) { logger.trace("{}: Received query: {}", request.getClientId(), request.getQuery()); } GrpcQueryResponseConsumer responseConsumer = new GrpcQueryResponseConsumer(responseObserver); queryDispatcher.query(new SerializedQuery(contextProvider.getContext(), request), responseConsumer::onNext, result -> responseConsumer.onCompleted()); } QueryService(Topology topology, QueryDispatcher queryDispatcher, ContextProvider contextProvider,
                        ClientIdRegistry clientIdRegistry,
                        ApplicationEventPublisher eventPublisher,
                        @Qualifier("queryInstructionAckSource")
                                InstructionAckSource<QueryProviderInbound> instructionAckSource); @PreDestroy void cleanup(); @Override StreamObserver<QueryProviderOutbound> openStream(
            StreamObserver<QueryProviderInbound> inboundStreamObserver); @Override void query(QueryRequest request, StreamObserver<QueryResponse> responseObserver); @Override StreamObserver<SubscriptionQueryRequest> subscription(
            StreamObserver<SubscriptionQueryResponse> responseObserver); Set<GrpcQueryDispatcherListener> listeners(); void completeStream(String clientId, ClientStreamIdentification clientStreamIdentification);  }