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