@Test void subscribeHandler() { when(mockQueryChannel.registerQueryHandler(any(), any())).thenReturn(() -> CompletableFuture.completedFuture(null)); Registration result = testSubject.subscribe(TEST_QUERY, String.class, q -> "test: " + q.getPayloadType()); assertNotNull(result); verify(mockQueryChannel).registerQueryHandler(any(), eq(new QueryDefinition(TEST_QUERY, String.class))); }
@Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { Registration localRegistration = localSegment.subscribe(queryName, responseType, handler); io.axoniq.axonserver.connector.Registration serverRegistration = axonServerConnectionManager.getConnection(context) .queryChannel() .registerQueryHandler( new QueryHandler() { @Override public void handle(QueryRequest query, ReplyChannel<QueryResponse> responseHandler) { queryExecutor.submit(new QueryProcessingTask( localSegment, query, responseHandler, serializer, configuration.getClientId() )); } @Override public io.axoniq.axonserver.connector.Registration registerSubscriptionQuery( SubscriptionQuery query, UpdateHandler sendUpdate ) { SubscriptionQueryBackpressure backpressure = SubscriptionQueryBackpressure.defaultBackpressure(); UpdateHandlerRegistration<Object> updateHandler = updateEmitter.registerUpdateHandler( subscriptionSerializer.deserialize(query), backpressure, 1024 ); updateHandler.getUpdates() .doOnError(e -> { ErrorMessage error = ExceptionSerializer.serialize( configuration.getClientId(), e ); String errorCode = ErrorCode.QUERY_EXECUTION_ERROR .errorCode(); QueryUpdate queryUpdate = QueryUpdate.newBuilder() .setErrorMessage(error) .setErrorCode(errorCode) .build(); sendUpdate.sendUpdate(queryUpdate); sendUpdate.complete(); }) .doOnComplete(sendUpdate::complete) .map(subscriptionSerializer::serialize) .subscribe(sendUpdate::sendUpdate); return () -> { updateHandler.getRegistration().close(); return CompletableFuture.completedFuture(null); }; } }, new QueryDefinition(queryName, responseType) ); return new AxonServerRegistration(localRegistration, serverRegistration::cancel); }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { Registration localRegistration = localSegment.subscribe(queryName, responseType, handler); io.axoniq.axonserver.connector.Registration serverRegistration = axonServerConnectionManager.getConnection(context) .queryChannel() .registerQueryHandler( new QueryHandler() { @Override public void handle(QueryRequest query, ReplyChannel<QueryResponse> responseHandler) { queryExecutor.submit(new QueryProcessingTask( localSegment, query, responseHandler, serializer, configuration.getClientId() )); } @Override public io.axoniq.axonserver.connector.Registration registerSubscriptionQuery( SubscriptionQuery query, UpdateHandler sendUpdate ) { SubscriptionQueryBackpressure backpressure = SubscriptionQueryBackpressure.defaultBackpressure(); UpdateHandlerRegistration<Object> updateHandler = updateEmitter.registerUpdateHandler( subscriptionSerializer.deserialize(query), backpressure, 1024 ); updateHandler.getUpdates() .doOnError(e -> { ErrorMessage error = ExceptionSerializer.serialize( configuration.getClientId(), e ); String errorCode = ErrorCode.QUERY_EXECUTION_ERROR .errorCode(); QueryUpdate queryUpdate = QueryUpdate.newBuilder() .setErrorMessage(error) .setErrorCode(errorCode) .build(); sendUpdate.sendUpdate(queryUpdate); sendUpdate.complete(); }) .doOnComplete(sendUpdate::complete) .map(subscriptionSerializer::serialize) .subscribe(sendUpdate::sendUpdate); return () -> { updateHandler.getRegistration().close(); return CompletableFuture.completedFuture(null); }; } }, new QueryDefinition(queryName, responseType) ); return new AxonServerRegistration(localRegistration, serverRegistration::cancel); } }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { Registration localRegistration = localSegment.subscribe(queryName, responseType, handler); io.axoniq.axonserver.connector.Registration serverRegistration = axonServerConnectionManager.getConnection(context) .queryChannel() .registerQueryHandler( new QueryHandler() { @Override public void handle(QueryRequest query, ReplyChannel<QueryResponse> responseHandler) { queryExecutor.submit(new QueryProcessingTask( localSegment, query, responseHandler, serializer, configuration.getClientId() )); } @Override public io.axoniq.axonserver.connector.Registration registerSubscriptionQuery( SubscriptionQuery query, UpdateHandler sendUpdate ) { SubscriptionQueryBackpressure backpressure = SubscriptionQueryBackpressure.defaultBackpressure(); UpdateHandlerRegistration<Object> updateHandler = updateEmitter.registerUpdateHandler( subscriptionSerializer.deserialize(query), backpressure, 1024 ); updateHandler.getUpdates() .doOnError(e -> { ErrorMessage error = ExceptionSerializer.serialize( configuration.getClientId(), e ); String errorCode = ErrorCode.QUERY_EXECUTION_ERROR .errorCode(); QueryUpdate queryUpdate = QueryUpdate.newBuilder() .setErrorMessage(error) .setErrorCode(errorCode) .build(); sendUpdate.sendUpdate(queryUpdate); sendUpdate.complete(); }) .doOnComplete(sendUpdate::complete) .map(subscriptionSerializer::serialize) .subscribe(sendUpdate::sendUpdate); return () -> { updateHandler.getRegistration().close(); return CompletableFuture.completedFuture(null); }; } }, new QueryDefinition(queryName, responseType) ); return new AxonServerRegistration(localRegistration, serverRegistration::cancel); } AxonServerQueryBus(Builder builder); }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { Registration localRegistration = localSegment.subscribe(queryName, responseType, handler); io.axoniq.axonserver.connector.Registration serverRegistration = axonServerConnectionManager.getConnection(context) .queryChannel() .registerQueryHandler( new QueryHandler() { @Override public void handle(QueryRequest query, ReplyChannel<QueryResponse> responseHandler) { queryExecutor.submit(new QueryProcessingTask( localSegment, query, responseHandler, serializer, configuration.getClientId() )); } @Override public io.axoniq.axonserver.connector.Registration registerSubscriptionQuery( SubscriptionQuery query, UpdateHandler sendUpdate ) { SubscriptionQueryBackpressure backpressure = SubscriptionQueryBackpressure.defaultBackpressure(); UpdateHandlerRegistration<Object> updateHandler = updateEmitter.registerUpdateHandler( subscriptionSerializer.deserialize(query), backpressure, 1024 ); updateHandler.getUpdates() .doOnError(e -> { ErrorMessage error = ExceptionSerializer.serialize( configuration.getClientId(), e ); String errorCode = ErrorCode.QUERY_EXECUTION_ERROR .errorCode(); QueryUpdate queryUpdate = QueryUpdate.newBuilder() .setErrorMessage(error) .setErrorCode(errorCode) .build(); sendUpdate.sendUpdate(queryUpdate); sendUpdate.complete(); }) .doOnComplete(sendUpdate::complete) .map(subscriptionSerializer::serialize) .subscribe(sendUpdate::sendUpdate); return () -> { updateHandler.getRegistration().close(); return CompletableFuture.completedFuture(null); }; } }, new QueryDefinition(queryName, responseType) ); return new AxonServerRegistration(localRegistration, serverRegistration::cancel); } AxonServerQueryBus(Builder builder); static Builder builder(); @StartHandler(phase = Phase.INBOUND_QUERY_CONNECTOR) void start(); @Override Registration subscribe(String queryName,
                                      Type responseType,
                                      MessageHandler<? super QueryMessage<?, R>> handler); @Override CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage); @Override Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage,
                                                                long timeout,
                                                                TimeUnit timeUnit); @Override SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> subscriptionQuery(
            SubscriptionQueryMessage<Q, I, U> query,
            SubscriptionQueryBackpressure backPressure,
            int updateBufferSize
    ); @Override QueryUpdateEmitter queryUpdateEmitter(); @Override QueryBus localSegment(); @Override Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?, ?>> interceptor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> dispatchInterceptor); @ShutdownHandler(phase = Phase.INBOUND_QUERY_CONNECTOR) void disconnect(); @ShutdownHandler(phase = Phase.OUTBOUND_QUERY_CONNECTORS) CompletableFuture<Void> shutdownDispatching(); }
