@Test void fragmentedMessagesGroupAcrossHistoricAndIncoming() { Instant now = Instant.now(); domainBuilder.topicMessage(t -> t.sequenceNumber(1)).block(); domainBuilder.topicMessage(t -> t.sequenceNumber(2).chunkNum(1).chunkTotal(2) .validStartTimestamp(now).payerAccountId(1L).consensusTimestamp(now.plusNanos(1))).block(); domainBuilder .topicMessage(t -> t.sequenceNumber(3).chunkNum(2).chunkTotal(2).validStartTimestamp(now.plusNanos(1)) .payerAccountId(1L).consensusTimestamp(now.plusNanos(2))).block(); domainBuilder.topicMessage(t -> t.sequenceNumber(4).consensusTimestamp(now.plusNanos(3))).block(); domainBuilder.topicMessage(t -> t.sequenceNumber(5).chunkNum(1).chunkTotal(3) .validStartTimestamp(now.plusNanos(3)).payerAccountId(1L).consensusTimestamp(now.plusNanos(4))).block(); Flux<TopicMessage> generator = Flux.concat( domainBuilder.topicMessage(t -> t.sequenceNumber(6).chunkNum(2).chunkTotal(3) .validStartTimestamp(now.plusNanos(4)).payerAccountId(1L) .consensusTimestamp(now.plusSeconds(5))), domainBuilder.topicMessage(t -> t.sequenceNumber(7).chunkNum(3).chunkTotal(3) .validStartTimestamp(now.plusNanos(5)).payerAccountId(1L) .consensusTimestamp(now.plusSeconds(6))), domainBuilder.topicMessage(t -> t.sequenceNumber(8).consensusTimestamp(now.plusSeconds(7))) ); ConsensusTopicQuery query = ConsensusTopicQuery.newBuilder() .setConsensusStartTime(Timestamp.newBuilder().setSeconds(0).build()) .setTopicID(TopicID.newBuilder().setRealmNum(0).setTopicNum(0).build()) .build(); grpcConsensusService.subscribeTopic(Mono.just(query)) .map(x -> x.hasChunkInfo() ? x.getChunkInfo().getNumber() : 0) .as(StepVerifier::create) .thenAwait(Duration.ofMillis(100)) .then(generator::blockLast) .expectNext(0, 1, 2, 0, 1, 2, 3, 0) .thenCancel() .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);  }