kmongo-dsl-coroutine/org.litote.kmongo.coroutine/watchIndefinitely

watchIndefinitely

fun <T : Any> MongoCollection<T>.watchIndefinitely(pipeline: List<Bson> = emptyList(), fullDocument: FullDocument = FullDocument.DEFAULT, retryDelay: Duration = 1.seconds, maxRetryDelay: Duration = 1.minutes, onError: suspend (Throwable) -> Unit = {}, configure: ChangeStreamFlow<T>.() -> ChangeStreamFlow<T> = { this }): Flow<ChangeStreamDocument<T>>

Watches the collection indefinitely: a change stream that survives the errors and the invalidations.

The official driver already resumes the change stream once after a resumable error. This flow goes further:

after any MongoException (server down, election, network...) the change stream is reopened after retryDelay (doubled after each consecutive error, up to maxRetryDelay), starting after the last received change (the resume token is kept) - no change is lost if the oplog still contains it
if the resume token is not usable any more (history lost), the change stream restarts from the current time
if the change stream is invalidated (the collection is dropped or renamed), it is reopened
a change that can not be decoded is skipped: the exception is sent to onError, the flow continues

The exceptions thrown by the collector are not caught (the flow is cancelled). Cancel the collecting coroutine to stop watching.

Sample:

scope.launch { col.watchIndefinitely().collect { change -> reload(change.fullDocument) } }

Parameters

pipeline

the aggregation pipeline used to filter the changes

fullDocument

the full document option (FullDocument.UPDATE_LOOKUP to get the document of the updates)

retryDelay

the delay before reopening the change stream after an error

maxRetryDelay

the maximum delay before reopening the change stream after consecutive errors

onError

called for each error, before the retry (for logging purposes)

configure

configures the change stream (batch size, collation, fullDocumentBeforeChange...)