Wake the poller up when the stream is registered

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`

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

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

easyssf-core/pom.xml

@@ -13,6 +13,15 @@
 
     <properties>
         <automatic.module.name>org.easyssf.core</automatic.module.name>
+        <!-- the test packages, opened to JUnit when the IDE runs the tests (profile "ide" of the root POM) -->
+        <ide.test.add-opens>
+            --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
+        </ide.test.add-opens>
     </properties>
 
     <dependencies>

easyssf-receiver-jdbc/pom.xml

@@ -18,6 +18,10 @@
         <database.groups/>
         <database.excludedGroups>database</database.excludedGroups>
         <automatic.module.name>org.easyssf.receiver.jdbc</automatic.module.name>
+        <!-- the test packages, opened to JUnit when the IDE runs the tests (profile "ide" of the root POM) -->
+        <ide.test.add-opens>
+            --add-opens org.easyssf.receiver.jdbc/org.easyssf.receiver.jdbc=org.junit.platform.commons
+        </ide.test.add-opens>
     </properties>
 
     <dependencies>

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

easyssf-receiver/pom.xml

@@ -13,6 +13,21 @@
 
     <properties>
         <automatic.module.name>org.easyssf.receiver</automatic.module.name>
+        <!-- the test packages, opened to JUnit when the IDE runs the tests (profile "ide" of the root POM) -->
+        <ide.test.add-opens>
+            --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
+        </ide.test.add-opens>
     </properties>
 
     <dependencies>

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

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.
+ *
+ * <p>
+ * {@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<Listener> 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;
     }
 

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());
+        }
     }
 
     /**

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

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<SsfStreamConfiguration> 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,