PollingTopicMessageRetriever implements TopicMessageRetriever { @Override public Flux<TopicMessage> retrieve(TopicMessageFilter filter) { if (!retrieverProperties.isEnabled()) { return Flux.empty(); } PollingContext context = new PollingContext(filter); return Flux.defer(() -> poll(context)) .repeatWhen(Repeat.create(r -> !context.isComplete(), Long.MAX_VALUE) .fixedBackoff(retrieverProperties.getPollingFrequency()) .jitter(Jitter.random(0.1)) .withBackoffScheduler(scheduler)) .name("retriever") .metrics() .retryBackoff(Long.MAX_VALUE, Duration.ofSeconds(1)) .timeout(retrieverProperties.getTimeout(), scheduler) .doOnCancel(context::onComplete) .doOnComplete(context::onComplete) .doOnNext(context::onNext); } PollingTopicMessageRetriever(RetrieverProperties retrieverProperties,
                                        TopicMessageRepository topicMessageRepository); @Override Flux<TopicMessage> retrieve(TopicMessageFilter filter);  }
@Test void notEnabled() { retrieverProperties.setEnabled(false); domainBuilder.topicMessage().block(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNextCount(0L) .expectComplete() .verify(Duration.ofMillis(500)); retrieverProperties.setEnabled(true); }
@Test void timeout() { int maxPageSize = retrieverProperties.getMaxPageSize(); Duration timeout = retrieverProperties.getTimeout(); retrieverProperties.setMaxPageSize(1); retrieverProperties.setTimeout(Duration.ofMillis(10)); domainBuilder.topicMessages(10, Instant.now()).blockLast(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .thenConsumeWhile(i -> true) .expectTimeout(Duration.ofMillis(500)) .verify(); retrieverProperties.setMaxPageSize(maxPageSize); retrieverProperties.setTimeout(timeout); }
@Test void noMessages() { TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNextCount(0L) .expectComplete() .verify(Duration.ofMillis(500)); }
@Test void lessThanPageSize() { domainBuilder.topicMessage().block(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNext(1L) .expectComplete() .verify(Duration.ofMillis(500)); }
@Test void equalPageSize() { int maxPageSize = retrieverProperties.getMaxPageSize(); retrieverProperties.setMaxPageSize(2); domainBuilder.topicMessage().block(); domainBuilder.topicMessage().block(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNext(1L, 2L) .expectComplete() .verify(Duration.ofMillis(500)); retrieverProperties.setMaxPageSize(maxPageSize); }
@Test void greaterThanPageSize() { int maxPageSize = retrieverProperties.getMaxPageSize(); retrieverProperties.setMaxPageSize(2); domainBuilder.topicMessage().block(); domainBuilder.topicMessage().block(); domainBuilder.topicMessage().block(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNext(1L, 2L, 3L) .expectComplete() .verify(Duration.ofMillis(500)); retrieverProperties.setMaxPageSize(maxPageSize); }
@Test void startTimeBefore() { domainBuilder.topicMessages(10, Instant.now()).blockLast(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNext(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L) .thenCancel() .verify(Duration.ofMillis(1000)); }
@Test void startTimeEquals() { Instant now = Instant.now(); domainBuilder.topicMessage(t -> t.consensusTimestamp(now)).block(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(now) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNext(1L) .thenCancel() .verify(Duration.ofMillis(500)); }
@Test void startTimeAfter() { Instant now = Instant.now(); domainBuilder.topicMessage(t -> t.consensusTimestamp(now.minusNanos(1))).block(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(now) .build(); pollingTopicMessageRetriever.retrieve(filter) .as(StepVerifier::create) .expectNextCount(0) .thenCancel() .verify(Duration.ofMillis(500)); }
@Test void topicNum() { domainBuilder.topicMessage(t -> t.topicNum(0)).block(); domainBuilder.topicMessage(t -> t.topicNum(1)).block(); domainBuilder.topicMessage(t -> t.topicNum(2)).block(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .topicNum(1) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNext(2L) .thenCancel() .verify(Duration.ofMillis(500)); }
@Test void realmNum() { domainBuilder.topicMessage(t -> t.realmNum(0)).block(); domainBuilder.topicMessage(t -> t.realmNum(1)).block(); domainBuilder.topicMessage(t -> t.realmNum(2)).block(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .realmNum(1) .build(); pollingTopicMessageRetriever.retrieve(filter) .map(TopicMessage::getSequenceNumber) .as(StepVerifier::create) .expectNext(2L) .thenCancel() .verify(Duration.ofMillis(500)); }