@Test void constraintViolationException() { ConsensusTopicQuery query = ConsensusTopicQuery.newBuilder() .setTopicID(TopicID.newBuilder().build()) .setLimit(-1) .build(); grpcConsensusService.subscribeTopic(Mono.just(query)) .as(StepVerifier::create) .expectErrorSatisfies(t -> assertException(t, Status.Code.INVALID_ARGUMENT, "limit: must be greater " + "than or equal to 0")) .verify(Duration.ofMillis(500)); }
@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);  }