listenIndefinitely
Listens the changes of the collection indefinitely, in a (daemon) thread: a change stream that survives the errors and the invalidations.
The official driver already resumes the change stream once after a resumable error. This listener 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
the exceptions thrown by the listener are sent to onError, the listening continues
a change that can not be decoded is skipped: the exception is sent to onError, the listening continues
Sample:
val listening = col.listenIndefinitely { change -> reload(change.fullDocument) }
// ...
listening.close()Return
an AutoCloseable that stops the listening
Parameters
the aggregation pipeline used to filter the changes
the full document option (FullDocument.UPDATE_LOOKUP to get the document of the updates)
the delay before reopening the change stream after an error
the maximum delay before reopening the change stream after consecutive errors
called for each error (for logging purposes)
configures the change stream (batch size, collation, fullDocumentBeforeChange...)
to listen changes