From 451ebbb7f792c8f69c6b962e3e5c00a5a5538de6 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Tue, 6 Oct 2026 09:39:35 +0200 Subject: [PATCH 1/2] Wake the poller up when the stream is registered SsfStreamRegistrar tells its listeners once the stream is registered, SsfPoller.wakeUp() ends the pause between two polls, and SsfTransmitter wires the two, so the first poll follows the registration at once instead of up to an interval later. The pause is a condition wait now rather than a sleep. Closes #43 Signed-off-by: Thomas Darimont --- CHANGELOG.md | 7 ++ .../README.md | 5 +- .../org/easyssf/receiver/poll/SsfPoller.java | 63 ++++++++++++- .../receiver/stream/SsfStreamRegistrar.java | 42 +++++++++ .../receiver/transmitter/SsfTransmitter.java | 4 + .../easyssf/receiver/poll/SsfPollerTests.java | 24 +++++ .../stream/SsfStreamManagementTests.java | 14 +++ .../transmitter/SsfTransmitterTests.java | 90 +++++++++++++++++++ 8 files changed, 245 insertions(+), 4 deletions(-) create mode 100644 easyssf-receiver/src/test/java/org/easyssf/receiver/transmitter/SsfTransmitterTests.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 81c232f..09031f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ The notable changes of every release. The section of a version is the text of it ## [Unreleased] +### Added + +- `SsfStreamRegistrar.addListener` tells a listener when the stream is registered, and + `SsfPoller.wakeUp()` ends the pause between two polls. `SsfTransmitter` wires the two, so the first + poll follows the registration of the stream at once; the Spring Boot starter and the Quarkus extension + build their transmitters through it. + ### Fixed - `SsfPoller` looks again after one second (`setEndpointRetry`) while the poll endpoint of the stream diff --git a/easyssf-receiver-spring-boot-starter/README.md b/easyssf-receiver-spring-boot-starter/README.md index e317f4d..09994d4 100644 --- a/easyssf-receiver-spring-boot-starter/README.md +++ b/easyssf-receiver-spring-boot-starter/README.md @@ -207,8 +207,9 @@ 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. + The first poll follows the registration of the stream at once: the registrar wakes the poller up, + and until the endpoint is known the poller looks again every second rather than waiting + `poll.interval`. 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 876f7ab..1fb13a8 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 @@ -8,6 +8,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Supplier; @@ -52,7 +53,9 @@ * {@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. + * the startup, not a request of the transmitter to slow down. {@link #wakeUp()} ends the + * pause between two polls altogether; {@code SsfTransmitter} calls it when the stream is + * registered, so that the first poll follows the registration at once. */ public class SsfPoller { @@ -126,6 +129,13 @@ public class SsfPoller { /** start() and stop() are idempotent and safe to call from any thread */ private final ReentrantLock lifecycle = new ReentrantLock(); + private final ReentrantLock pausing = new ReentrantLock(); + + private final Condition wokenUp = this.pausing.newCondition(); + + /** only read and written while holding {@code pausing}, hence not volatile */ + private boolean wakeUpRequested; + /** * @param httpClient used to call the transmitter * @param tokenProvider provides the access token to authenticate with @@ -323,6 +333,24 @@ public boolean isRunning() { return this.thread != null; } + /** + * Ends the pause between two polls, so that the next poll runs now, on the poller's + * thread and with its error handling. Does nothing while the poller is not running; + * {@link #pollNow()} polls on the caller's thread regardless. + */ + public void wakeUp() { + this.pausing.lock(); + try { + if (this.thread != null) { + this.wakeUpRequested = true; + this.wokenUp.signalAll(); + } + } + finally { + this.pausing.unlock(); + } + } + /** * @return when the transmitter was last polled, {@code null} if it was not polled yet */ @@ -365,6 +393,7 @@ private void run() { try { Thread.sleep(this.initialDelay); while (this.running) { + clearWakeUp(); Duration pause = this.interval; long started = System.nanoTime(); try { @@ -390,7 +419,7 @@ else if (isLongPolling()) { logger.debug("Cause of the failed poll", ex); } if (this.running && pause.isPositive()) { - Thread.sleep(pause); + pause(pause); } } } @@ -399,6 +428,36 @@ else if (isLongPolling()) { } } + /** + * Waits the pause, unless {@link #wakeUp()} ends it earlier. + */ + private void pause(Duration pause) throws InterruptedException { + long nanos = pause.toNanos(); + this.pausing.lock(); + try { + while (!this.wakeUpRequested && nanos > 0 && this.running) { + nanos = this.wokenUp.awaitNanos(nanos); + } + } + finally { + this.pausing.unlock(); + } + } + + /** + * A wake-up during a poll is not kept: a registration that completes meanwhile is + * picked up by the endpoint retry. + */ + private void clearWakeUp() { + this.pausing.lock(); + try { + this.wakeUpRequested = false; + } + finally { + this.pausing.unlock(); + } + } + private PollResult pollLocked() { try { return poll(); diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java index 5c496ac..ad3fe46 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java @@ -7,6 +7,7 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; import org.easyssf.core.SsfDeliveryMethod; import org.easyssf.core.event.SsfEventTypes; @@ -24,6 +25,11 @@ * Either looks up a stream that was created at the transmitter by its identifier, or * manages the stream itself: it reuses the stream the receiver already has at the * transmitter, brings it in line with the desired configuration, or creates it. + * + *

