mongo/me.ahoo.wow.mongo.subscription/MongoCheckpointStore

MongoCheckpointStore

class MongoCheckpointStore(client: MongoClient, databaseName: String, dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher, ddl: Ddl = Ddl.AUTO, operationTimeout: Duration = MongoTimeouts.OPERATION) : CheckpointStore

The CheckpointStore on MongoDB: one document per (group, subscription, partition) in wow_checkpoint, keyed by a unique index.

A save is one conditional upsert whose filter admits only a lower stored position, so concurrent saves leave the highest and a lower one changes nothing (see conditionalUpsert). Calls run on dispatcher, virtual threads by default.

Constructors

constructor(client: MongoClient, databaseName: String, dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher, ddl: Ddl = Ddl.AUTO, operationTimeout: Duration = MongoTimeouts.OPERATION)

Functions

createSchema

fun createSchema()

Creates the key index when missing.

list

open suspend override fun list(group: String): List<Checkpoint>

load

open suspend override fun load(group: String, subscription: String, partition: Int): Long?

save

open suspend override fun save(group: String, subscription: String, partition: Int, position: Long)

saveAll

open suspend override fun saveAll(group: String, checkpoints: List<Checkpoint>)

One unordered bulk write of the conditional upserts save makes one by one: one round trip. An upsert the key index refuses (the guard failed on a stored checkpoint, or a concurrent insert won) is written again without upsert, as conditionalUpsert does.