rows = new ArrayList<>();
diff --git a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfPollAckStore.java b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfPollAckStore.java
new file mode 100644
index 0000000..92410c7
--- /dev/null
+++ b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfPollAckStore.java
@@ -0,0 +1,157 @@
+package org.easyssf.receiver.jdbc;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collection;
+import java.util.List;
+
+import org.easyssf.core.support.SsfAssert;
+import org.easyssf.receiver.poll.SsfPendingAck;
+import org.easyssf.receiver.poll.SsfPollAckStore;
+
+/**
+ * {@link SsfPollAckStore} that keeps the acknowledgements a poller owes its transmitter
+ * in a database table, so that they survive a restart: a SET handled right before the
+ * application stopped is acknowledged with the first poll after it started again, instead
+ * of being delivered once more.
+ *
+ *
+ * Entries a transmitter never accepts are forgotten after the retention time (7 days by
+ * default): the store deletes them when it is written to, at most once a minute, and
+ * {@link #purgeExpired()} does so on demand, see {@link JdbcSsfStoreCleanup}.
+ *
+ * @see JdbcSsfSchema
+ */
+public class JdbcSsfPollAckStore implements SsfPollAckStore, JdbcSsfExpiringStore {
+
+ private static final Duration CLEANUP_INTERVAL = Duration.ofMinutes(1);
+
+ private static final int MAX_DESCRIPTION_LENGTH = 1024;
+
+ /**
+ * How many entries one DELETE removes by default; databases limit the number of
+ * parameters of a statement.
+ */
+ public static final int DEFAULT_DELETE_BATCH_SIZE = 100;
+
+ private final SsfJdbcOperations jdbc;
+
+ private final String insert;
+
+ private final String select;
+
+ private final String deletePrefix;
+
+ private final String count;
+
+ private final String deleteExpired;
+
+ private Duration retention = Duration.ofDays(7);
+
+ private int deleteBatchSize = DEFAULT_DELETE_BATCH_SIZE;
+
+ private Clock clock = Clock.systemUTC();
+
+ private volatile Instant nextCleanup = Instant.MIN;
+
+ /**
+ * @param jdbc used to access the database
+ * @param tablePrefix the prefix of the table name, see
+ * {@link JdbcSsfSchema#DEFAULT_TABLE_PREFIX}
+ */
+ public JdbcSsfPollAckStore(SsfJdbcOperations jdbc, String tablePrefix) {
+ SsfAssert.notNull(jdbc, "jdbc must not be null");
+ String table = JdbcSsfSchema.pollAckTable(tablePrefix);
+ this.jdbc = jdbc;
+ this.insert = "INSERT INTO " + table
+ + " (ISSUER, JTI, ERROR_CODE, ERROR_DESCRIPTION, RECORDED_AT) VALUES (?, ?, ?, ?, ?)";
+ // the limit is applied by the JDBC driver (Statement.setMaxRows), the SQL for it
+ // differs
+ // between databases
+ this.select = "SELECT JTI, ERROR_CODE, ERROR_DESCRIPTION FROM " + table
+ + " WHERE ISSUER = ? ORDER BY RECORDED_AT, JTI";
+ this.deletePrefix = "DELETE FROM " + table + " WHERE ISSUER = ? AND JTI IN (";
+ this.count = "SELECT COUNT(*) FROM " + table + " WHERE ISSUER = ?";
+ this.deleteExpired = "DELETE FROM " + table + " WHERE RECORDED_AT < ?";
+ }
+
+ /**
+ * Sets how long an entry the transmitter does not accept is kept.
+ */
+ public void setRetention(Duration retention) {
+ SsfAssert.isTrue(retention != null && retention.isPositive(), "retention must be positive");
+ this.retention = retention;
+ }
+
+ /**
+ * Sets how many entries one DELETE statement removes ({@code JTI IN (...)}), 100 by
+ * default. Databases limit the number of parameters of a statement (SQL Server to
+ * 2100, Oracle to 1000 list members); a smaller batch means more statements per poll.
+ */
+ public void setDeleteBatchSize(int deleteBatchSize) {
+ SsfAssert.isTrue(deleteBatchSize > 0, "deleteBatchSize must be positive");
+ this.deleteBatchSize = deleteBatchSize;
+ }
+
+ public void setClock(Clock clock) {
+ SsfAssert.notNull(clock, "clock must not be null");
+ this.clock = clock;
+ }
+
+ @Override
+ public void record(String issuer, SsfPendingAck ack) {
+ Instant now = this.clock.instant();
+ removeExpired(now);
+ String description = ack.errorDescription();
+ if (description != null && description.length() > MAX_DESCRIPTION_LENGTH) {
+ description = description.substring(0, MAX_DESCRIPTION_LENGTH);
+ }
+ try {
+ this.jdbc.update(this.insert, issuer, ack.jti(), ack.errorCode(), description, now.toEpochMilli());
+ }
+ catch (SsfJdbcDuplicateKeyException ex) {
+ // recorded already, for example a SET delivered twice before it was
+ // acknowledged
+ }
+ }
+
+ @Override
+ public List pending(String issuer, int limit) {
+ return this.jdbc.query(this.select, limit,
+ (row) -> new SsfPendingAck(row.getString(1), row.getString(2), row.getString(3)), issuer);
+ }
+
+ @Override
+ public void remove(String issuer, Collection jtis) {
+ List remaining = List.copyOf(jtis);
+ for (int from = 0; from < remaining.size(); from += this.deleteBatchSize) {
+ List chunk = remaining.subList(from, Math.min(from + this.deleteBatchSize, remaining.size()));
+ Object[] args = new Object[chunk.size() + 1];
+ args[0] = issuer;
+ for (int i = 0; i < chunk.size(); i++) {
+ args[i + 1] = chunk.get(i);
+ }
+ this.jdbc.update(this.deletePrefix + "?, ".repeat(chunk.size() - 1) + "?)", args);
+ }
+ }
+
+ @Override
+ public int size(String issuer) {
+ return this.jdbc.query(this.count, (row) -> row.getInt(1), issuer).get(0);
+ }
+
+ @Override
+ public int purgeExpired() {
+ Instant now = this.clock.instant();
+ this.nextCleanup = now.plus(CLEANUP_INTERVAL);
+ return this.jdbc.update(this.deleteExpired, now.minus(this.retention).toEpochMilli());
+ }
+
+ private void removeExpired(Instant now) {
+ if (now.isAfter(this.nextCleanup)) {
+ purgeExpired();
+ }
+ }
+
+}
diff --git a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfSchema.java b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfSchema.java
index 11d7914..b4a3066 100644
--- a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfSchema.java
+++ b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfSchema.java
@@ -39,6 +39,13 @@ public static String revocationTable(String tablePrefix) {
return validPrefix(tablePrefix) + "REVOCATION";
}
+ /**
+ * The name of the table of the {@link JdbcSsfPollAckStore}.
+ */
+ public static String pollAckTable(String tablePrefix) {
+ return validPrefix(tablePrefix) + "POLL_ACK";
+ }
+
private static String validPrefix(String tablePrefix) {
SsfAssert.isTrue(tablePrefix != null && tablePrefix.matches("[A-Za-z0-9_.]*"),
"The table prefix may only consist of letters, digits, '_' and '.'");
@@ -77,6 +84,25 @@ ID VARCHAR(255) NOT NULL,
)""".formatted(table, name), "CREATE INDEX %s_IX1 ON %s (EXPIRES_AT)".formatted(name, table));
}
+ /**
+ * The statements that create the table of the {@link JdbcSsfPollAckStore}.
+ */
+ public static List createPollAckTable(String tablePrefix) {
+ String table = pollAckTable(tablePrefix);
+ String name = table.replace('.', '_');
+ String tableDdl = """
+ CREATE TABLE %s (
+ ISSUER VARCHAR(255) NOT NULL,
+ JTI VARCHAR(255) NOT NULL,
+ ERROR_CODE VARCHAR(64),
+ ERROR_DESCRIPTION VARCHAR(1024),
+ RECORDED_AT BIGINT NOT NULL,
+ CONSTRAINT %s_PK PRIMARY KEY (ISSUER, JTI)
+ )""".formatted(table, name);
+ String indexDdl = "CREATE INDEX %s_IX1 ON %s (RECORDED_AT)".formatted(name, table);
+ return List.of(tableDdl, indexDdl);
+ }
+
/**
* Makes sure a table exists.
* @param jdbc runs the statements
diff --git a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/SsfJdbcOperations.java b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/SsfJdbcOperations.java
index 73d7b1b..1e08bf7 100644
--- a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/SsfJdbcOperations.java
+++ b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/SsfJdbcOperations.java
@@ -29,6 +29,24 @@ public interface SsfJdbcOperations {
*/
List query(String sql, SsfJdbcRowMapper mapper, Object... args);
+ /**
+ * Runs a query and returns at most {@code maxRows} rows. The limit is applied with
+ * {@code Statement.setMaxRows}, which every JDBC driver supports, rather than with
+ * the SQL for it, which differs between databases ({@code FETCH FIRST},
+ * {@code LIMIT}, {@code TOP}). This default runs the query without a limit and cuts
+ * the result; implementations should override it.
+ * @param sql the query with {@code ?} placeholders
+ * @param maxRows the most rows to return, positive
+ * @param mapper maps each row of the result
+ * @param args the values of the placeholders
+ * @return the mapped rows
+ * @throws SsfJdbcException if the query fails
+ */
+ default List query(String sql, int maxRows, SsfJdbcRowMapper mapper, Object... args) {
+ List rows = query(sql, mapper, args);
+ return (rows.size() > maxRows) ? List.copyOf(rows.subList(0, maxRows)) : rows;
+ }
+
/**
* Runs a statement that returns nothing, for example DDL.
* @throws SsfJdbcException if the statement fails
diff --git a/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql b/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql
index fa7ca3e..1696710 100644
--- a/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql
+++ b/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql
@@ -20,3 +20,15 @@ CREATE TABLE EASYSSF_REVOCATION (
CONSTRAINT EASYSSF_REVOCATION_PK PRIMARY KEY (KIND, ISSUER, ID)
);
CREATE INDEX EASYSSF_REVOCATION_IX1 ON EASYSSF_REVOCATION (EXPIRES_AT);
+
+-- acknowledgements and error reports a poller owes its transmitter until a poll request carried them,
+-- see JdbcSsfPollAckStore (ERROR_CODE NULL: acknowledgement)
+CREATE TABLE EASYSSF_POLL_ACK (
+ ISSUER VARCHAR(255) NOT NULL,
+ JTI VARCHAR(255) NOT NULL,
+ ERROR_CODE VARCHAR(64),
+ ERROR_DESCRIPTION VARCHAR(1024),
+ RECORDED_AT BIGINT NOT NULL,
+ CONSTRAINT EASYSSF_POLL_ACK_PK PRIMARY KEY (ISSUER, JTI)
+);
+CREATE INDEX EASYSSF_POLL_ACK_IX1 ON EASYSSF_POLL_ACK (RECORDED_AT);
diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java
new file mode 100644
index 0000000..4a51731
--- /dev/null
+++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java
@@ -0,0 +1,183 @@
+package org.easyssf.receiver.jdbc;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.easyssf.core.event.SsfSubjectIdentifiers;
+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.NimbusSsfSetVerifier;
+import org.easyssf.receiver.set.SsfSetProcessor;
+import org.easyssf.test.TestTransmitter;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+
+/**
+ * Several receivers (instances of one application) polling the same stream and sharing
+ * the JDBC stores: a SET is handled once, and an acknowledgement one instance could not
+ * deliver leaves with the next poll of another.
+ */
+abstract class AbstractJdbcSharedReceiversTests {
+
+ /**
+ * One transmitter per test: a static one would be shared by the subclasses for the
+ * different databases and closed by the first of them to finish.
+ */
+ private TestTransmitter transmitter;
+
+ private static final String PREFIX = JdbcSsfSchema.DEFAULT_TABLE_PREFIX;
+
+ private JdbcSsfJtiDedupStore dedupStore;
+
+ private JdbcSsfPollAckStore ackStore;
+
+ /**
+ * Returns operations on an empty database: none of the tables of the tests exist.
+ */
+ protected abstract SsfJdbcOperations emptyDatabase();
+
+ @BeforeEach
+ void setUp() {
+ this.transmitter = new TestTransmitter();
+ SsfJdbcOperations jdbc = emptyDatabase();
+ JdbcSsfSchema.prepareTable(jdbc, JdbcSsfSchema.processedSetTable(PREFIX),
+ JdbcSsfSchema.createProcessedSetTable(PREFIX), true, null);
+ JdbcSsfSchema.prepareTable(jdbc, JdbcSsfSchema.pollAckTable(PREFIX), JdbcSsfSchema.createPollAckTable(PREFIX),
+ true, null);
+ this.dedupStore = new JdbcSsfJtiDedupStore(jdbc, PREFIX);
+ this.ackStore = new JdbcSsfPollAckStore(jdbc, PREFIX);
+ }
+
+ @AfterEach
+ void stopTransmitter() {
+ this.transmitter.close();
+ }
+
+ @Test
+ void setIsHandledOnceAndAcknowledgedByWhicheverInstancePollsNext() {
+ Receiver a = receiver();
+ Receiver b = receiver();
+ String jti = queue("session-1");
+ // a handles the SET, then loses the transmitter before the acknowledgement leaves
+ a.afterHandling = () -> this.transmitter.setAvailable(false);
+
+ assertThatIllegalStateException().isThrownBy(a.poller::pollNow);
+
+ assertThat(a.handled).containsExactly(jti);
+ assertThat(this.transmitter.acknowledgedSets()).isEmpty();
+ // the acknowledgement waits in the shared store, visible to both instances
+ assertThat(a.poller.getPendingAckCount()).isEqualTo(1);
+ assertThat(b.poller.getPendingAckCount()).isEqualTo(1);
+
+ // b polls: its request carries a's acknowledgement from the shared store, so the
+ // transmitter
+ // releases the SET before answering and b does not even see it
+ this.transmitter.setAvailable(true);
+ assertThat(b.poller.pollNow()).isZero();
+
+ assertThat(b.handled).isEmpty();
+ assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti);
+ assertThat(a.poller.getPendingAckCount()).isZero();
+ assertThat(b.poller.getPendingAckCount()).isZero();
+ assertThat(this.transmitter.queuedSetCount()).isZero();
+ }
+
+ @Test
+ void errorReportOfOneInstanceLeavesWithTheNextPollOfAnother() {
+ Receiver a = receiver();
+ Receiver b = receiver();
+ String jti = queue("session-1");
+ this.transmitter.queueSet("broken", "not-a-set");
+ a.afterHandling = () -> this.transmitter.setAvailable(false);
+
+ assertThatIllegalStateException().isThrownBy(a.poller::pollNow);
+ assertThat(this.transmitter.reportedErrors()).isEmpty();
+
+ this.transmitter.setAvailable(true);
+ b.poller.pollNow();
+
+ assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti);
+ assertThat(this.transmitter.reportedErrors()).containsOnlyKeys("broken");
+ assertThat(b.handled).isEmpty();
+ }
+
+ @Test
+ void setDeliveredAgainLaterIsNotHandledTwiceByAnyInstance() {
+ Receiver a = receiver();
+ Receiver b = receiver();
+ String set = this.transmitter.set("CaepSessionRevoked", SsfSubjectIdentifiers.opaque("session-1"));
+ String jti = jti(set);
+ this.transmitter.queueSet(set);
+ assertThat(a.poller.pollNow()).isEqualTo(1);
+ assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti);
+
+ // the transmitter delivers the same SET once more, to the other instance
+ this.transmitter.queueSet(jti, set);
+ assertThat(b.poller.pollNow()).isEqualTo(1);
+
+ assertThat(a.handled).containsExactly(jti);
+ assertThat(b.handled).isEmpty();
+ assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti, jti);
+ }
+
+ private Receiver receiver() {
+ Receiver receiver = new Receiver();
+ SsfHttpClient http = new JdkSsfHttpClient();
+ NimbusSsfSetVerifier verifier = new NimbusSsfSetVerifier(this.transmitter.issuer(), transmitter::jwksUri, http);
+ SsfEventHandler handler = (eventContext) -> {
+ receiver.handled.add(eventContext.eventToken().jti());
+ receiver.afterHandling.run();
+ };
+ SsfSetProcessor processor = new SsfSetProcessor(verifier, this.dedupStore, List.of(handler));
+ receiver.poller = new SsfPoller(http, () -> TestTransmitter.ACCESS_TOKEN,
+ () -> URI.create(this.transmitter.pollUri()), processor);
+ receiver.poller.setAckStore(this.ackStore);
+ receiver.poller.setTransmitter(this.transmitter.issuer());
+ receiver.poller.setFlushOnStop(false);
+ return receiver;
+ }
+
+ private String queue(String sessionId) {
+ String set = this.transmitter.set("CaepSessionRevoked", SsfSubjectIdentifiers.opaque(sessionId));
+ this.transmitter.queueSet(set);
+ return jti(set);
+ }
+
+ /**
+ * The {@code jti} of a SET, read from its payload without a JOSE library: this module
+ * does not read Nimbus on the module path.
+ */
+ private static String jti(String encodedSet) {
+ String payload = new String(Base64.getUrlDecoder().decode(encodedSet.split("\\.")[1]), StandardCharsets.UTF_8);
+ Matcher jti = Pattern.compile("\"jti\"\\s*:\\s*\"([^\"]+)\"").matcher(payload);
+ assertThat(jti.find()).isTrue();
+ return jti.group(1);
+ }
+
+ /**
+ * One instance of the application: its poller and what its handler saw.
+ */
+ private static final class Receiver {
+
+ private SsfPoller poller;
+
+ private final List handled = new ArrayList<>();
+
+ /** runs after a SET was handled, before it is acknowledged */
+ private Runnable afterHandling = () -> {
+ };
+
+ }
+
+}
diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java
index 89505f0..2d0ac82 100644
--- a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java
+++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java
@@ -12,6 +12,7 @@
import java.util.Map;
import org.easyssf.core.event.SsfEventToken;
+import org.easyssf.receiver.poll.SsfPendingAck;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -47,6 +48,15 @@ void createTables() {
JdbcSsfSchema.createProcessedSetTable(PREFIX), true, null);
JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.revocationTable(PREFIX),
JdbcSsfSchema.createRevocationTable(PREFIX), true, null);
+ JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.pollAckTable(PREFIX),
+ JdbcSsfSchema.createPollAckTable(PREFIX), true, null);
+ }
+
+ private JdbcSsfPollAckStore ackStore() {
+ JdbcSsfPollAckStore store = new JdbcSsfPollAckStore(this.jdbc, PREFIX);
+ store.setRetention(Duration.ofDays(1));
+ store.setClock(this.clock);
+ return store;
}
private JdbcSsfJtiDedupStore dedupStore() {
@@ -71,6 +81,7 @@ void schemaFileMatchesTheStatementsThatCreateTheTables() throws Exception {
.toList();
List expected = new ArrayList<>(JdbcSsfSchema.createProcessedSetTable(PREFIX));
expected.addAll(JdbcSsfSchema.createRevocationTable(PREFIX));
+ expected.addAll(JdbcSsfSchema.createPollAckTable(PREFIX));
assertThat(statements).isEqualTo(expected.stream().map(AbstractJdbcSsfStoresTests::normalize).toList());
}
@@ -201,6 +212,71 @@ void revocationsExpireAndAreRemoved() {
.containsExactly("session-2");
}
+ @Test
+ void acknowledgementsWaitUntilRemovedAndAreSharedBetweenInstances() {
+ JdbcSsfPollAckStore store = ackStore();
+ store.record(ISSUER, SsfPendingAck.ack("jti-1"));
+ this.clock.advance(Duration.ofSeconds(1));
+ store.record(ISSUER, SsfPendingAck.error("jti-2", "invalid_request", "not a SET"));
+ store.record(ISSUER, SsfPendingAck.ack("jti-2"));
+ store.record("https://other.example", SsfPendingAck.ack("jti-1"));
+
+ assertThat(ackStore().size(ISSUER)).isEqualTo(2);
+ List pending = ackStore().pending(ISSUER, 10);
+ assertThat(pending).extracting(SsfPendingAck::jti).containsExactly("jti-1", "jti-2");
+ assertThat(pending.get(0).isError()).isFalse();
+ assertThat(pending.get(1).errorCode()).isEqualTo("invalid_request");
+ assertThat(pending.get(1).errorDescription()).isEqualTo("not a SET");
+ assertThat(ackStore().pending(ISSUER, 1)).extracting(SsfPendingAck::jti).containsExactly("jti-1");
+
+ ackStore().remove(ISSUER, List.of("jti-1", "jti-unknown"));
+ assertThat(ackStore().pending(ISSUER, 10)).extracting(SsfPendingAck::jti).containsExactly("jti-2");
+ assertThat(ackStore().size("https://other.example")).isEqualTo(1);
+ }
+
+ @Test
+ void manyAcknowledgementsAreListedWithALimitAndRemovedInChunks() {
+ JdbcSsfPollAckStore store = ackStore();
+ store.setDeleteBatchSize(50);
+ for (int i = 0; i < 250; i++) {
+ store.record(ISSUER, SsfPendingAck.ack("jti-%03d".formatted(i)));
+ this.clock.advance(Duration.ofMillis(1));
+ }
+ assertThat(store.size(ISSUER)).isEqualTo(250);
+ List pending = store.pending(ISSUER, 120);
+ assertThat(pending).hasSize(120);
+ assertThat(pending.get(0).jti()).isEqualTo("jti-000");
+ assertThat(pending.get(119).jti()).isEqualTo("jti-119");
+
+ store.remove(ISSUER, pending.stream().map(SsfPendingAck::jti).toList());
+ assertThat(store.size(ISSUER)).isEqualTo(130);
+ assertThat(store.pending(ISSUER, 1).get(0).jti()).isEqualTo("jti-120");
+ store.remove(ISSUER, List.of());
+ assertThat(store.size(ISSUER)).isEqualTo(130);
+ }
+
+ @Test
+ void longErrorDescriptionsAreCut() {
+ JdbcSsfPollAckStore store = ackStore();
+ store.record(ISSUER, SsfPendingAck.error("jti-1", "invalid_request", "x".repeat(5000)));
+ assertThat(store.pending(ISSUER, 1).get(0).errorDescription()).hasSize(1024);
+ }
+
+ @Test
+ void acknowledgementsAreForgottenAfterTheRetentionTime() {
+ JdbcSsfPollAckStore store = ackStore();
+ store.record(ISSUER, SsfPendingAck.ack("old"));
+ this.clock.advance(Duration.ofHours(12));
+ store.record(ISSUER, SsfPendingAck.ack("recent"));
+ this.clock.advance(Duration.ofHours(12).plusSeconds(1));
+ assertThat(store.purgeExpired()).isEqualTo(1);
+ assertThat(store.pending(ISSUER, 10)).extracting(SsfPendingAck::jti).containsExactly("recent");
+ // writing purges as well, at most once a minute
+ this.clock.advance(Duration.ofDays(1));
+ store.record(ISSUER, SsfPendingAck.ack("newer"));
+ assertThat(store.pending(ISSUER, 10)).extracting(SsfPendingAck::jti).containsExactly("newer");
+ }
+
@Test
void purgeExpiredDeletesOnlyTheExpiredRows() {
// processed SETs are retained for a day in this test
diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/JdbcSharedReceiversTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/JdbcSharedReceiversTests.java
new file mode 100644
index 0000000..fa3d82d
--- /dev/null
+++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/JdbcSharedReceiversTests.java
@@ -0,0 +1,19 @@
+package org.easyssf.receiver.jdbc;
+
+import java.util.UUID;
+
+import org.h2.jdbcx.JdbcDataSource;
+
+/**
+ * Several receivers on one H2 database, a new in-memory database for every test.
+ */
+class JdbcSharedReceiversTests extends AbstractJdbcSharedReceiversTests {
+
+ @Override
+ protected SsfJdbcOperations emptyDatabase() {
+ JdbcDataSource dataSource = new JdbcDataSource();
+ dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
+ return new DataSourceSsfJdbcOperations(dataSource);
+ }
+
+}
diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSharedReceiversTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSharedReceiversTests.java
new file mode 100644
index 0000000..924100c
--- /dev/null
+++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSharedReceiversTests.java
@@ -0,0 +1,51 @@
+package org.easyssf.receiver.jdbc;
+
+import java.util.List;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.testcontainers.mysql.MySQLContainer;
+
+import com.mysql.cj.jdbc.MysqlDataSource;
+
+/**
+ * Several receivers on one MySQL, in a Testcontainers container shared by the tests of
+ * the class. Tagged {@code database}: the normal build excludes it,
+ * {@code -Pdatabase-tests} runs it (Docker required).
+ */
+@Tag("database")
+class MySqlJdbcSharedReceiversTests extends AbstractJdbcSharedReceiversTests {
+
+ private static final String IMAGE = "mysql:8.4";
+
+ private static final List TABLES = List.of("EASYSSF_PROCESSED_SET", "EASYSSF_REVOCATION",
+ "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION", "EASYSSF_POLL_ACK");
+
+ private static MySQLContainer mysql;
+
+ @BeforeAll
+ static void startDatabase() {
+ mysql = new MySQLContainer(IMAGE);
+ mysql.start();
+ }
+
+ @AfterAll
+ static void stopDatabase() {
+ mysql.stop();
+ }
+
+ @Override
+ protected SsfJdbcOperations emptyDatabase() {
+ MysqlDataSource dataSource = new MysqlDataSource();
+ dataSource.setUrl(mysql.getJdbcUrl());
+ dataSource.setUser(mysql.getUsername());
+ dataSource.setPassword(mysql.getPassword());
+ SsfJdbcOperations jdbc = new DataSourceSsfJdbcOperations(dataSource);
+ for (String table : TABLES) {
+ jdbc.execute("DROP TABLE IF EXISTS " + table);
+ }
+ return jdbc;
+ }
+
+}
diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSsfStoresTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSsfStoresTests.java
new file mode 100644
index 0000000..6032b7a
--- /dev/null
+++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSsfStoresTests.java
@@ -0,0 +1,51 @@
+package org.easyssf.receiver.jdbc;
+
+import java.util.List;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.testcontainers.mysql.MySQLContainer;
+
+import com.mysql.cj.jdbc.MysqlDataSource;
+
+/**
+ * The stores on MySQL, in a Testcontainers container shared by the tests of the class.
+ * Tagged {@code database}: the normal build excludes it, {@code -Pdatabase-tests} runs it
+ * (Docker required).
+ */
+@Tag("database")
+class MySqlJdbcSsfStoresTests extends AbstractJdbcSsfStoresTests {
+
+ private static final String IMAGE = "mysql:8.4";
+
+ private static final List TABLES = List.of("EASYSSF_PROCESSED_SET", "EASYSSF_REVOCATION",
+ "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION", "EASYSSF_POLL_ACK");
+
+ private static MySQLContainer mysql;
+
+ @BeforeAll
+ static void startDatabase() {
+ mysql = new MySQLContainer(IMAGE);
+ mysql.start();
+ }
+
+ @AfterAll
+ static void stopDatabase() {
+ mysql.stop();
+ }
+
+ @Override
+ protected SsfJdbcOperations emptyDatabase() {
+ MysqlDataSource dataSource = new MysqlDataSource();
+ dataSource.setUrl(mysql.getJdbcUrl());
+ dataSource.setUser(mysql.getUsername());
+ dataSource.setPassword(mysql.getPassword());
+ SsfJdbcOperations jdbc = new DataSourceSsfJdbcOperations(dataSource);
+ for (String table : TABLES) {
+ jdbc.execute("DROP TABLE IF EXISTS " + table);
+ }
+ return jdbc;
+ }
+
+}
diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSharedReceiversTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSharedReceiversTests.java
new file mode 100644
index 0000000..775b694
--- /dev/null
+++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSharedReceiversTests.java
@@ -0,0 +1,50 @@
+package org.easyssf.receiver.jdbc;
+
+import java.util.List;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.postgresql.ds.PGSimpleDataSource;
+import org.testcontainers.postgresql.PostgreSQLContainer;
+
+/**
+ * Several receivers on one PostgreSQL, in a Testcontainers container shared by the tests
+ * of the class. Tagged {@code database}: the normal build excludes it,
+ * {@code -Pdatabase-tests} runs it (Docker required).
+ */
+@Tag("database")
+class PostgresJdbcSharedReceiversTests extends AbstractJdbcSharedReceiversTests {
+
+ private static final String IMAGE = "postgres:18.4";
+
+ private static final List TABLES = List.of("EASYSSF_PROCESSED_SET", "EASYSSF_REVOCATION",
+ "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION", "EASYSSF_POLL_ACK");
+
+ private static PostgreSQLContainer postgres;
+
+ @BeforeAll
+ static void startDatabase() {
+ postgres = new PostgreSQLContainer(IMAGE);
+ postgres.start();
+ }
+
+ @AfterAll
+ static void stopDatabase() {
+ postgres.stop();
+ }
+
+ @Override
+ protected SsfJdbcOperations emptyDatabase() {
+ PGSimpleDataSource dataSource = new PGSimpleDataSource();
+ dataSource.setUrl(postgres.getJdbcUrl());
+ dataSource.setUser(postgres.getUsername());
+ dataSource.setPassword(postgres.getPassword());
+ SsfJdbcOperations jdbc = new DataSourceSsfJdbcOperations(dataSource);
+ for (String table : TABLES) {
+ jdbc.execute("DROP TABLE IF EXISTS " + table);
+ }
+ return jdbc;
+ }
+
+}
diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSsfStoresTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSsfStoresTests.java
index 206540b..6d0ccbf 100644
--- a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSsfStoresTests.java
+++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSsfStoresTests.java
@@ -19,7 +19,7 @@ class PostgresJdbcSsfStoresTests extends AbstractJdbcSsfStoresTests {
private static final String IMAGE = "postgres:18.4";
private static final List TABLES = List.of("EASYSSF_PROCESSED_SET", "EASYSSF_REVOCATION",
- "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION");
+ "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION", "EASYSSF_POLL_ACK");
private static PostgreSQLContainer postgres;
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java
index 3971718..7241be8 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java
@@ -7,6 +7,7 @@
import java.util.Map;
import org.easyssf.core.event.SsfSubject;
+import org.easyssf.receiver.jdbc.JdbcSsfPollAckStore;
import org.easyssf.receiver.session.SsfSubjectClaimsMatcher;
import org.easyssf.receiver.transmitter.SsfTransmitter;
import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -261,6 +262,18 @@ public static class Jdbc {
*/
private Duration cleanupInterval = Duration.ofMinutes(15);
+ /**
+ * How long the JDBC store keeps an acknowledgement a polling receiver owes its
+ * transmitter if no poll request manages to deliver it.
+ */
+ private Duration ackRetention = Duration.ofDays(7);
+
+ /**
+ * How many acknowledgements one DELETE of the JDBC store removes at once;
+ * databases limit the parameters of a statement.
+ */
+ private int ackDeleteBatchSize = JdbcSsfPollAckStore.DEFAULT_DELETE_BATCH_SIZE;
+
public boolean isEnabled() {
return this.enabled;
}
@@ -293,6 +306,22 @@ public void setCleanupInterval(Duration cleanupInterval) {
this.cleanupInterval = cleanupInterval;
}
+ public Duration getAckRetention() {
+ return this.ackRetention;
+ }
+
+ public void setAckRetention(Duration ackRetention) {
+ this.ackRetention = ackRetention;
+ }
+
+ public int getAckDeleteBatchSize() {
+ return this.ackDeleteBatchSize;
+ }
+
+ public void setAckDeleteBatchSize(int ackDeleteBatchSize) {
+ this.ackDeleteBatchSize = ackDeleteBatchSize;
+ }
+
public enum InitializeSchema {
EMBEDDED, ALWAYS, NEVER
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java
index 768509a..c8b3d54 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java
@@ -1,6 +1,7 @@
package org.easyssf.receiver.spring.boot;
import java.net.URI;
+import java.time.Duration;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Stream;
@@ -9,8 +10,10 @@
import org.apache.commons.logging.LogFactory;
import org.easyssf.core.SsfDeliveryMethod;
import org.easyssf.core.stream.SsfStreamConfiguration;
+import org.easyssf.receiver.http.JdkSsfHttpClient;
import org.easyssf.receiver.http.SsfHttpClient;
import org.easyssf.receiver.metrics.SsfReceiverMetrics;
+import org.easyssf.receiver.poll.SsfPollAckStore;
import org.easyssf.receiver.poll.SsfPoller;
import org.easyssf.receiver.set.NimbusSsfSetVerifier;
import org.easyssf.receiver.set.SsfSetProcessor;
@@ -180,12 +183,62 @@ public static SsfStreamRegistrar streamRegistrar(String name, SsfTransmitterProp
public static SsfPoller poller(String name, SsfTransmitterProperties properties, SsfHttpClient httpClient,
SsfTransmitterTokenProvider tokenProvider, SsfSetProcessor processor, SsfReceiverStream receiverStream,
SsfReceiverMetrics metrics) {
+ return poller(name, properties, httpClient, tokenProvider, processor, receiverStream, metrics, null);
+ }
+
+ /**
+ * The HTTP client the poller of a transmitter calls the transmitter with: the given
+ * client, unless long polling is on and the client is not a {@link JdkSsfHttpClient}.
+ * A long poll waits longer than the read timeout of the application's
+ * {@code RestClient} allows, and the {@code RestClient} has no timeout per request,
+ * so the poller then gets a JDK client of its own, with the connect timeout and user
+ * agent of the {@code http} properties and a read timeout that covers the hold time.
+ * The auto-configuration uses it; an application that builds its own poller bean
+ * passes whatever client it wants to {@link #poller}.
+ * @param httpClient the client of the application
+ * @param http the {@code easyssf.receiver.http.*} properties, {@code null} for their
+ * defaults
+ */
+ public static SsfHttpClient pollHttpClient(String name, SsfTransmitterProperties properties,
+ SsfHttpClient httpClient, SsfReceiverProperties.Http http) {
+ SsfTransmitterProperties.Poll poll = properties.getPoll();
+ if (properties.getDeliveryMethod() != SsfDeliveryMethod.POLL || !poll.isLongPolling()
+ || httpClient instanceof JdkSsfHttpClient) {
+ return httpClient;
+ }
+ Duration connectTimeout = (http != null && http.getConnectTimeout() != null) ? http.getConnectTimeout()
+ : SsfReceiverProperties.Http.DEFAULT_TIMEOUT;
+ JdkSsfHttpClient pollClient = new JdkSsfHttpClient(connectTimeout, poll.getLongPollingHold().plusSeconds(15));
+ if (http != null) {
+ pollClient.setUserAgent(http.getUserAgent());
+ }
+ logger.info(prefix(name) + "poll.long-polling is on, the poller calls the transmitter with the JDK HTTP "
+ + "client rather than the RestClient of the application");
+ return pollClient;
+ }
+
+ /**
+ * @param httpClient the client the poller calls the transmitter with, see
+ * {@link #pollHttpClient} for the one the auto-configuration chooses
+ * @param ackStore keeps the acknowledgements the poller owes the transmitter,
+ * {@code null} for the in-memory default
+ * @return the poller, {@code null} with PUSH delivery
+ */
+ public static SsfPoller poller(String name, SsfTransmitterProperties properties, SsfHttpClient httpClient,
+ SsfTransmitterTokenProvider tokenProvider, SsfSetProcessor processor, SsfReceiverStream receiverStream,
+ SsfReceiverMetrics metrics, SsfPollAckStore ackStore) {
if (properties.getDeliveryMethod() != SsfDeliveryMethod.POLL) {
return null;
}
SsfTransmitterProperties.Poll poll = properties.getPoll();
SsfPoller poller = new SsfPoller(httpClient, tokenProvider, pollEndpoint(name, properties, receiverStream),
processor);
+ if (ackStore != null) {
+ poller.setAckStore(ackStore);
+ }
+ if (poll.isLongPolling()) {
+ poller.setLongPolling(poll.getLongPollingHold());
+ }
poller.setTransmitter(properties.getTransmitterIssuer());
poller.setMetrics((metrics != null) ? metrics : SsfReceiverMetrics.NOOP);
poller.setInterval(poll.getInterval());
@@ -201,6 +254,30 @@ public static SsfPoller poller(String name, SsfTransmitterProperties properties,
*/
public static SsfTransmitter.Builder builder(String name, SsfTransmitterProperties properties,
SsfHttpClient httpClient, SsfSetProcessor processor, SsfReceiverMetrics metrics) {
+ return builder(name, properties, httpClient, processor, metrics, null);
+ }
+
+ /**
+ * A builder with every part of the transmitter built from its properties.
+ * @param ackStore keeps the acknowledgements a poller owes the transmitter,
+ * {@code null} for the in-memory default
+ */
+ public static SsfTransmitter.Builder builder(String name, SsfTransmitterProperties properties,
+ SsfHttpClient httpClient, SsfSetProcessor processor, SsfReceiverMetrics metrics, SsfPollAckStore ackStore) {
+ return builder(name, properties, httpClient, processor, metrics, ackStore, null);
+ }
+
+ /**
+ * A builder with every part of the transmitter built from its properties.
+ * @param ackStore keeps the acknowledgements a poller owes the transmitter,
+ * {@code null} for the in-memory default
+ * @param http the {@code easyssf.receiver.http.*} properties, for the
+ * {@link #pollHttpClient poll client} of a long polling transmitter; {@code null} for
+ * their defaults
+ */
+ public static SsfTransmitter.Builder builder(String name, SsfTransmitterProperties properties,
+ SsfHttpClient httpClient, SsfSetProcessor processor, SsfReceiverMetrics metrics, SsfPollAckStore ackStore,
+ SsfReceiverProperties.Http http) {
SsfReceiverStream receiverStream = new SsfReceiverStream();
SsfTransmitterMetadataResolver metadataResolver = metadataResolver(name, properties, httpClient);
SsfTransmitterTokenProvider tokenProvider = tokenProvider(name, properties, httpClient);
@@ -213,8 +290,8 @@ public static SsfTransmitter.Builder builder(String name, SsfTransmitterProperti
.receiverStream(receiverStream)
.streamVerification(streamVerification(receiverStream))
.streamRegistrar(streamRegistrar(name, properties, streamClient, receiverStream))
- .poller(poller(name, properties, httpClient, tokenProvider, processor, receiverStream, metrics),
- properties.getPoll().isAutoStartup())
+ .poller(poller(name, properties, pollHttpClient(name, properties, httpClient, http), tokenProvider,
+ processor, receiverStream, metrics, ackStore), properties.getPoll().isAutoStartup())
.pushAuthorizationHeader(properties.getPush().getExpectedAuthHeader());
}
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterProperties.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterProperties.java
index 75a3a38..5e68dc9 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterProperties.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterProperties.java
@@ -363,6 +363,21 @@ public static class Poll {
*/
private int maxEvents = 100;
+ /**
+ * Whether to long poll (RFC 8936): the transmitter holds a request until SETs are
+ * available, and the next request goes out as soon as a response was handled.
+ * Otherwise the transmitter is asked to answer at once and polled every
+ * 'interval'. Requires a transmitter that holds requests; one that answers at
+ * once is polled every 'interval' nevertheless.
+ */
+ private boolean longPolling = false;
+
+ /**
+ * How long the transmitter holds a long poll request, part of the agreement with
+ * it. The request waits that long plus a margin for the response.
+ */
+ private Duration longPollingHold = Duration.ofSeconds(30);
+
private final RateLimit rateLimit = new RateLimit();
public URI getEndpointUrl() {
@@ -405,6 +420,22 @@ public void setMaxEvents(int maxEvents) {
this.maxEvents = maxEvents;
}
+ public boolean isLongPolling() {
+ return this.longPolling;
+ }
+
+ public void setLongPolling(boolean longPolling) {
+ this.longPolling = longPolling;
+ }
+
+ public Duration getLongPollingHold() {
+ return this.longPollingHold;
+ }
+
+ public void setLongPollingHold(Duration longPollingHold) {
+ this.longPollingHold = longPollingHold;
+ }
+
public RateLimit getRateLimit() {
return this.rateLimit;
}
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java
index 28c52f2..44fe97b 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java
@@ -12,6 +12,7 @@
import org.easyssf.receiver.http.JdkSsfHttpClient;
import org.easyssf.receiver.http.SsfHttpClient;
import org.easyssf.receiver.metrics.SsfReceiverMetrics;
+import org.easyssf.receiver.poll.SsfPollAckStore;
import org.easyssf.receiver.poll.SsfPoller;
import org.easyssf.receiver.set.InMemorySsfJtiDedupStore;
import org.easyssf.receiver.set.NimbusSsfSetVerifier;
@@ -167,9 +168,11 @@ SsfStreamRegistrar ssfStreamRegistrar(SsfReceiverProperties properties, SsfStrea
@ConditionalOnProperty(name = "easyssf.receiver.delivery-method", havingValue = "poll")
SsfPoller ssfPoller(SsfReceiverProperties properties, SsfHttpClient httpClient,
SsfTransmitterTokenProvider tokenProvider, SsfSetProcessor processor, SsfReceiverStream receiverStream,
- ObjectProvider metrics) {
- return SsfTransmitterFactory.poller(SsfTransmitter.DEFAULT_NAME, properties, httpClient, tokenProvider,
- processor, receiverStream, metrics.getIfAvailable());
+ ObjectProvider metrics, ObjectProvider ackStore) {
+ SsfHttpClient pollClient = SsfTransmitterFactory.pollHttpClient(SsfTransmitter.DEFAULT_NAME, properties,
+ httpClient, properties.getHttp());
+ return SsfTransmitterFactory.poller(SsfTransmitter.DEFAULT_NAME, properties, pollClient, tokenProvider,
+ processor, receiverStream, metrics.getIfAvailable(), ackStore.getIfAvailable());
}
// ---- shared by all transmitters
@@ -213,7 +216,7 @@ SsfTransmitters ssfTransmitters(SsfReceiverProperties properties, SsfHttpClient
ObjectProvider tokenProvider, ObjectProvider streamClient,
ObjectProvider receiverStream, ObjectProvider streamVerification,
ObjectProvider streamRegistrar, ObjectProvider poller,
- ObjectProvider customizers) {
+ ObjectProvider customizers, ObjectProvider ackStore) {
Map configured;
try {
configured = properties.getConfiguredTransmitters();
@@ -245,7 +248,7 @@ SsfTransmitters ssfTransmitters(SsfReceiverProperties properties, SsfHttpClient
}
else {
builder = SsfTransmitterFactory.builder(name, transmitterProperties, httpClient, processor,
- metrics.getIfAvailable());
+ metrics.getIfAvailable(), ackStore.getIfAvailable(), properties.getHttp());
}
customizers.orderedStream().forEach((customizer) -> customizer.customize(builder));
transmitters.add(builder.build());
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverJdbcAutoConfiguration.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverJdbcAutoConfiguration.java
index 9d2fab5..fd4d802 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverJdbcAutoConfiguration.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverJdbcAutoConfiguration.java
@@ -6,9 +6,11 @@
import org.apache.commons.logging.LogFactory;
import org.easyssf.receiver.jdbc.JdbcSsfExpiringStore;
import org.easyssf.receiver.jdbc.JdbcSsfJtiDedupStore;
+import org.easyssf.receiver.jdbc.JdbcSsfPollAckStore;
import org.easyssf.receiver.jdbc.JdbcSsfSchema;
import org.easyssf.receiver.jdbc.JdbcSsfTokenRevocationStore;
import org.easyssf.receiver.jdbc.SsfJdbcOperations;
+import org.easyssf.receiver.poll.SsfPollAckStore;
import org.easyssf.receiver.revocation.SsfTokenRevocationStore;
import org.easyssf.receiver.set.SsfJtiDedupStore;
import org.easyssf.receiver.spring.boot.SsfReceiverProperties;
@@ -87,6 +89,27 @@ JdbcSsfTokenRevocationStore jdbcSsfTokenRevocationStore(JdbcOperations jdbc, Obj
properties.getResourceServer().getRevocationTtl());
}
+ /**
+ * Keeps the acknowledgements a polling receiver owes its transmitter, so that a SET
+ * handled right before a restart is acknowledged afterwards instead of delivered
+ * again.
+ */
+ @Bean
+ @ConditionalOnMissingBean(SsfPollAckStore.class)
+ JdbcSsfPollAckStore jdbcSsfPollAckStore(JdbcOperations jdbc, ObjectProvider dataSource,
+ SsfReceiverProperties properties) {
+ String tablePrefix = properties.getJdbc().getTablePrefix();
+ String table = JdbcSsfSchema.pollAckTable(tablePrefix);
+ SsfJdbcOperations operations = new JdbcTemplateSsfJdbcOperations(jdbc);
+ JdbcSsfSchema.prepareTable(operations, table, JdbcSsfSchema.createPollAckTable(tablePrefix),
+ createTables(properties, dataSource), SCHEMA_HINT);
+ JdbcSsfPollAckStore store = new JdbcSsfPollAckStore(operations, tablePrefix);
+ store.setRetention(properties.getJdbc().getAckRetention());
+ store.setDeleteBatchSize(properties.getJdbc().getAckDeleteBatchSize());
+ logger.info("Pending poll acknowledgements are kept in the table " + table);
+ return store;
+ }
+
/**
* Purges the expired rows of the stores every
* {@code easyssf.receiver.jdbc.cleanup-interval}.
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/health/SsfReceiverHealthIndicator.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/health/SsfReceiverHealthIndicator.java
index 00179d6..5cc3e78 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/health/SsfReceiverHealthIndicator.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/health/SsfReceiverHealthIndicator.java
@@ -95,6 +95,7 @@ else if (state != SsfStreamRegistrar.State.REGISTERED) {
details.put("lastPoll", poller.getLastPollAt());
details.put("lastSuccessfulPoll", lastSuccess);
details.put("pollError", error);
+ details.put("pendingAcks", poller.getPendingAckCount());
Instant pausedUntil = poller.getPausedUntil();
if (pausedUntil != null && pausedUntil.isAfter(Instant.now())) {
details.put("pausedUntil", pausedUntil);
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperations.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperations.java
index 8f09381..b1e7e30 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperations.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperations.java
@@ -1,5 +1,6 @@
package org.easyssf.receiver.spring.boot.jdbc;
+import java.sql.PreparedStatement;
import java.util.List;
import org.easyssf.receiver.jdbc.SsfJdbcDuplicateKeyException;
@@ -9,6 +10,7 @@
import org.springframework.dao.DataAccessException;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.BadSqlGrammarException;
+import org.springframework.jdbc.core.ArgumentPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.util.Assert;
@@ -49,6 +51,21 @@ public List query(String sql, SsfJdbcRowMapper mapper, Object... args)
}
}
+ @Override
+ public List query(String sql, int maxRows, SsfJdbcRowMapper mapper, Object... args) {
+ try {
+ return this.jdbc.query((connection) -> {
+ PreparedStatement statement = connection.prepareStatement(sql);
+ statement.setMaxRows(Math.max(0, maxRows));
+ new ArgumentPreparedStatementSetter(args).setValues(statement);
+ return statement;
+ }, (row, index) -> mapper.map(row));
+ }
+ catch (DataAccessException ex) {
+ throw new SsfJdbcException("Could not run " + sql + ": " + ex.getMessage(), null, ex);
+ }
+ }
+
@Override
public void execute(String sql) {
try {
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java
index 9e8b4e0..8481e58 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java
@@ -4,15 +4,18 @@
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
+import org.easyssf.core.SsfDeliveryMethod;
import org.easyssf.core.event.SsfEventTypes;
import org.easyssf.receiver.http.JdkSsfHttpClient;
import org.easyssf.receiver.http.SsfHttpClient;
import org.easyssf.receiver.http.SsfHttpRequest;
import org.easyssf.receiver.jdbc.JdbcSsfJtiDedupStore;
+import org.easyssf.receiver.jdbc.JdbcSsfPollAckStore;
import org.easyssf.receiver.jdbc.JdbcSsfStoreCleanup;
import org.easyssf.receiver.jdbc.JdbcSsfTokenRevocationStore;
import org.easyssf.receiver.metrics.MicrometerSsfReceiverMetrics;
import org.easyssf.receiver.metrics.SsfReceiverMetrics;
+import org.easyssf.receiver.poll.SsfPollAckStore;
import org.easyssf.receiver.poll.SsfPoller;
import org.easyssf.receiver.revocation.InMemorySsfTokenRevocationStore;
import org.easyssf.receiver.revocation.SsfTokenRevocationEventHandler;
@@ -23,7 +26,10 @@
import org.easyssf.receiver.set.SsfJtiDedupStore;
import org.easyssf.receiver.set.SsfSetProcessor;
import org.easyssf.receiver.set.SsfSetVerifier;
+import org.easyssf.receiver.spring.boot.SsfReceiverProperties;
import org.easyssf.receiver.spring.boot.SsfTransmitterCustomizer;
+import org.easyssf.receiver.spring.boot.SsfTransmitterFactory;
+import org.easyssf.receiver.spring.boot.SsfTransmitterProperties;
import org.easyssf.receiver.spring.boot.http.RestClientSsfHttpClient;
import org.easyssf.receiver.spring.boot.resourceserver.SsfRevokedTokenValidator;
import org.easyssf.receiver.spring.boot.web.SsfPushEndpoint;
@@ -90,6 +96,43 @@ class SsfReceiverAutoConfigurationTests {
AutoConfigurations.of(DispatcherServletAutoConfiguration.class, WebMvcAutoConfiguration.class,
SecurityAutoConfiguration.class, ServletWebSecurityAutoConfiguration.class));
+ @Test
+ void longPollingIsConfiguredOnThePoller() {
+ this.webContextRunner
+ .withPropertyValues("easyssf.receiver.delivery-method=poll",
+ "easyssf.receiver.poll.endpoint-url=https://idp.example/realms/test/poll",
+ "easyssf.receiver.poll.long-polling=true", "easyssf.receiver.poll.long-polling-hold=20s")
+ .run((context) -> {
+ assertThat(context).hasNotFailed().hasSingleBean(SsfPoller.class);
+ assertThat(context.getBean(SsfPoller.class).isLongPolling()).isTrue();
+ });
+ this.webContextRunner
+ .withPropertyValues("easyssf.receiver.delivery-method=poll",
+ "easyssf.receiver.poll.endpoint-url=https://idp.example/realms/test/poll")
+ .run((context) -> assertThat(context.getBean(SsfPoller.class).isLongPolling()).isFalse());
+ }
+
+ @Test
+ void longPollingGetsAJdkClientUnlessTheApplicationClientIsOne() {
+ SsfTransmitterProperties properties = new SsfTransmitterProperties();
+ properties.setTransmitterIssuer("https://idp.example/realms/test");
+ properties.setDeliveryMethod(SsfDeliveryMethod.POLL);
+ SsfHttpClient restClient = (request) -> null;
+ JdkSsfHttpClient jdkClient = new JdkSsfHttpClient();
+ SsfReceiverProperties.Http http = new SsfReceiverProperties.Http();
+ http.setUserAgent("my-receiver/1.0");
+
+ // short polling: the client of the application, whatever it is
+ assertThat(SsfTransmitterFactory.pollHttpClient("default", properties, restClient, http)).isSameAs(restClient);
+ properties.getPoll().setLongPolling(true);
+ assertThat(SsfTransmitterFactory.pollHttpClient("default", properties, jdkClient, http)).isSameAs(jdkClient);
+ SsfHttpClient pollClient = SsfTransmitterFactory.pollHttpClient("default", properties, restClient, http);
+ assertThat(pollClient).isInstanceOf(JdkSsfHttpClient.class).isNotSameAs(restClient);
+ assertThat(((JdkSsfHttpClient) pollClient).getUserAgent()).isEqualTo("my-receiver/1.0");
+ properties.setDeliveryMethod(SsfDeliveryMethod.PUSH);
+ assertThat(SsfTransmitterFactory.pollHttpClient("default", properties, restClient, http)).isSameAs(restClient);
+ }
+
@Test
void namedTransmittersWorkWithoutTheDefaultOne() {
this.namedOnlyContextRunner
@@ -337,6 +380,7 @@ void stateIsKeptInMemoryWithoutJdbcTemplate() {
void stateIsKeptInTheDatabaseWhenApplicationHasJdbcTemplate() {
this.webContextRunner.withConfiguration(JDBC).run((context) -> {
assertThat(context).hasSingleBean(SsfJtiDedupStore.class).hasSingleBean(JdbcSsfJtiDedupStore.class);
+ assertThat(context).hasSingleBean(SsfPollAckStore.class).hasSingleBean(JdbcSsfPollAckStore.class);
assertThat(context).hasSingleBean(SsfTokenRevocationStore.class)
.hasSingleBean(JdbcSsfTokenRevocationStore.class);
// the tables are created in an embedded database
diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java
index b9872fb..f68ce63 100644
--- a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java
+++ b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java
@@ -42,4 +42,21 @@ void mapsDuplicateKeysAndMissingTables() {
assertThat(store.purgeExpired()).isZero();
}
+ @Test
+ void limitsAQueryThroughTheDriver() {
+ String prefix = JdbcSsfSchema.DEFAULT_TABLE_PREFIX;
+ JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.pollAckTable(prefix),
+ JdbcSsfSchema.createPollAckTable(prefix), true, null);
+ for (int i = 0; i < 3; i++) {
+ this.jdbc.update("INSERT INTO EASYSSF_POLL_ACK (ISSUER, JTI, RECORDED_AT) VALUES (?, ?, ?)", "i",
+ "jti-" + i, (long) i);
+ }
+ assertThat(this.jdbc.query("SELECT JTI FROM EASYSSF_POLL_ACK WHERE ISSUER = ? ORDER BY RECORDED_AT", 2,
+ (row) -> row.getString(1), "i"))
+ .containsExactly("jti-0", "jti-1");
+ assertThat(this.jdbc.query("SELECT JTI FROM EASYSSF_POLL_ACK WHERE ISSUER = ? ORDER BY RECORDED_AT",
+ (row) -> row.getString(1), "i"))
+ .hasSize(3);
+ }
+
}
diff --git a/easyssf-receiver-spring-boot-starter/README.md b/easyssf-receiver-spring-boot-starter/README.md
index 4f7a91d..69c8ea9 100644
--- a/easyssf-receiver-spring-boot-starter/README.md
+++ b/easyssf-receiver-spring-boot-starter/README.md
@@ -190,7 +190,24 @@ easyssf:
`easyssf.receiver.poll.rate-limit.max-backoff` (5 minutes). A `429` without the header pauses for
`easyssf.receiver.poll.rate-limit.fallback-backoff`, if set.
- With `easyssf.receiver.poll.auto-startup=false` nothing is polled until you call `SsfPoller.pollNow()`.
-- Long polling is not used, requests always ask the transmitter to return immediately.
+- **Acknowledgements** ride on the next poll request (RFC 8936, section 2.4) and wait in an
+ `SsfPollAckStore` until a request has carried them. In memory by default; with a `JdbcTemplate` they
+ are kept in the table `EASYSSF_POLL_ACK` (see [Keeping state in the database](#keeping-state-in-the-database)),
+ so that a SET handled right before a restart is acknowledged afterwards instead of being delivered
+ again and skipped as a duplicate. `stop()` sends the pending acknowledgements with a last request.
+ The health details show them as `pendingAcks`, the gauge `easyssf.receiver.poll.pending-acks` too.
+- **Long polling**: by default every request asks the transmitter to answer at once
+ (`returnImmediately: true`) and the transmitter is polled every `poll.interval`. With
+ `easyssf.receiver.poll.long-polling=true` the poller keeps one request outstanding instead: the
+ transmitter holds it until SETs are available or its hold time elapses (section 2.5), and the next
+ request goes out as soon as the response was handled. Set `poll.long-polling-hold` to the hold time
+ 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.
+ 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
+ `SsfTransmitterFactory.poller(...)`, uses whatever client you pass.
## Stream management
@@ -296,6 +313,7 @@ the state is kept in its database instead, without further configuration:
|---|---|---|
| `EASYSSF_PROCESSED_SET` | `JdbcSsfJtiDedupStore` | the SETs that were processed, forgotten after `easyssf.receiver.dedup.retention` (7 days) |
| `EASYSSF_REVOCATION` | `JdbcSsfTokenRevocationStore` | revoked sessions and subjects, per issuer, removed after `easyssf.receiver.resource-server.revocation-ttl` |
+| `EASYSSF_POLL_ACK` | `JdbcSsfPollAckStore` | acknowledgements and error reports a polling receiver owes its transmitter until a poll request carried them, forgotten after `easyssf.receiver.jdbc.ack-retention` (7 days) |
- **Tables**: in an embedded database (H2, HSQLDB, Derby) the tables are created on startup. For any
other database create them with your migration tool, the statements are in
@@ -303,21 +321,22 @@ the state is kept in its database instead, without further configuration:
(`classpath:org/easyssf/receiver/jdbc/schema.sql`), or set
`easyssf.receiver.jdbc.initialize-schema=always`. If a table is missing the application fails on
startup and says so, rather than on the first event. The stores use plain SQL (`VARCHAR`,
- `BIGINT`, no vendor syntax) and are tested on H2 and PostgreSQL.
+ `BIGINT`, no vendor syntax; a result limit is applied by the JDBC driver) and are tested on H2,
+ PostgreSQL and MySQL.
- **Opting out**: `easyssf.receiver.jdbc.enabled=false` keeps the state in memory although the
- application has a database. Your own `SsfJtiDedupStore` or `SsfTokenRevocationStore` bean takes
- precedence in any case.
+ application has a database. Your own `SsfJtiDedupStore`, `SsfTokenRevocationStore` or
+ `SsfPollAckStore` bean takes precedence in any case.
- **Cleanup**: expired rows are deleted when a store is written to (the SET table at most once a
minute) and every `easyssf.receiver.jdbc.cleanup-interval` (15 minutes) by `JdbcSsfStoreCleanup`
on a thread of its own; `easyssf.receiver.jdbc.cleanup-interval=0` turns the periodic cleanup off.
- Both stores offer `purgeExpired()` to do it from your own scheduler.
+ The stores offer `purgeExpired()` to do it from your own scheduler.
- **Cost**: a resource server checks every access token with one query on the primary key of
`EASYSSF_REVOCATION`.
- The stores live in `easyssf-receiver-jdbc` and have no Spring dependency: `SsfJdbcOperations` runs
their SQL, over a `DataSource` (`DataSourceSsfJdbcOperations`) or, in Spring Boot, the `JdbcTemplate`
of the application, so its transactions and exception translation apply.
- The stores use plain SQL (timestamps are stored as milliseconds since the epoch) and were tested
- with H2 and PostgreSQL. The revocation table is only used by resource servers. Sessions of an
+ with H2, PostgreSQL and MySQL. The revocation table is only used by resource servers. Sessions of an
OIDC client are not affected, see `SsfSessionTerminator`.
## Calling the transmitter
@@ -361,6 +380,7 @@ If the application has a Micrometer `MeterRegistry` (for example with
| `easyssf.receiver.sets` | counter | `transmitter` (the name of the transmitter, `default` for the one at `easyssf.receiver.*`), `delivery` (`push`, `poll`), `outcome` (`handled`, `duplicate`, `invalid`, `unauthenticated`, `unavailable`, `failed`) |
| `easyssf.receiver.events` | counter | `transmitter`, `delivery`, `event` (for example `CaepSessionRevoked`) |
| `easyssf.receiver.poll` | timer | `transmitter`, `outcome` (`success`, `failure`) |
+| `easyssf.receiver.poll.pending-acks` | gauge | `transmitter`; acknowledgements and error reports waiting for the next poll request, see [POLL delivery](#poll-delivery) |
Switch it off with `easyssf.receiver.metrics.enabled=false`, or provide your own `SsfReceiverMetrics` bean.
@@ -414,6 +434,8 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled
| `easyssf.receiver.poll.interval` | `30s` | |
| `easyssf.receiver.poll.initial-delay` | `1s` | |
| `easyssf.receiver.poll.max-events` | `100` | SETs per request. |
+| `easyssf.receiver.poll.long-polling` | `false` | Keep one request outstanding that the transmitter holds until SETs are available, instead of polling every `interval`. |
+| `easyssf.receiver.poll.long-polling-hold` | `30s` | How long the transmitter holds a long poll request, agreed with it; the request waits that long plus a margin. |
| `easyssf.receiver.poll.rate-limit.fallback-backoff` | | Pause after a `429` without `Retry-After`; unset, the next poll comes at the regular interval. |
| `easyssf.receiver.poll.rate-limit.max-backoff` | `5m` | Longest pause a `Retry-After` header or the fallback can cause. |
| `easyssf.receiver.metrics.enabled` | `true` | |
@@ -434,6 +456,8 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled
| `easyssf.receiver.jdbc.initialize-schema` | `embedded` | When to create missing tables: `embedded`, `always` or `never`. |
| `easyssf.receiver.jdbc.table-prefix` | `EASYSSF_` | Prefix of the table names. |
| `easyssf.receiver.jdbc.cleanup-interval` | `15m` | How often expired rows are purged; `0` turns it off. |
+| `easyssf.receiver.jdbc.ack-retention` | `7d` | How long the JDBC store keeps a poll acknowledgement no request managed to deliver. |
+| `easyssf.receiver.jdbc.ack-delete-batch-size` | `100` | How many delivered acknowledgements one `DELETE` of the JDBC store removes at once. |
| `easyssf.receiver.resource-server.enabled` | `true` | |
| `easyssf.receiver.resource-server.event-types` | `CaepSessionRevoked` | Events that revoke access tokens. |
| `easyssf.receiver.resource-server.revocation-ttl` | `1h` | |
diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java
index 4bae5fa..13f8e8d 100644
--- a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java
+++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java
@@ -64,7 +64,7 @@ public SsfHttpResponse execute(SsfHttpRequest request) throws IOException {
HttpRequest.BodyPublisher body = (request.body() != null) ? HttpRequest.BodyPublishers.ofString(request.body())
: HttpRequest.BodyPublishers.noBody();
HttpRequest.Builder builder = HttpRequest.newBuilder(request.uri())
- .timeout(this.readTimeout)
+ .timeout((request.timeout() != null) ? request.timeout() : this.readTimeout)
.method(request.method(), body);
request.headers().forEach(builder::header);
if (this.userAgent != null && request.headers().keySet().stream().noneMatch(USER_AGENT::equalsIgnoreCase)) {
diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpRequest.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpRequest.java
index a99237e..ef2fbd5 100644
--- a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpRequest.java
+++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpRequest.java
@@ -1,6 +1,7 @@
package org.easyssf.receiver.http;
import java.net.URI;
+import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -11,11 +12,17 @@
* @param uri the target of the request
* @param headers the request headers
* @param body the request body, {@code null} for none
+ * @param timeout how long to wait for the response, {@code null} for the timeout of the
+ * client; a long poll (RFC 8936) waits longer than other calls
*/
-public record SsfHttpRequest(String method, URI uri, Map headers, String body) {
+public record SsfHttpRequest(String method, URI uri, Map headers, String body, Duration timeout) {
+
+ public SsfHttpRequest(String method, URI uri, Map headers, String body) {
+ this(method, uri, headers, body, null);
+ }
public static SsfHttpRequest of(String method, URI uri) {
- return new SsfHttpRequest(method, uri, Map.of(), null);
+ return new SsfHttpRequest(method, uri, Map.of(), null, null);
}
public static SsfHttpRequest get(URI uri) {
@@ -25,7 +32,15 @@ public static SsfHttpRequest get(URI uri) {
public SsfHttpRequest withHeader(String name, String value) {
Map headers = new LinkedHashMap<>(this.headers);
headers.put(name, value);
- return new SsfHttpRequest(this.method, this.uri, headers, this.body);
+ return new SsfHttpRequest(this.method, this.uri, headers, this.body, this.timeout);
+ }
+
+ /**
+ * @param timeout how long to wait for the response of this request, instead of the
+ * timeout of the client
+ */
+ public SsfHttpRequest withTimeout(Duration timeout) {
+ return new SsfHttpRequest(this.method, this.uri, this.headers, this.body, timeout);
}
/**
@@ -37,7 +52,8 @@ public SsfHttpRequest withBearerToken(String accessToken) {
}
public SsfHttpRequest withBody(String contentType, String body) {
- return new SsfHttpRequest(this.method, this.uri, this.headers, body).withHeader("Content-Type", contentType);
+ return new SsfHttpRequest(this.method, this.uri, this.headers, body, this.timeout).withHeader("Content-Type",
+ contentType);
}
public SsfHttpRequest withJsonBody(String json) {
diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/MicrometerSsfReceiverMetrics.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/MicrometerSsfReceiverMetrics.java
index a17e16d..7bfcea9 100644
--- a/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/MicrometerSsfReceiverMetrics.java
+++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/MicrometerSsfReceiverMetrics.java
@@ -3,12 +3,14 @@
import java.time.Duration;
import java.util.Locale;
import java.util.function.Function;
+import java.util.function.IntSupplier;
import org.easyssf.core.SsfDeliveryMethod;
import org.easyssf.core.event.SsfEventTypes;
import org.easyssf.core.support.SsfAssert;
import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
@@ -73,6 +75,15 @@ public void pollCompleted(Duration duration, boolean success) {
pollCompleted(null, duration, success);
}
+ @Override
+ public void pollerStarted(String transmitter, IntSupplier pendingAcks) {
+ Gauge.builder("easyssf.receiver.poll.pending-acks", pendingAcks, IntSupplier::getAsInt)
+ .description(
+ "Acknowledgements and error reports a poller owes its transmitter, waiting for the next poll request")
+ .tag("transmitter", transmitter(transmitter))
+ .register(this.meterRegistry);
+ }
+
@Override
public void pollCompleted(String transmitter, Duration duration, boolean success) {
Timer.builder("easyssf.receiver.poll")
diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java
index dde2332..ed544ef 100644
--- a/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java
+++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java
@@ -1,6 +1,7 @@
package org.easyssf.receiver.metrics;
import java.time.Duration;
+import java.util.function.IntSupplier;
import org.easyssf.core.SsfDeliveryMethod;
@@ -101,4 +102,13 @@ default void pollCompleted(String transmitter, Duration duration, boolean succes
pollCompleted(duration, success);
}
+ /**
+ * A poller started and tells how to read the number of acknowledgements and error
+ * reports it still owes its transmitter, for a gauge.
+ * @param transmitter the issuer of the transmitter, {@code null} if unknown
+ * @param pendingAcks reads the current number
+ */
+ default void pollerStarted(String transmitter, IntSupplier pendingAcks) {
+ }
+
}
diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/InMemorySsfPollAckStore.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/InMemorySsfPollAckStore.java
new file mode 100644
index 0000000..818c478
--- /dev/null
+++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/InMemorySsfPollAckStore.java
@@ -0,0 +1,90 @@
+package org.easyssf.receiver.poll;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * {@link SsfPollAckStore} that keeps the pending acknowledgements in memory, in the order
+ * they were recorded. Lost when the application stops, the default of the poller.
+ */
+public class InMemorySsfPollAckStore implements SsfPollAckStore {
+
+ private final Map pending = new ConcurrentHashMap<>();
+
+ @Override
+ public void record(String issuer, SsfPendingAck ack) {
+ Acks acks = acks(issuer);
+ acks.lock.lock();
+ try {
+ acks.byJti.putIfAbsent(ack.jti(), ack);
+ }
+ finally {
+ acks.lock.unlock();
+ }
+ }
+
+ @Override
+ public List pending(String issuer, int limit) {
+ Acks acks = acks(issuer);
+ acks.lock.lock();
+ try {
+ List result = new ArrayList<>(Math.min(limit, acks.byJti.size()));
+ for (SsfPendingAck ack : acks.byJti.values()) {
+ if (result.size() >= limit) {
+ break;
+ }
+ result.add(ack);
+ }
+ return result;
+ }
+ finally {
+ acks.lock.unlock();
+ }
+ }
+
+ @Override
+ public void remove(String issuer, Collection jtis) {
+ Acks acks = acks(issuer);
+ acks.lock.lock();
+ try {
+ jtis.forEach(acks.byJti::remove);
+ }
+ finally {
+ acks.lock.unlock();
+ }
+ }
+
+ @Override
+ public int size(String issuer) {
+ Acks acks = acks(issuer);
+ acks.lock.lock();
+ try {
+ return acks.byJti.size();
+ }
+ finally {
+ acks.lock.unlock();
+ }
+ }
+
+ private Acks acks(String issuer) {
+ return this.pending.computeIfAbsent(String.valueOf(issuer), (key) -> new Acks());
+ }
+
+ /**
+ * The pending acknowledgements of one transmitter. A lock rather than
+ * {@code synchronized}, which pins virtual threads before JDK 24.
+ */
+ private static final class Acks {
+
+ private final ReentrantLock lock = new ReentrantLock();
+
+ private final Map byJti = new LinkedHashMap<>();
+
+ }
+
+}
diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPendingAck.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPendingAck.java
new file mode 100644
index 0000000..d12db32
--- /dev/null
+++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPendingAck.java
@@ -0,0 +1,44 @@
+package org.easyssf.receiver.poll;
+
+import org.easyssf.core.support.SsfAssert;
+
+/**
+ * What the poller owes the transmitter for a SET it fetched (RFC 8936, section 2.2): an
+ * acknowledgement that the SET was received and handled, or an error that it was
+ * rejected, which goes into the {@code setErrs} member of the next poll request.
+ *
+ * @param jti the identifier of the SET
+ * @param errorCode the error code of a rejected SET ({@code invalid_request}, ...),
+ * {@code null} for an acknowledgement
+ * @param errorDescription a description of the error, may be {@code null}
+ */
+public record SsfPendingAck(String jti, String errorCode, String errorDescription) {
+
+ public SsfPendingAck {
+ SsfAssert.notNull(jti, "jti must not be null");
+ SsfAssert.isTrue(errorCode != null || errorDescription == null, "an acknowledgement has no error description");
+ }
+
+ /**
+ * The acknowledgement of a handled SET.
+ */
+ public static SsfPendingAck ack(String jti) {
+ return new SsfPendingAck(jti, null, null);
+ }
+
+ /**
+ * The report of a rejected SET.
+ */
+ public static SsfPendingAck error(String jti, String errorCode, String errorDescription) {
+ SsfAssert.notNull(errorCode, "errorCode must not be null");
+ return new SsfPendingAck(jti, errorCode, errorDescription);
+ }
+
+ /**
+ * Whether this reports an error rather than acknowledging the SET.
+ */
+ public boolean isError() {
+ return this.errorCode != null;
+ }
+
+}
diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPollAckStore.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPollAckStore.java
new file mode 100644
index 0000000..e710654
--- /dev/null
+++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPollAckStore.java
@@ -0,0 +1,48 @@
+package org.easyssf.receiver.poll;
+
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Keeps the acknowledgements and error reports the poller owes a transmitter until a poll
+ * request has carried them (RFC 8936, section 2.4: they ride on the next request). The
+ * {@link InMemorySsfPollAckStore} loses them when the application stops, so a SET handled
+ * but not yet acknowledged is delivered again after a restart and skipped as a duplicate.
+ * A durable implementation, such as the JDBC store of {@code easyssf-receiver-jdbc},
+ * acknowledges it with the first poll after the restart instead.
+ *
+ *
+ * Entries are kept per transmitter, identified by its issuer, because a {@code jti} is
+ * only unique per issuer and a store may serve several transmitters. Implementations have
+ * to be safe for concurrent use.
+ */
+public interface SsfPollAckStore {
+
+ /**
+ * Records an acknowledgement or error report. Recording the same {@code jti} again
+ * does nothing.
+ * @param issuer the issuer of the transmitter the SET came from
+ * @param ack what to tell the transmitter
+ */
+ void record(String issuer, SsfPendingAck ack);
+
+ /**
+ * The acknowledgements and error reports waiting for the transmitter, oldest first.
+ * @param issuer the issuer of the transmitter
+ * @param limit the maximum number to return
+ */
+ List pending(String issuer, int limit);
+
+ /**
+ * Removes entries the transmitter has received.
+ * @param issuer the issuer of the transmitter
+ * @param jtis the identifiers of the SETs whose entries were delivered
+ */
+ void remove(String issuer, Collection jtis);
+
+ /**
+ * @return the number of entries waiting for the transmitter
+ */
+ int size(String issuer);
+
+}
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 11fd522..42f2f50 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
@@ -7,9 +7,7 @@
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;
@@ -35,7 +33,18 @@
*
* A SET is acknowledged once it was handled. An invalid SET is reported as an error to
* the transmitter. A SET that could not be handled is neither acknowledged nor reported,
- * the transmitter delivers it again with a later poll.
+ * the transmitter delivers it again with a later poll. Acknowledgements and error reports
+ * ride on the next poll request (section 2.4) and wait in the {@link SsfPollAckStore}
+ * until that request succeeded; with the in-memory store of the default they are lost
+ * when the application stops, with a durable store they survive a restart.
+ *
+ *
+ * By default the poller polls at a fixed {@link #setInterval(Duration) interval} and asks
+ * the transmitter to answer right away ({@code returnImmediately: true}, short polling).
+ * With {@link #setLongPolling(Duration) long polling} it keeps one request outstanding
+ * 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.
*/
public class SsfPoller {
@@ -45,6 +54,19 @@ public class SsfPoller {
private static final int MAX_REQUESTS_PER_POLL = 100;
+ /**
+ * How much longer than the hold time of the transmitter a long poll waits for the
+ * response.
+ */
+ private static final Duration LONG_POLL_MARGIN = Duration.ofSeconds(10);
+
+ /**
+ * A long poll that comes back empty faster than this was not held: the transmitter
+ * does not support long polling, and the poller waits the interval before the next
+ * request instead of asking again at once.
+ */
+ private static final Duration MIN_HOLD = Duration.ofSeconds(1);
+
private final SsfHttpClient httpClient;
private final SsfTransmitterTokenProvider tokenProvider;
@@ -53,9 +75,9 @@ public class SsfPoller {
private final SsfSetProcessor processor;
- private final List pendingAcks = new ArrayList<>();
+ private SsfPollAckStore ackStore = new InMemorySsfPollAckStore();
- private final Map pendingErrors = new LinkedHashMap<>();
+ private int maxAckBatch = 1000;
private SsfReceiverMetrics metrics = SsfReceiverMetrics.NOOP;
@@ -63,6 +85,12 @@ public class SsfPoller {
private Duration initialDelay = Duration.ZERO;
+ private Duration longPollingHold;
+
+ private boolean flushOnStop = true;
+
+ private ThreadFactory threadFactory = Thread.ofPlatform().name("ssf-poller").daemon().factory();
+
private int maxEvents = 100;
private String transmitter;
@@ -79,7 +107,9 @@ public class SsfPoller {
private volatile String lastPollError;
- private volatile ScheduledExecutorService scheduler;
+ private volatile Thread thread;
+
+ private volatile boolean running;
private final ReentrantLock polling = new ReentrantLock();
@@ -109,6 +139,28 @@ public void setMetrics(SsfReceiverMetrics metrics) {
this.metrics = metrics;
}
+ /**
+ * @param ackStore keeps the acknowledgements until a poll request carried them; an
+ * {@link InMemorySsfPollAckStore} by default
+ */
+ public void setAckStore(SsfPollAckStore ackStore) {
+ SsfAssert.notNull(ackStore, "ackStore must not be null");
+ this.ackStore = ackStore;
+ }
+
+ /**
+ * @param maxAckBatch the most acknowledgements and error reports one request carries,
+ * the rest goes with the next one; 1000 by default
+ */
+ public void setMaxAckBatch(int maxAckBatch) {
+ SsfAssert.isTrue(maxAckBatch > 0, "maxAckBatch must be positive");
+ this.maxAckBatch = maxAckBatch;
+ }
+
+ /**
+ * Sets the time between two polls. With long polling, the pause after a failed
+ * request or after a transmitter answered an empty long poll at once.
+ */
public void setInterval(Duration interval) {
SsfAssert.isTrue(interval != null && interval.isPositive(), "interval must be positive");
this.interval = interval;
@@ -120,10 +172,48 @@ public void setInitialDelay(Duration initialDelay) {
}
/**
- * Sets the maximum number of SETs fetched with one request.
+ * Switches to long polling (RFC 8936, section 2.5): the first request of a poll asks
+ * the transmitter to hold it until SETs are available
+ * ({@code returnImmediately: false}), and a running poller sends the next request as
+ * soon as a response was handled.
+ * @param hold how long the transmitter holds a request, part of the agreement with it
+ * (section 2.2); the request waits that long plus a margin for the response.
+ * {@code null} returns to short polling.
*/
+ public void setLongPolling(Duration hold) {
+ SsfAssert.isTrue(hold == null || hold.isPositive(), "hold must be positive");
+ this.longPollingHold = hold;
+ }
+
+ public boolean isLongPolling() {
+ return this.longPollingHold != null;
+ }
+
+ /**
+ * @param flushOnStop whether {@link #stop()} sends the pending acknowledgements with
+ * a last request, so that the transmitter does not deliver the SETs again after a
+ * restart; {@code true} by default
+ */
+ public void setFlushOnStop(boolean flushOnStop) {
+ this.flushOnStop = flushOnStop;
+ }
+
+ /**
+ * Sets what creates the thread {@link #start()} polls on, for a framework that
+ * manages its threads, names them per transmitter or prefers virtual threads
+ * ({@code Thread.ofVirtual().name("ssf-poller").factory()} works: the loop only
+ * blocks in the HTTP call and in sleeps). By default a platform daemon thread named
+ * {@code ssf-poller}. {@link #stop()} interrupts the thread, so the factory must not
+ * hand out threads that swallow interrupts.
+ */
+ public void setThreadFactory(ThreadFactory threadFactory) {
+ SsfAssert.notNull(threadFactory, "threadFactory must not be null");
+ this.threadFactory = threadFactory;
+ }
+
/**
* @param transmitter the issuer of the transmitter that is polled, for the metrics
+ * and as the key of its acknowledgements in the {@link SsfPollAckStore}
*/
public void setTransmitter(String transmitter) {
this.transmitter = transmitter;
@@ -149,26 +239,32 @@ public void setMaxPause(Duration maxPause) {
this.maxPause = maxPause;
}
+ /**
+ * Sets the maximum number of SETs fetched with one request.
+ */
public void setMaxEvents(int maxEvents) {
SsfAssert.isTrue(maxEvents > 0, "maxEvents must be positive");
this.maxEvents = maxEvents;
}
/**
- * Starts polling the transmitter periodically. Without it, SETs are only fetched by
- * Calling it again while running has no effect. calling {@link #pollNow()}.
+ * Starts polling the transmitter on a thread of its own (see
+ * {@link #setThreadFactory(ThreadFactory)}): periodically, or with one request always
+ * outstanding when long polling. Without it, SETs are only fetched by calling
+ * {@link #pollNow()}. Calling it again while running has no effect.
*/
public void start() {
this.lifecycle.lock();
try {
- if (this.scheduler != null) {
+ if (this.thread != null) {
return;
}
- ScheduledExecutorService scheduler = Executors
- .newSingleThreadScheduledExecutor(Thread.ofPlatform().name("ssf-poller").daemon().factory());
- scheduler.scheduleWithFixedDelay(this::pollQuietly, this.initialDelay.toMillis(), this.interval.toMillis(),
- TimeUnit.MILLISECONDS);
- this.scheduler = scheduler;
+ this.running = true;
+ this.metrics.pollerStarted(this.transmitter, this::getPendingAckCount);
+ Thread thread = this.threadFactory.newThread(this::run);
+ SsfAssert.notNull(thread, "the thread factory returned no thread");
+ this.thread = thread;
+ thread.start();
}
finally {
this.lifecycle.unlock();
@@ -176,15 +272,27 @@ public void start() {
}
/**
- * Stops polling the transmitter periodically.
+ * Stops polling: ends the polling thread, interrupting an outstanding long poll, and
+ * sends the pending acknowledgements with a last request unless
+ * {@link #setFlushOnStop(boolean)} says otherwise.
*/
public void stop() {
this.lifecycle.lock();
try {
- ScheduledExecutorService scheduler = this.scheduler;
- this.scheduler = null;
- if (scheduler != null) {
- scheduler.shutdownNow();
+ Thread thread = this.thread;
+ this.thread = null;
+ this.running = false;
+ if (thread != null) {
+ thread.interrupt();
+ try {
+ thread.join(Duration.ofSeconds(5));
+ }
+ catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ }
+ if (this.flushOnStop) {
+ flushAcks();
+ }
}
}
finally {
@@ -193,7 +301,7 @@ public void stop() {
}
public boolean isRunning() {
- return this.scheduler != null;
+ return this.thread != null;
}
/**
@@ -225,20 +333,69 @@ public Instant getPausedUntil() {
return this.pausedUntil;
}
- private void pollQuietly() {
+ /**
+ * @return the number of acknowledgements and error reports waiting for the next poll
+ * request
+ */
+ public int getPendingAckCount() {
+ String key = ackKey(this.endpoint.get());
+ return (key != null) ? this.ackStore.size(key) : 0;
+ }
+
+ private void run() {
try {
- pollNow();
+ Thread.sleep(this.initialDelay);
+ while (this.running) {
+ Duration pause = this.interval;
+ long started = System.nanoTime();
+ try {
+ PollResult result = this.polling.tryLock() ? pollLocked() : PollResult.SKIPPED;
+ if (result.skipped()) {
+ pause = pauseWhileSkipped();
+ }
+ else if (isLongPolling()) {
+ Duration elapsed = Duration.ofNanos(System.nanoTime() - started);
+ boolean held = result.fetched() > 0 || elapsed.compareTo(MIN_HOLD) >= 0;
+ pause = held ? Duration.ZERO : this.interval;
+ }
+ }
+ catch (RuntimeException ex) {
+ if (!this.running) {
+ return;
+ }
+ logger.warn("Could not poll the SSF transmitter: " + ex.getMessage());
+ logger.debug("Cause of the failed poll", ex);
+ }
+ if (this.running && pause.isPositive()) {
+ Thread.sleep(pause);
+ }
+ }
}
- catch (RuntimeException ex) {
- logger.warn("Could not poll the SSF transmitter: " + ex.getMessage());
- logger.debug("Cause of the failed poll", ex);
+ catch (InterruptedException ex) {
+ // stop() ends the thread
}
}
+ private PollResult pollLocked() {
+ try {
+ return poll();
+ }
+ finally {
+ this.polling.unlock();
+ }
+ }
+
+ private Duration pauseWhileSkipped() {
+ Instant pausedUntil = this.pausedUntil;
+ Duration untilResume = Duration.between(Instant.now(), pausedUntil);
+ return (untilResume.isPositive() && untilResume.compareTo(this.interval) > 0) ? untilResume : this.interval;
+ }
+
/**
* Fetches and processes the SETs that are available at the transmitter and
- * acknowledges them. One poll runs at a time: a call while another is in progress
- * returns right away.
+ * acknowledges them, with the request answered immediately whether or not long
+ * polling is on. One poll runs at a time: a call while another is in progress returns
+ * right away.
* @return the number of SETs fetched, {@code 0} as well if the poll endpoint is not
* known yet, the transmitter asked to slow down or a poll is in progress
*/
@@ -248,43 +405,51 @@ public int pollNow() {
return 0;
}
try {
- return poll();
+ return poll(false).fetched();
}
finally {
this.polling.unlock();
}
}
- private int poll() {
+ private PollResult poll() {
+ return poll(isLongPolling());
+ }
+
+ 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 0;
+ return PollResult.SKIPPED;
}
if (Instant.now().isBefore(this.pausedUntil)) {
logger.debug("Not polling, the SSF transmitter asked to wait until " + this.pausedUntil);
- return 0;
+ return PollResult.SKIPPED;
}
this.lastPollAt = Instant.now();
+ String key = ackKey(endpoint);
try {
int fetched = 0;
boolean moreAvailable = true;
for (int request = 0; moreAvailable && request < MAX_REQUESTS_PER_POLL; request++) {
- Map response = poll(endpoint, this.maxEvents);
+ // only the first request of a poll is held, a backlog is fetched right
+ // away
+ Map response = poll(endpoint, key, this.maxEvents, !(hold && request == 0));
Map sets = sets(response);
- sets.forEach(this::process);
+ sets.forEach((jti, set) -> process(key, jti, set));
fetched += sets.size();
moreAvailable = Boolean.TRUE.equals(response.get("moreAvailable")) && !sets.isEmpty();
}
- if (!this.pendingAcks.isEmpty() || !this.pendingErrors.isEmpty()) {
- // acknowledge right away instead of with the next poll
- Map sets = sets(poll(endpoint, 0));
- sets.forEach(this::process);
+ if (!hold && this.ackStore.size(key) > 0) {
+ // acknowledge right away instead of with the next poll; a long poll sends
+ // its next request at once anyway
+ Map sets = sets(poll(endpoint, key, 0, true));
+ sets.forEach((jti, set) -> process(key, jti, set));
fetched += sets.size();
}
this.lastSuccessfulPollAt = Instant.now();
this.lastPollError = null;
- return fetched;
+ return new PollResult(fetched, false);
}
catch (RuntimeException ex) {
this.lastPollError = ex.getMessage();
@@ -292,17 +457,14 @@ private int poll() {
}
}
- private void process(String jti, Object encodedSet) {
+ private void process(String key, String jti, Object encodedSet) {
try {
this.processor.process(String.valueOf(encodedSet), SsfDeliveryMethod.POLL);
- this.pendingAcks.add(jti);
+ this.ackStore.record(key, SsfPendingAck.ack(jti));
}
catch (SsfSetVerificationException ex) {
logger.warn("Rejecting polled SET " + jti + ": " + ex.getMessage());
- Map error = new LinkedHashMap<>();
- error.put("err", ex.getErrorCode());
- error.put("description", ex.getMessage());
- this.pendingErrors.put(jti, error);
+ this.ackStore.record(key, SsfPendingAck.error(jti, ex.getErrorCode(), ex.getMessage()));
}
catch (SsfTransmitterUnavailableException | SsfEventHandlingException ex) {
// not acknowledged, the transmitter delivers the SET again
@@ -310,12 +472,26 @@ private void process(String jti, Object encodedSet) {
}
}
- private Map poll(URI endpoint, int maxEvents) {
+ private Map poll(URI endpoint, String key, int maxEvents, boolean returnImmediately) {
+ List pending = this.ackStore.pending(key, this.maxAckBatch);
+ List acks = new ArrayList<>();
+ Map errors = new LinkedHashMap<>();
+ for (SsfPendingAck ack : pending) {
+ if (ack.isError()) {
+ Map error = new LinkedHashMap<>();
+ error.put("err", ack.errorCode());
+ if (ack.errorDescription() != null) {
+ error.put("description", ack.errorDescription());
+ }
+ errors.put(ack.jti(), error);
+ }
+ else {
+ acks.add(ack.jti());
+ }
+ }
Map request = new LinkedHashMap<>();
request.put("maxEvents", maxEvents);
- request.put("returnImmediately", true);
- List acks = List.copyOf(this.pendingAcks);
- Map errors = Map.copyOf(this.pendingErrors);
+ request.put("returnImmediately", returnImmediately);
if (!acks.isEmpty()) {
request.put("ack", acks);
}
@@ -326,11 +502,11 @@ private Map poll(URI endpoint, int maxEvents) {
SsfHttpResponse httpResponse;
Map response;
try {
- httpResponse = send(endpoint, request);
+ httpResponse = send(endpoint, request, returnImmediately);
if (httpResponse.status() == 401) {
// the access token expired or was revoked: try once more with a new one
this.tokenProvider.invalidate();
- httpResponse = send(endpoint, request);
+ httpResponse = send(endpoint, request, returnImmediately);
}
String body = httpResponse.body();
response = (httpResponse.isSuccessful() && body != null && !body.isBlank()) ? JSONObjectUtils.parse(body)
@@ -346,17 +522,53 @@ private Map poll(URI endpoint, int maxEvents) {
throw new IllegalStateException(
"Poll request to " + endpoint + " failed with status " + httpResponse.status());
}
- this.pendingAcks.removeAll(acks);
- errors.keySet().forEach(this.pendingErrors::remove);
+ if (!pending.isEmpty()) {
+ this.ackStore.remove(key, pending.stream().map(SsfPendingAck::jti).toList());
+ }
this.metrics.pollCompleted(this.transmitter, Duration.ofNanos(System.nanoTime() - started), true);
return response;
}
- private SsfHttpResponse send(URI endpoint, Map request) throws java.io.IOException {
- return this.httpClient.execute(SsfHttpRequest.of("POST", endpoint)
+ private SsfHttpResponse send(URI endpoint, Map request, boolean returnImmediately)
+ throws java.io.IOException {
+ SsfHttpRequest httpRequest = SsfHttpRequest.of("POST", endpoint)
.withHeader("Accept", "application/json")
.withBearerToken(this.tokenProvider.getAccessToken())
- .withJsonBody(JSONObjectUtils.toJSONString(request)));
+ .withJsonBody(JSONObjectUtils.toJSONString(request));
+ if (!returnImmediately && this.longPollingHold != null) {
+ httpRequest = httpRequest.withTimeout(this.longPollingHold.plus(LONG_POLL_MARGIN));
+ }
+ return this.httpClient.execute(httpRequest);
+ }
+
+ /**
+ * Sends the pending acknowledgements with a request that asks for no SETs, best
+ * effort.
+ */
+ private void flushAcks() {
+ URI endpoint = this.endpoint.get();
+ String key = ackKey(endpoint);
+ if (key == null || this.ackStore.size(key) == 0 || !this.polling.tryLock()) {
+ return;
+ }
+ try {
+ poll(endpoint, key, 0, true);
+ logger.debug("Acknowledged the pending SETs before stopping");
+ }
+ catch (RuntimeException ex) {
+ logger.debug("Could not acknowledge the pending SETs before stopping, they are sent with the next poll: "
+ + ex.getMessage());
+ }
+ finally {
+ this.polling.unlock();
+ }
+ }
+
+ private String ackKey(URI endpoint) {
+ if (this.transmitter != null) {
+ return this.transmitter;
+ }
+ return (endpoint != null) ? endpoint.toString() : null;
}
private void pauseIfAsked(SsfHttpResponse response) {
@@ -384,4 +596,10 @@ private static Map sets(Map response) {
return (response.get("sets") instanceof Map sets) ? (Map) sets : Map.of();
}
+ private record PollResult(int fetched, boolean skipped) {
+
+ static final PollResult SKIPPED = new PollResult(0, true);
+
+ }
+
}
diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStore.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStore.java
index b66288a..16d1780 100644
--- a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStore.java
+++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStore.java
@@ -2,6 +2,7 @@
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
import org.easyssf.core.event.SsfEventToken;
@@ -13,6 +14,9 @@ public class InMemorySsfJtiDedupStore implements SsfJtiDedupStore {
private final Map seen;
+ /** a lock rather than synchronized, which pins virtual threads before JDK 24 */
+ private final ReentrantLock lock = new ReentrantLock();
+
public InMemorySsfJtiDedupStore(int capacity) {
int maxEntries = Math.max(1, capacity);
this.seen = new LinkedHashMap<>() {
@@ -24,17 +28,35 @@ protected boolean removeEldestEntry(Map.Entry eldest) {
}
@Override
- public synchronized boolean seenBefore(SsfEventToken eventToken) {
- return this.seen.put(SsfJtiDedupStore.key(eventToken), Boolean.TRUE) != null;
+ public boolean seenBefore(SsfEventToken eventToken) {
+ this.lock.lock();
+ try {
+ return this.seen.put(SsfJtiDedupStore.key(eventToken), Boolean.TRUE) != null;
+ }
+ finally {
+ this.lock.unlock();
+ }
}
@Override
- public synchronized void forget(SsfEventToken eventToken) {
- this.seen.remove(SsfJtiDedupStore.key(eventToken));
+ public void forget(SsfEventToken eventToken) {
+ this.lock.lock();
+ try {
+ this.seen.remove(SsfJtiDedupStore.key(eventToken));
+ }
+ finally {
+ this.lock.unlock();
+ }
}
- public synchronized int size() {
- return this.seen.size();
+ public int size() {
+ this.lock.lock();
+ try {
+ return this.seen.size();
+ }
+ finally {
+ this.lock.unlock();
+ }
}
}
diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/InMemorySsfPollAckStoreTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/InMemorySsfPollAckStoreTests.java
new file mode 100644
index 0000000..5a4f095
--- /dev/null
+++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/InMemorySsfPollAckStoreTests.java
@@ -0,0 +1,57 @@
+package org.easyssf.receiver.poll;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+
+class InMemorySsfPollAckStoreTests {
+
+ private static final String ISSUER = "https://idp.example";
+
+ private final InMemorySsfPollAckStore store = new InMemorySsfPollAckStore();
+
+ @Test
+ void keepsAcknowledgementsAndErrorsInOrderPerIssuer() {
+ this.store.record(ISSUER, SsfPendingAck.ack("jti-1"));
+ this.store.record(ISSUER, SsfPendingAck.error("jti-2", "invalid_request", "not a SET"));
+ this.store.record(ISSUER, SsfPendingAck.ack("jti-3"));
+ this.store.record("https://other.example", SsfPendingAck.ack("jti-1"));
+
+ assertThat(this.store.size(ISSUER)).isEqualTo(3);
+ assertThat(this.store.pending(ISSUER, 10)).extracting(SsfPendingAck::jti)
+ .containsExactly("jti-1", "jti-2", "jti-3");
+ assertThat(this.store.pending(ISSUER, 2)).extracting(SsfPendingAck::jti).containsExactly("jti-1", "jti-2");
+ assertThat(this.store.pending(ISSUER, 10).get(1).isError()).isTrue();
+ assertThat(this.store.pending(ISSUER, 10).get(1).errorDescription()).isEqualTo("not a SET");
+ assertThat(this.store.size("https://other.example")).isEqualTo(1);
+ assertThat(this.store.size("https://unknown.example")).isZero();
+ }
+
+ @Test
+ void recordingTheSameSetAgainChangesNothing() {
+ this.store.record(ISSUER, SsfPendingAck.ack("jti-1"));
+ this.store.record(ISSUER, SsfPendingAck.error("jti-1", "invalid_request", null));
+ assertThat(this.store.pending(ISSUER, 10)).singleElement()
+ .satisfies((ack) -> assertThat(ack.isError()).isFalse());
+ }
+
+ @Test
+ void removesWhatTheTransmitterReceived() {
+ this.store.record(ISSUER, SsfPendingAck.ack("jti-1"));
+ this.store.record(ISSUER, SsfPendingAck.ack("jti-2"));
+ this.store.remove(ISSUER, List.of("jti-1", "jti-unknown"));
+ assertThat(this.store.pending(ISSUER, 10)).extracting(SsfPendingAck::jti).containsExactly("jti-2");
+ }
+
+ @Test
+ void pendingAckIsValidated() {
+ assertThatIllegalArgumentException().isThrownBy(() -> SsfPendingAck.error("jti", null, "x"));
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> new SsfPendingAck("jti", null, "description without code"));
+ assertThat(SsfPendingAck.ack("jti").isError()).isFalse();
+ }
+
+}
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 36dbe6c..ff857e4 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
@@ -34,6 +34,10 @@ class SsfPollerTests {
private final List failures = new ArrayList<>();
+ /** runs after a SET was handled, before it is acknowledged */
+ private Runnable afterHandling = () -> {
+ };
+
private final SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry();
private SsfPoller poller;
@@ -41,6 +45,7 @@ class SsfPollerTests {
@BeforeEach
void setUp() {
transmitter.reset();
+ transmitter.setAvailable(true);
this.poller = poller(() -> URI.create(transmitter.pollUri()), TestTransmitter.ACCESS_TOKEN);
}
@@ -59,6 +64,7 @@ private SsfPoller poller(java.util.function.Supplier endpoint, SsfTransmitt
throw this.failures.remove(0);
}
this.handled.add(eventContext.eventToken().jti());
+ this.afterHandling.run();
};
SsfHttpClient http = new JdkSsfHttpClient();
NimbusSsfSetVerifier verifier = new NimbusSsfSetVerifier(transmitter.issuer(), transmitter::jwksUri, http);
@@ -84,14 +90,43 @@ void startIsIdempotentAndStopEndsThePollingThread() {
}
private static List pollerThreads() {
+ return pollerThreads("ssf-poller");
+ }
+
+ private static List pollerThreads(String name) {
return Thread.getAllStackTraces()
.keySet()
.stream()
- .filter((thread) -> thread.getName().equals("ssf-poller"))
+ .filter((thread) -> thread.getName().equals(name))
.filter(Thread::isAlive)
.toList();
}
+ @Test
+ void pollsOnTheThreadOfTheGivenFactory() {
+ this.poller.setThreadFactory(Thread.ofPlatform().name("ssf-poller-idp").daemon().factory());
+ this.poller.setInitialDelay(Duration.ofHours(1));
+ this.poller.start();
+ assertThat(pollerThreads("ssf-poller-idp")).hasSize(1);
+ assertThat(pollerThreads()).isEmpty();
+ this.poller.stop();
+ await().atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> assertThat(pollerThreads("ssf-poller-idp")).isEmpty());
+ }
+
+ @Test
+ void pollsOnAVirtualThreadIfTheFactoryMakesOne() {
+ transmitter.setLongPollHold(Duration.ofSeconds(5));
+ this.poller.setLongPolling(Duration.ofSeconds(5));
+ this.poller.setThreadFactory(Thread.ofVirtual().name("ssf-poller-virtual").factory());
+ this.poller.start();
+ await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() >= 1);
+ String jti = queue("session-1");
+ await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> assertThat(this.handled).containsExactly(jti));
+ this.poller.stop();
+ assertThat(this.poller.isRunning()).isFalse();
+ }
+
@Test
void handlesAndAcknowledgesAvailableSets() {
String first = queue("session-1");
@@ -179,6 +214,140 @@ void reportsFailedPollRequest() {
.isEqualTo(1);
}
+ @Test
+ void acknowledgementsWaitInTheStoreUntilARequestCarriedThem() {
+ InMemorySsfPollAckStore store = new InMemorySsfPollAckStore();
+ this.poller.setAckStore(store);
+ this.poller.setTransmitter(transmitter.issuer());
+ String jti = queue("session-1");
+ transmitter.queueSet("broken", "not-a-set");
+ // the transmitter goes away between handling the SETs and the request that
+ // acknowledges them
+ this.afterHandling = () -> transmitter.setAvailable(false);
+
+ assertThatIllegalStateException().isThrownBy(this.poller::pollNow).withMessageContaining("503");
+
+ assertThat(this.handled).containsExactly(jti);
+ assertThat(transmitter.acknowledgedSets()).isEmpty();
+ assertThat(store.pending(transmitter.issuer(), 10)).extracting(SsfPendingAck::jti)
+ .containsExactly(jti, "broken");
+ assertThat(this.poller.getPendingAckCount()).isEqualTo(2);
+
+ transmitter.setAvailable(true);
+ this.afterHandling = () -> {
+ };
+ this.poller.pollNow();
+ assertThat(transmitter.acknowledgedSets()).containsExactly(jti);
+ assertThat(transmitter.reportedErrors()).containsOnlyKeys("broken");
+ assertThat(store.size(transmitter.issuer())).isZero();
+ }
+
+ @Test
+ void acknowledgementsAreSentInBatches() {
+ this.poller.setMaxAckBatch(2);
+ List queued = List.of(queue("session-1"), queue("session-2"), queue("session-3"));
+
+ assertThat(this.poller.pollNow()).isEqualTo(3);
+ // the request after handling carried two, the third waits for the next poll
+ assertThat(transmitter.acknowledgedSets()).containsExactly(queued.get(0), queued.get(1));
+ assertThat(this.poller.getPendingAckCount()).isEqualTo(1);
+
+ this.poller.pollNow();
+ assertThat(transmitter.acknowledgedSets()).containsExactlyElementsOf(queued);
+ assertThat(this.poller.getPendingAckCount()).isZero();
+ }
+
+ @Test
+ void pendingAcknowledgementsAreAGauge() {
+ this.poller.setTransmitter(transmitter.issuer());
+ this.poller.setInitialDelay(Duration.ofHours(1));
+ this.poller.start();
+ this.poller.stop();
+ assertThat(this.meterRegistry.get("easyssf.receiver.poll.pending-acks")
+ .tag("transmitter", transmitter.issuer())
+ .gauge()
+ .value()).isZero();
+ String jti = queue("session-1");
+ this.afterHandling = () -> transmitter.setAvailable(false);
+ assertThatIllegalStateException().isThrownBy(this.poller::pollNow);
+ assertThat(this.handled).containsExactly(jti);
+ assertThat(this.meterRegistry.get("easyssf.receiver.poll.pending-acks").gauge().value()).isEqualTo(1);
+ }
+
+ @Test
+ void stopFlushesThePendingAcknowledgements() {
+ this.poller.setInitialDelay(Duration.ofHours(1));
+ String jti = queue("session-1");
+ this.afterHandling = () -> transmitter.setAvailable(false);
+ assertThatIllegalStateException().isThrownBy(this.poller::pollNow);
+ assertThat(this.poller.getPendingAckCount()).isEqualTo(1);
+ transmitter.setAvailable(true);
+
+ this.poller.start();
+ this.poller.stop();
+
+ assertThat(transmitter.acknowledgedSets()).containsExactly(jti);
+ assertThat(this.poller.getPendingAckCount()).isZero();
+ }
+
+ @Test
+ void longPollIsHeldUntilASetArrives() throws Exception {
+ transmitter.setLongPollHold(Duration.ofSeconds(5));
+ this.poller.setLongPolling(Duration.ofSeconds(5));
+ this.poller.setInterval(Duration.ofSeconds(30));
+ this.poller.start();
+ await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() >= 1);
+ assertThat(transmitter.lastPollRequest()).containsEntry("returnImmediately", false);
+ assertThat(this.handled).isEmpty();
+
+ long queuedAt = System.nanoTime();
+ String jti = queue("session-1");
+ await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> assertThat(this.handled).containsExactly(jti));
+ // handled while the request was held, long before the hold time elapsed
+ assertThat(Duration.ofNanos(System.nanoTime() - queuedAt)).isLessThan(Duration.ofSeconds(3));
+ // the next request went out right away and carries the acknowledgement
+ await().atMost(Duration.ofSeconds(3))
+ .untilAsserted(() -> assertThat(transmitter.acknowledgedSets()).containsExactly(jti));
+ assertThat(transmitter.lastPollRequest()).containsEntry("returnImmediately", false);
+ this.poller.stop();
+ }
+
+ @Test
+ void longPollAgainstATransmitterThatAnswersAtOnceWaitsTheInterval() {
+ // the test transmitter does not hold by default
+ this.poller.setLongPolling(Duration.ofSeconds(5));
+ this.poller.setInterval(Duration.ofMinutes(5));
+ this.poller.start();
+ await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() >= 1);
+ await().during(Duration.ofSeconds(1))
+ .atMost(Duration.ofSeconds(2))
+ .untilAsserted(() -> assertThat(transmitter.pollRequests()).isEqualTo(1));
+ this.poller.stop();
+ }
+
+ @Test
+ void stopInterruptsAHeldLongPoll() {
+ transmitter.setLongPollHold(Duration.ofSeconds(30));
+ this.poller.setLongPolling(Duration.ofSeconds(30));
+ this.poller.start();
+ await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() >= 1);
+ long stoppedAt = System.nanoTime();
+ this.poller.stop();
+ assertThat(Duration.ofNanos(System.nanoTime() - stoppedAt)).isLessThan(Duration.ofSeconds(5));
+ assertThat(this.poller.isRunning()).isFalse();
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> assertThat(pollerThreads()).isEmpty());
+ }
+
+ @Test
+ void pollNowAsksForAnImmediateAnswerEvenWhenLongPolling() {
+ transmitter.setLongPollHold(Duration.ofSeconds(30));
+ this.poller.setLongPolling(Duration.ofSeconds(30));
+ long started = System.nanoTime();
+ assertThat(this.poller.pollNow()).isZero();
+ assertThat(Duration.ofNanos(System.nanoTime() - started)).isLessThan(Duration.ofSeconds(5));
+ assertThat(transmitter.lastPollRequest()).containsEntry("returnImmediately", true);
+ }
+
private String queue(String sessionId) {
var claims = transmitter.setClaims("CaepSessionRevoked", opaque(sessionId)).build();
transmitter.queueSet(transmitter.signSet(claims));
diff --git a/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java b/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java
index 4f56c18..a095049 100644
--- a/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java
+++ b/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java
@@ -4,9 +4,9 @@
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.time.Instant;
import java.util.Base64;
-import java.util.Collections;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
@@ -14,7 +14,10 @@
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
import org.easyssf.core.event.SsfEventTypes;
@@ -52,7 +55,15 @@ public final class TestTransmitter implements AutoCloseable {
private final List> streams = new CopyOnWriteArrayList<>();
- private final Map queuedSets = Collections.synchronizedMap(new LinkedHashMap<>());
+ /**
+ * guarded by queueLock; a lock rather than synchronized, which pins virtual threads
+ * before JDK 24
+ */
+ private final Map queuedSets = new LinkedHashMap<>();
+
+ private final ReentrantLock queueLock = new ReentrantLock();
+
+ private final Condition setQueued = this.queueLock.newCondition();
private final List acknowledgedSets = new CopyOnWriteArrayList<>();
@@ -66,6 +77,12 @@ public final class TestTransmitter implements AutoCloseable {
private final AtomicInteger tokenRequests = new AtomicInteger();
+ private final AtomicInteger pollRequests = new AtomicInteger();
+
+ private volatile Map lastPollRequest = Map.of();
+
+ private volatile Duration longPollHold = Duration.ZERO;
+
private volatile boolean available = true;
private volatile String validAccessToken = ACCESS_TOKEN;
@@ -88,6 +105,10 @@ public TestTransmitter(MetadataLocation metadataLocation) {
try {
this.key = generateKey(2048, "test-key");
this.server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+ // one thread per request, so that a held poll request does not block the
+ // others
+ this.server.setExecutor(
+ Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("test-transmitter-", 0).factory()));
this.server.createContext("/", (exchange) -> {
try {
handle(exchange);
@@ -212,29 +233,44 @@ private void handleStreams(HttpExchange exchange, String method, String streamId
}
@SuppressWarnings("unchecked")
- private Map poll(Map request) {
- if (request.get("ack") instanceof List acks) {
- acks.forEach((jti) -> {
- if (this.queuedSets.remove(jti) != null) {
- this.acknowledgedSets.add((String) jti);
+ private Map poll(Map request) throws InterruptedException {
+ this.pollRequests.incrementAndGet();
+ this.lastPollRequest = Map.copyOf(request);
+ this.queueLock.lock();
+ try {
+ if (request.get("ack") instanceof List acks) {
+ acks.forEach((jti) -> {
+ if (this.queuedSets.remove(jti) != null) {
+ this.acknowledgedSets.add((String) jti);
+ }
+ });
+ }
+ if (request.get("setErrs") instanceof Map errors) {
+ ((Map) errors).forEach((jti, error) -> {
+ this.queuedSets.remove(jti);
+ this.reportedErrors.put(jti, error);
+ });
+ }
+ int maxEvents = ((Number) request.getOrDefault("maxEvents", 10)).intValue();
+ // a long poll (returnImmediately false or absent, RFC 8936 section 2.5) is
+ // held until a
+ // SET is queued or the hold time elapses
+ if (!Boolean.TRUE.equals(request.get("returnImmediately")) && maxEvents > 0) {
+ long remaining = this.longPollHold.toNanos();
+ while (this.queuedSets.isEmpty() && remaining > 0) {
+ remaining = this.setQueued.awaitNanos(remaining);
}
- });
- }
- if (request.get("setErrs") instanceof Map errors) {
- ((Map) errors).forEach((jti, error) -> {
- this.queuedSets.remove(jti);
- this.reportedErrors.put(jti, error);
- });
- }
- int maxEvents = ((Number) request.getOrDefault("maxEvents", 10)).intValue();
- Map sets = new LinkedHashMap<>();
- synchronized (this.queuedSets) {
+ }
+ Map sets = new LinkedHashMap<>();
this.queuedSets.entrySet()
.stream()
.limit(maxEvents)
.forEach((set) -> sets.put(set.getKey(), set.getValue()));
+ return Map.of("sets", sets, "moreAvailable", this.queuedSets.size() > sets.size());
+ }
+ finally {
+ this.queueLock.unlock();
}
- return Map.of("sets", sets, "moreAvailable", this.queuedSets.size() > sets.size());
}
@SuppressWarnings("unchecked")
@@ -330,7 +366,7 @@ public List> streams() {
*/
public void queueSet(String set) {
try {
- this.queuedSets.put(SignedJWT.parse(set).getJWTClaimsSet().getJWTID(), set);
+ queueSet(SignedJWT.parse(set).getJWTClaimsSet().getJWTID(), set);
}
catch (java.text.ParseException ex) {
throw new IllegalArgumentException(ex);
@@ -338,7 +374,51 @@ public void queueSet(String set) {
}
public void queueSet(String jti, String set) {
- this.queuedSets.put(jti, set);
+ this.queueLock.lock();
+ try {
+ this.queuedSets.put(jti, set);
+ // wakes up a held poll request
+ this.setQueued.signalAll();
+ }
+ finally {
+ this.queueLock.unlock();
+ }
+ }
+
+ /**
+ * How long a poll request without {@code returnImmediately: true} is held while no
+ * SET is queued. Zero by default: the transmitter then answers at once, as if it did
+ * not support long polling.
+ */
+ public void setLongPollHold(Duration longPollHold) {
+ this.longPollHold = (longPollHold != null) ? longPollHold : Duration.ZERO;
+ }
+
+ /**
+ * @return the number of SETs queued and not yet acknowledged
+ */
+ public int queuedSetCount() {
+ this.queueLock.lock();
+ try {
+ return this.queuedSets.size();
+ }
+ finally {
+ this.queueLock.unlock();
+ }
+ }
+
+ /**
+ * @return the number of poll requests received
+ */
+ public int pollRequests() {
+ return this.pollRequests.get();
+ }
+
+ /**
+ * @return the body of the last poll request, empty if none was received
+ */
+ public Map lastPollRequest() {
+ return this.lastPollRequest;
}
public List acknowledgedSets() {
@@ -366,7 +446,17 @@ public List> removedSubjects() {
*/
public void reset() {
this.streams.clear();
- this.queuedSets.clear();
+ this.queueLock.lock();
+ try {
+ this.queuedSets.clear();
+ this.setQueued.signalAll();
+ }
+ finally {
+ this.queueLock.unlock();
+ }
+ this.pollRequests.set(0);
+ this.lastPollRequest = Map.of();
+ this.longPollHold = Duration.ZERO;
this.acknowledgedSets.clear();
this.reportedErrors.clear();
this.verificationRequests.clear();