@Test void subscribeTopicBlocking() { TopicMessage topicMessage1 = domainBuilder.topicMessage().block(); TopicMessage topicMessage2 = domainBuilder.topicMessage().block(); TopicMessage topicMessage3 = domainBuilder.topicMessage().block(); ConsensusTopicQuery query = ConsensusTopicQuery.newBuilder() .setLimit(3L) .setConsensusStartTime(Timestamp.newBuilder().setSeconds(0).build()) .setTopicID(TopicID.newBuilder().setRealmNum(0).setTopicNum(0).build()) .build(); assertThat(blockingService.subscribeTopic(query)) .toIterable() .hasSize(3) .containsSequence(topicMessage1.toResponse(), topicMessage2.toResponse(), topicMessage3.toResponse()); }
@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);  }