@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)); }
@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 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 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); }
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); }
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);  }