From 74e236bc6ed9f05ead8600d27437275e2d1e0b3a Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Tue, 6 Oct 2026 09:31:45 +0200 Subject: [PATCH] Retry soon while the poll endpoint is not known yet A poller that starts while the stream is still being registered found no endpoint and slept the full interval before looking again, so the first poll came up to an interval after startup. Not knowing the endpoint is a condition of the startup, not a request to slow down: the poller now looks again after a second, capped by the interval. Closes #42 Signed-off-by: Thomas Darimont --- CHANGELOG.md | 6 ++++ .../README.md | 2 ++ .../org/easyssf/receiver/poll/SsfPoller.java | 34 +++++++++++++++++-- .../easyssf/receiver/poll/SsfPollerTests.java | 21 ++++++++++++ 4 files changed, 60 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e2b1fd..81c232f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ The notable changes of every release. The section of a version is the text of it ## [Unreleased] +### Fixed + +- `SsfPoller` looks again after one second (`setEndpointRetry`) while the poll endpoint of the stream + is not known yet, instead of waiting the full interval: an application whose stream was still being + registered when the poller started polled for the first time up to an interval later. + ## [0.4.0] - 2026-10-06 One schema change: the column PROCESSED_AT of EASYSSF_PROCESSED_SET becomes diff --git a/easyssf-receiver-spring-boot-starter/README.md b/easyssf-receiver-spring-boot-starter/README.md index b3f5dcf..e317f4d 100644 --- a/easyssf-receiver-spring-boot-starter/README.md +++ b/easyssf-receiver-spring-boot-starter/README.md @@ -207,6 +207,8 @@ easyssf: agreed with the transmitter (30 seconds by default); the request waits that long plus a margin. A transmitter that answers an empty long poll at once is polled every `poll.interval` nevertheless. Whether Keycloak holds poll requests has not been verified; its default remains short polling. + Until the stream is registered and its poll endpoint known, the poller looks again every second + rather than waiting `poll.interval`, so the first poll follows the registration closely. With long polling on, the poller calls the transmitter with the JDK HTTP client even if the application has a `RestClient`, because the `RestClient` has no timeout per request (`SsfTransmitterFactory.pollHttpClient`); a poller bean of your own, built with diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java index 7d266ca..876f7ab 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java @@ -46,6 +46,13 @@ * instead: the transmitter holds the request until SETs are available or its hold time * elapses (section 2.5), and the poller sends the next request as soon as it has handled * the response. A backlog is fetched with immediate requests in both modes. + * + *

+ * Until the poll endpoint of the stream is known, typically because the + * {@code SsfStreamRegistrar} is still registering the stream when the poller starts, the + * poller looks again every {@link #setEndpointRetry(Duration) endpoint retry} (one + * second) rather than waiting the interval: not knowing the endpoint is a condition of + * the startup, not a request of the transmitter to slow down. */ public class SsfPoller { @@ -86,6 +93,8 @@ public class SsfPoller { private Duration initialDelay = Duration.ZERO; + private Duration endpointRetry = Duration.ofSeconds(1); + private Duration longPollingHold; private boolean flushOnStop = true; @@ -181,6 +190,15 @@ public void setInitialDelay(Duration initialDelay) { * (section 2.2); the request waits that long plus a margin for the response. * {@code null} returns to short polling. */ + /** + * How soon the poller looks again while the poll endpoint of the stream is not known + * yet, one second by default; never longer than the interval. + */ + public void setEndpointRetry(Duration endpointRetry) { + SsfAssert.isTrue(endpointRetry != null && endpointRetry.isPositive(), "endpointRetry must be positive"); + this.endpointRetry = endpointRetry; + } + public void setLongPolling(Duration hold) { SsfAssert.isTrue(hold == null || hold.isPositive(), "hold must be positive"); this.longPollingHold = hold; @@ -351,7 +369,11 @@ private void run() { long started = System.nanoTime(); try { PollResult result = this.polling.tryLock() ? pollLocked() : PollResult.SKIPPED; - if (result.skipped()) { + if (result.endpointUnknown()) { + // the stream is still being registered: look again soon + pause = (this.endpointRetry.compareTo(this.interval) < 0) ? this.endpointRetry : this.interval; + } + else if (result.skipped()) { pause = pauseWhileSkipped(); } else if (isLongPolling()) { @@ -421,7 +443,7 @@ private PollResult poll(boolean hold) { URI endpoint = this.endpoint.get(); if (endpoint == null) { logger.debug("Not polling, the poll endpoint of the SSF stream is not known yet"); - return PollResult.SKIPPED; + return PollResult.NO_ENDPOINT; } if (Instant.now().isBefore(this.pausedUntil)) { logger.debug("Not polling, the SSF transmitter asked to wait until " + this.pausedUntil); @@ -601,10 +623,16 @@ private static Map sets(Map response) { return (response.get("sets") instanceof Map sets) ? (Map) sets : Map.of(); } - private record PollResult(int fetched, boolean skipped) { + private record PollResult(int fetched, boolean skipped, boolean endpointUnknown) { + + PollResult(int fetched, boolean skipped) { + this(fetched, skipped, false); + } static final PollResult SKIPPED = new PollResult(0, true); + static final PollResult NO_ENDPOINT = new PollResult(0, true, true); + } } diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java index ff857e4..afcb283 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java @@ -4,6 +4,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import org.easyssf.receiver.event.SsfEventHandler; import org.easyssf.receiver.http.JdkSsfHttpClient; @@ -206,6 +207,26 @@ void doesNotPollWhileThePollEndpointIsUnknown() { assertThat(this.handled).isEmpty(); } + @Test + void pollsSoonAfterTheEndpointBecomesKnownInsteadOfWaitingTheInterval() { + AtomicReference endpoint = new AtomicReference<>(); + SsfPoller waiting = poller(endpoint::get, TestTransmitter.ACCESS_TOKEN); + waiting.setInterval(Duration.ofHours(1)); + waiting.setEndpointRetry(Duration.ofMillis(100)); + String jti = queue("session-1"); + waiting.start(); + try { + // the stream is still being registered: nothing is polled + await().during(Duration.ofMillis(300)).untilAsserted(() -> assertThat(this.handled).isEmpty()); + // registered: the poller notices within the retry, not after the interval + endpoint.set(URI.create(transmitter.pollUri())); + await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> assertThat(this.handled).containsExactly(jti)); + } + finally { + waiting.stop(); + } + } + @Test void reportsFailedPollRequest() { SsfPoller unauthenticated = poller(() -> URI.create(transmitter.pollUri()), "wrong-token");