+ * {@link #addListener Listeners} learn when the stream is registered; + * {@code SsfTransmitter} uses that to have the poller fetch right away instead of an + * interval later. */ public class SsfStreamRegistrar { @@ -49,6 +55,23 @@ public class SsfStreamRegistrar { private volatile String lastError; + private final List listeners = new CopyOnWriteArrayList<>(); + + /** + * Told when the stream was looked up or registered. + */ + @FunctionalInterface + public interface Listener { + + /** + * The stream is registered: its configuration is known and the + * {@code SsfReceiverStream} is set. + * @param stream the stream as the transmitter has it + */ + void streamRegistered(SsfStreamConfiguration stream); + + } + /** * Where the registration stands. */ @@ -113,6 +136,16 @@ public void setDeleteOnShutdown(boolean deleteOnShutdown) { this.deleteOnShutdown = deleteOnShutdown; } + /** + * Adds a listener that is told whenever {@link #register()} succeeded, on the thread + * that registered the stream. A listener that throws is logged and does not keep the + * stream from being used. + */ + public void addListener(Listener listener) { + SsfAssert.notNull(listener, "listener must not be null"); + this.listeners.add(listener); + } + public void setInitialRetryDelay(Duration initialRetryDelay) { SsfAssert.isTrue(initialRetryDelay != null && initialRetryDelay.isPositive(), "initialRetryDelay must be positive"); @@ -248,6 +281,15 @@ else if (existing != null) { + stream.audience() + ")"); this.lastError = null; this.state = State.REGISTERED; + for (Listener listener : this.listeners) { + try { + listener.streamRegistered(stream); + } + catch (RuntimeException ex) { + logger.warn("A listener failed on the registration of SSF stream " + stream.streamId() + ": " + ex); + logger.debug("Cause of the failed listener", ex); + } + } return stream; } diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/transmitter/SsfTransmitter.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/transmitter/SsfTransmitter.java index 1fcf405..b0612c6 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/transmitter/SsfTransmitter.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/transmitter/SsfTransmitter.java @@ -57,6 +57,10 @@ private SsfTransmitter(Builder builder) { this.poller = builder.poller; this.autoStartPolling = builder.autoStartPolling; this.pushAuthorizationHeader = builder.pushAuthorizationHeader; + if (this.streamRegistrar != null && this.poller != null) { + // the first poll right after the stream is registered, not an interval later + this.streamRegistrar.addListener((stream) -> this.poller.wakeUp()); + } } /** 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 afcb283..0eb1659 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 @@ -227,6 +227,30 @@ void pollsSoonAfterTheEndpointBecomesKnownInsteadOfWaitingTheInterval() { } } + @Test + void wakeUpEndsThePauseAndPollsAtOnce() { + this.poller.setInterval(Duration.ofHours(1)); + this.poller.start(); + try { + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() == 1); + String jti = queue("session-1"); + await().during(Duration.ofMillis(300)).untilAsserted(() -> assertThat(this.handled).isEmpty()); + this.poller.wakeUp(); + await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> assertThat(this.handled).containsExactly(jti)); + } + finally { + this.poller.stop(); + } + } + + @Test + void wakeUpDoesNothingWhileThePollerIsNotRunning() { + queue("session-1"); + this.poller.wakeUp(); + await().during(Duration.ofMillis(300)).until(() -> transmitter.pollRequests() == 0); + assertThat(this.handled).isEmpty(); + } + @Test void reportsFailedPollRequest() { SsfPoller unauthenticated = poller(() -> URI.create(transmitter.pollUri()), "wrong-token"); diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java index 09db201..3fb5b05 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java @@ -2,6 +2,7 @@ import java.net.URI; import java.time.Duration; +import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -181,6 +182,19 @@ void registrarCreatesStream() { .containsExactly(TestTransmitter.CLIENT_ID + "/" + stream.streamId()); } + @Test + void registrarTellsItsListenersOnceTheStreamIsRegistered() { + List registered = new ArrayList<>(); + SsfStreamRegistrar registrar = registrar(SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.addListener((stream) -> { + throw new IllegalStateException("a failing listener does not keep the others from being told"); + }); + registrar.addListener(registered::add); + SsfStreamConfiguration stream = registrar.register(); + assertThat(registered).containsExactly(stream); + assertThat(this.receiverStream.getStreamId()).isEqualTo(stream.streamId()); + } + @Test void registrarReusesExistingStream() { SsfStreamConfiguration desired = SsfStreamConfiguration.push(PUSH_ENDPOINT, null, diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/transmitter/SsfTransmitterTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/transmitter/SsfTransmitterTests.java new file mode 100644 index 0000000..76b002a --- /dev/null +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/transmitter/SsfTransmitterTests.java @@ -0,0 +1,90 @@ +package org.easyssf.receiver.transmitter; + +import java.net.URI; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; + +import org.easyssf.core.stream.SsfStreamConfiguration; +import org.easyssf.receiver.event.SsfEventHandler; +import org.easyssf.receiver.http.JdkSsfHttpClient; +import org.easyssf.receiver.http.SsfHttpClient; +import org.easyssf.receiver.poll.SsfPoller; +import org.easyssf.receiver.set.InMemorySsfJtiDedupStore; +import org.easyssf.receiver.set.NimbusSsfSetVerifier; +import org.easyssf.receiver.set.SsfSetProcessor; +import org.easyssf.receiver.stream.SsfReceiverStream; +import org.easyssf.receiver.stream.SsfStreamClient; +import org.easyssf.receiver.stream.SsfStreamRegistrar; +import org.easyssf.test.TestTransmitter; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.easyssf.core.event.SsfSubjectIdentifiers.opaque; + +/** + * The parts of a transmitter working together: the registrar registers the stream in the + * background and the poller fetches right after, not an interval later. + */ +class SsfTransmitterTests { + + private static final TestTransmitter transmitter = new TestTransmitter(); + + private final SsfHttpClient http = new JdkSsfHttpClient(); + + private final List handled = new ArrayList<>(); + + @AfterAll + static void stopTransmitter() { + transmitter.close(); + } + + @Test + void pollerFetchesAsSoonAsTheStreamIsRegistered() { + transmitter.reset(); + transmitter.setAvailable(true); + SsfReceiverStream receiverStream = new SsfReceiverStream(); + SsfTransmitterMetadataResolver metadataResolver = new SsfTransmitterMetadataResolver(transmitter.issuer(), null, + this.http); + SsfTransmitterTokenProvider tokenProvider = () -> TestTransmitter.ACCESS_TOKEN; + SsfStreamClient streamClient = new SsfStreamClient(this.http, tokenProvider, metadataResolver); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(streamClient, receiverStream, + SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + NimbusSsfSetVerifier verifier = new NimbusSsfSetVerifier(transmitter.issuer(), transmitter::jwksUri, this.http); + SsfEventHandler handler = (eventContext) -> this.handled.add(eventContext.eventToken().jti()); + SsfSetProcessor processor = new SsfSetProcessor(verifier, new InMemorySsfJtiDedupStore(10), List.of(handler)); + SsfPoller poller = new SsfPoller(this.http, tokenProvider, + () -> receiverStream.getConfiguration().map(SsfStreamConfiguration::deliveryEndpointUrl).orElse(null), + processor); + // only the wake-up of the registration can make the poller fetch within the test + poller.setInterval(Duration.ofHours(1)); + poller.setEndpointRetry(Duration.ofHours(1)); + SsfTransmitter ssfTransmitter = SsfTransmitter.builder("test", transmitter.issuer()) + .metadataResolver(metadataResolver) + .verifier(verifier) + .tokenProvider(tokenProvider) + .streamClient(streamClient) + .receiverStream(receiverStream) + .streamRegistrar(registrar) + .poller(poller, true) + .build(); + var claims = transmitter.setClaims("CaepSessionRevoked", opaque("session-1")).build(); + transmitter.queueSet(transmitter.signSet(claims)); + + ssfTransmitter.start(); + try { + await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> { + assertThat(registrar.getState()).isEqualTo(SsfStreamRegistrar.State.REGISTERED); + assertThat(this.handled).containsExactly(claims.getJWTID()); + }); + assertThat(URI.create(transmitter.pollUri())) + .isEqualTo(receiverStream.getConfiguration().orElseThrow().deliveryEndpointUrl()); + } + finally { + ssfTransmitter.stop(); + } + } + +} From 6398f4e4c27a5bc44ba9de3a5c8e7311f2d298c5 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Tue, 6 Oct 2026 09:51:04 +0200 Subject: [PATCH 2/2] Open the test packages to JUnit for runs from the IDE IntelliJ runs the tests of an explicit module with JUnit on the module path, where surefire keeps it on the class path, so the IDE failed with "does not opens ... to module org.junit.platform.commons" since the module descriptors came in. A profile active only in the IDE hands surefire an argLine with an add-opens per test package, which the IDE copies into its run configurations; Maven itself is unaffected. Signed-off-by: Thomas Darimont --- AGENTS.md | 5 +++-- CONTRIBUTING.md | 9 ++++++++- easyssf-core/pom.xml | 9 +++++++++ easyssf-receiver-jdbc/pom.xml | 4 ++++ easyssf-receiver/pom.xml | 15 +++++++++++++++ pom.xml | 27 +++++++++++++++++++++++++++ 6 files changed, 66 insertions(+), 3 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index d94fae5..c51e9f9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -97,8 +97,9 @@ pins a released version. Consequences for a change here: - **Handlers are idempotent** and a SET that any handler fails on is delivered again. State that must survive a restart belongs in a store interface with an in-memory and a JDBC implementation. - **Java modules.** `core`, `test`, `receiver` and `receiver-jdbc` have a `module-info.java`; a new - package needs an `exports` line or the dependants do not compile. Micrometer stays - `requires static`. + package needs an `exports` line or the dependants do not compile, and a new test package needs + an `--add-opens` line in the module's `ide.test.add-opens` property or the IDE cannot run its + tests (Maven can, see CONTRIBUTING.md). Micrometer stays `requires static`. - **Logging** is SLF4J with string concatenation in the receiver (the messages are built once per event) and named by the event alias, e.g. `CaepSessionRevoked: terminated 2 session(s) of ...`. - No Lombok, no reflection-based mapping, no new runtime dependencies in `core` and `receiver` diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index bd27061..4659d1a 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -40,7 +40,14 @@ to submit the change under the project's license. The CI checks it for every com manifest entry, which every module sets through the `automatic.module.name` property. A new package in an explicit module needs an `exports` line. Surefire runs the tests of the explicit modules on the module path; `easyssf-receiver-jdbc` adds `java.naming` for test compilation - because H2's data source implements a `javax.naming` interface. + because H2's data source implements a `javax.naming` interface. Surefire keeps JUnit on the class + path and opens the test packages to it; IntelliJ IDEA puts JUnit on the module path instead, so + the profile `ide` of the root POM, active only in the IDE (it activates on the `idea.version` + property the IDE sets), hands surefire an `argLine` that opens every test package of an explicit + module to `org.junit.platform.commons`, which the IDE copies into its JUnit run configurations. + The packages are listed in the property `ide.test.add-opens` of each explicit module; a new test + package there needs a line, or the IDE fails with "does not opens ... to module + org.junit.platform.commons" while Maven passes. - **Every module ships an SBOM**, `target/bom.json` in CycloneDX format, attached to the artifact with the classifier `cyclonedx`. It lists the compile and runtime dependencies. - **`.mvn/jvm.config`** silences the schema validator inside the CycloneDX plugin, which warns about diff --git a/easyssf-core/pom.xml b/easyssf-core/pom.xml index 4912327..d7a329c 100644 --- a/easyssf-core/pom.xml +++ b/easyssf-core/pom.xml @@ -13,6 +13,15 @@ org.easyssf.core + + + --add-opens org.easyssf.core/org.easyssf.core=org.junit.platform.commons + --add-opens org.easyssf.core/org.easyssf.core.caep=org.junit.platform.commons + --add-opens org.easyssf.core/org.easyssf.core.event=org.junit.platform.commons + --add-opens org.easyssf.core/org.easyssf.core.metadata=org.junit.platform.commons + --add-opens org.easyssf.core/org.easyssf.core.risc=org.junit.platform.commons + --add-opens org.easyssf.core/org.easyssf.core.scim=org.junit.platform.commons + diff --git a/easyssf-receiver-jdbc/pom.xml b/easyssf-receiver-jdbc/pom.xml index 9c672ed..caea4b2 100644 --- a/easyssf-receiver-jdbc/pom.xml +++ b/easyssf-receiver-jdbc/pom.xml @@ -18,6 +18,10 @@ database org.easyssf.receiver.jdbc + + + --add-opens org.easyssf.receiver.jdbc/org.easyssf.receiver.jdbc=org.junit.platform.commons + diff --git a/easyssf-receiver/pom.xml b/easyssf-receiver/pom.xml index 4969fe5..700a53c 100644 --- a/easyssf-receiver/pom.xml +++ b/easyssf-receiver/pom.xml @@ -13,6 +13,21 @@ org.easyssf.receiver + + + --add-opens org.easyssf.receiver/org.easyssf.receiver.caep=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.event=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.http=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.poll=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.push=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.revocation=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.risc=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.scim=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.session=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.set=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.stream=org.junit.platform.commons + --add-opens org.easyssf.receiver/org.easyssf.receiver.transmitter=org.junit.platform.commons + diff --git a/pom.xml b/pom.xml index d950102..d1494c6 100644 --- a/pom.xml +++ b/pom.xml @@ -58,6 +58,8 @@ 21 + + ${java.version} UTF-8 UTF-8 @@ -400,6 +402,31 @@ + + + ide + + + idea.version + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + ${ide.test.add-opens} + + + + +