@Test void subscribeVerifySequence() { domainBuilder.topicMessage().block(); domainBuilder.topicMessage().block(); domainBuilder.topicMessage().block(); ConsensusTopicQuery query = ConsensusTopicQuery.newBuilder() .setLimit(7L) .setConsensusStartTime(Timestamp.newBuilder().setSeconds(0).build()) .setTopicID(TopicID.newBuilder().setRealmNum(0).setTopicNum(0).build()) .build(); Flux<TopicMessage> generator = domainBuilder.topicMessages(4, Instant.now().plusSeconds(10L)); grpcConsensusService.subscribeTopic(Mono.just(query)) .map(ConsensusTopicResponse::getSequenceNumber) .as(StepVerifier::create) .expectNext(1L, 2L, 3L) .thenAwait(Duration.ofMillis(50)) .then(() -> generator.blockLast()) .expectNext(4L, 5L, 6L, 7L) .expectComplete() .verify(Duration.ofMillis(1000)); }
@Override public Flux<ConsensusTopicResponse> subscribeTopic(Mono<ConsensusTopicQuery> request) { return request.map(this::toFilter) .flatMapMany(topicMessageService::subscribeTopic) .map(TopicMessage::toResponse) .onErrorMap(ConstraintViolationException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(IllegalArgumentException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(NonTransientDataAccessResourceException.class, e -> error(e, Status.UNAVAILABLE, DB_ERROR)) .onErrorMap(TimeoutException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(TopicNotFoundException.class, e -> error(e, Status.NOT_FOUND)) .onErrorMap(TransientDataAccessException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(Exceptions::isOverflow, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(ClientTimeoutException.class, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(t -> unknownError(t)); }
ConsensusController extends ReactorConsensusServiceGrpc.ConsensusServiceImplBase { @Override public Flux<ConsensusTopicResponse> subscribeTopic(Mono<ConsensusTopicQuery> request) { return request.map(this::toFilter) .flatMapMany(topicMessageService::subscribeTopic) .map(TopicMessage::toResponse) .onErrorMap(ConstraintViolationException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(IllegalArgumentException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(NonTransientDataAccessResourceException.class, e -> error(e, Status.UNAVAILABLE, DB_ERROR)) .onErrorMap(TimeoutException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(TopicNotFoundException.class, e -> error(e, Status.NOT_FOUND)) .onErrorMap(TransientDataAccessException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(Exceptions::isOverflow, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(ClientTimeoutException.class, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(t -> unknownError(t)); } }
ConsensusController extends ReactorConsensusServiceGrpc.ConsensusServiceImplBase { @Override public Flux<ConsensusTopicResponse> subscribeTopic(Mono<ConsensusTopicQuery> request) { return request.map(this::toFilter) .flatMapMany(topicMessageService::subscribeTopic) .map(TopicMessage::toResponse) .onErrorMap(ConstraintViolationException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(IllegalArgumentException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(NonTransientDataAccessResourceException.class, e -> error(e, Status.UNAVAILABLE, DB_ERROR)) .onErrorMap(TimeoutException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(TopicNotFoundException.class, e -> error(e, Status.NOT_FOUND)) .onErrorMap(TransientDataAccessException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(Exceptions::isOverflow, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(ClientTimeoutException.class, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(t -> unknownError(t)); }  }
ConsensusController extends ReactorConsensusServiceGrpc.ConsensusServiceImplBase { @Override public Flux<ConsensusTopicResponse> subscribeTopic(Mono<ConsensusTopicQuery> request) { return request.map(this::toFilter) .flatMapMany(topicMessageService::subscribeTopic) .map(TopicMessage::toResponse) .onErrorMap(ConstraintViolationException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(IllegalArgumentException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(NonTransientDataAccessResourceException.class, e -> error(e, Status.UNAVAILABLE, DB_ERROR)) .onErrorMap(TimeoutException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(TopicNotFoundException.class, e -> error(e, Status.NOT_FOUND)) .onErrorMap(TransientDataAccessException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(Exceptions::isOverflow, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(ClientTimeoutException.class, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(t -> unknownError(t)); }  @Override Flux<ConsensusTopicResponse> subscribeTopic(Mono<ConsensusTopicQuery> request); }
ConsensusController extends ReactorConsensusServiceGrpc.ConsensusServiceImplBase { @Override public Flux<ConsensusTopicResponse> subscribeTopic(Mono<ConsensusTopicQuery> request) { return request.map(this::toFilter) .flatMapMany(topicMessageService::subscribeTopic) .map(TopicMessage::toResponse) .onErrorMap(ConstraintViolationException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(IllegalArgumentException.class, e -> error(e, Status.INVALID_ARGUMENT)) .onErrorMap(NonTransientDataAccessResourceException.class, e -> error(e, Status.UNAVAILABLE, DB_ERROR)) .onErrorMap(TimeoutException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(TopicNotFoundException.class, e -> error(e, Status.NOT_FOUND)) .onErrorMap(TransientDataAccessException.class, e -> error(e, Status.RESOURCE_EXHAUSTED)) .onErrorMap(Exceptions::isOverflow, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(ClientTimeoutException.class, e -> error(e, Status.DEADLINE_EXCEEDED, OVERFLOW_ERROR)) .onErrorMap(t -> unknownError(t)); }  @Override Flux<ConsensusTopicResponse> subscribeTopic(Mono<ConsensusTopicQuery> request);  }