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");