AxonServerQueryBus implements QueryBus, Distributed<QueryBus> { @Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { Registration localRegistration = localSegment.subscribe(queryName, responseType, handler); io.axoniq.axonserver.connector.Registration serverRegistration = axonServerConnectionManager.getConnection(context) .queryChannel() .registerQueryHandler( new QueryHandler() { @Override public void handle(QueryRequest query, ReplyChannel<QueryResponse> responseHandler) { queryExecutor.submit(new QueryProcessingTask( localSegment, query, responseHandler, serializer, configuration.getClientId() )); } @Override public io.axoniq.axonserver.connector.Registration registerSubscriptionQuery( SubscriptionQuery query, UpdateHandler sendUpdate ) { SubscriptionQueryBackpressure backpressure = SubscriptionQueryBackpressure.defaultBackpressure(); UpdateHandlerRegistration<Object> updateHandler = updateEmitter.registerUpdateHandler( subscriptionSerializer.deserialize(query), backpressure, 1024 ); updateHandler.getUpdates() .doOnError(e -> { ErrorMessage error = ExceptionSerializer.serialize( configuration.getClientId(), e ); String errorCode = ErrorCode.QUERY_EXECUTION_ERROR .errorCode(); QueryUpdate queryUpdate = QueryUpdate.newBuilder() .setErrorMessage(error) .setErrorCode(errorCode) .build(); sendUpdate.sendUpdate(queryUpdate); sendUpdate.complete(); }) .doOnComplete(sendUpdate::complete) .map(subscriptionSerializer::serialize) .subscribe(sendUpdate::sendUpdate); return () -> { updateHandler.getRegistration().close(); return CompletableFuture.completedFuture(null); }; } }, new QueryDefinition(queryName, responseType) ); return new AxonServerRegistration(localRegistration, serverRegistration::cancel); } AxonServerQueryBus(Builder builder); static Builder builder(); @StartHandler(phase = Phase.INBOUND_QUERY_CONNECTOR) void start(); @Override Registration subscribe(String queryName,
                                      Type responseType,
                                      MessageHandler<? super QueryMessage<?, R>> handler); @Override CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage); @Override Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage,
                                                                long timeout,
                                                                TimeUnit timeUnit); @Override SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> subscriptionQuery(
            SubscriptionQueryMessage<Q, I, U> query,
            SubscriptionQueryBackpressure backPressure,
            int updateBufferSize
    ); @Override QueryUpdateEmitter queryUpdateEmitter(); @Override QueryBus localSegment(); @Override Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?, ?>> interceptor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> dispatchInterceptor); @ShutdownHandler(phase = Phase.INBOUND_QUERY_CONNECTOR) void disconnect(); @ShutdownHandler(phase = Phase.OUTBOUND_QUERY_CONNECTORS) CompletableFuture<Void> shutdownDispatching();  }