Struct ChangeStream

pub struct ChangeStream<T>
where
    T: DeserializeOwned,
{ /* private fields */ }

A ChangeStream streams the ongoing changes of its associated collection, database or deployment. ChangeStream instances should be created with method watch against the relevant target.

ChangeStreams are “resumable”, meaning that they can be restarted at a given place in the stream of events. This is done automatically when the ChangeStream encounters certain “resumable” errors, such as transient network failures. It can also be done manually by passing a ResumeToken retrieved from a past event into either the resume_after or start_after (4.2+) options used to create the ChangeStream. Issuing a raw change stream aggregation is discouraged unless users wish to explicitly opt out of resumability.

A ChangeStream can be iterated like any other Stream:

let mut change_stream = coll.watch().await?;
let coll_ref = coll.clone();
task::spawn(async move {
    coll_ref.insert_one(doc! { "x": 1 }).await;
});
while let Some(event) = change_stream.next().await.transpose()? {
    println!("operation performed: {:?}, document: {:?}", event.operation_type, event.full_document);
    // operation performed: Insert, document: Some(Document({"x": Int32(1)}))
}

If a ChangeStream is still open when it goes out of scope, it will automatically be closed via an asynchronous killCursors command executed from its Drop implementation.

See the documentation here for more details. Also see the documentation on usage recommendations.

Implementations

impl<T> ChangeStream<T>

where T: DeserializeOwned,

pub fn resume_token(&self) -> Option<ResumeToken>

Returns the cached resume token that can be used to resume after the most recently returned change.

See the documentation here for more information on change stream resume tokens.

pub fn with_type<D: DeserializeOwned>(self) -> ChangeStream<D>

Update the type streamed values will be parsed as.

pub fn is_alive(&self) -> bool

Returns whether the change stream will continue to receive events.

pub async fn next_if_any(&mut self) -> Result<Option<T>>

Retrieves the next result from the change stream, if any.

Where calling Stream::next will internally loop until a change document is received, this will make at most one request and return None if the returned document batch is empty. This method should be used when storing the resume token in order to ensure the most up to date token is received, e.g.

let mut change_stream = coll.watch().await?;
let mut resume_token = None;
while change_stream.is_alive() {
    if let Some(event) = change_stream.next_if_any().await? {
        // process event
    }
    resume_token = change_stream.resume_token();
}

Trait Implementations

impl<T> Debug for ChangeStream<T>

where T: DeserializeOwned,

fn fmt(&self, __f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> Stream for ChangeStream<T>

where T: DeserializeOwned,

type Item = Result<T, Error>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.