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