Struct mongodb::change_stream::session::SessionChangeStream

pub struct SessionChangeStream<T>
where
    T: DeserializeOwned + Unpin,
{ /* private fields */ }

A SessionChangeStream is a change stream that was created with a ClientSession that must be iterated using one. To iterate, use SessionChangeStream::next:

let mut cs = coll.watch().session(&mut session).await?;
while let Some(event) = cs.next(&mut session).await? {
    println!("{:?}", event)
}

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

Implementations

impl<T> SessionChangeStream<T>

where T: DeserializeOwned + Unpin + Send + Sync,

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 + Unpin + Send + Sync>( self, ) -> SessionChangeStream<D>

Update the type streamed values will be parsed as.

pub async fn next(&mut self, session: &mut ClientSession) -> Result<Option<T>>

Retrieve the next result from the change stream. The session provided must be the same session used to create the change stream.

let mut cs = coll.watch().session(&mut session).await?;
while let Some(event) = cs.next(&mut session).await? {
    let id = bson::to_bson(&event.id)?;
    other_coll.insert_one(doc! { "id": id }).session(&mut session).await?;
}

pub fn is_alive(&self) -> bool

Returns whether the change stream will continue to receive events.

pub async fn next_if_any( &mut self, session: &mut ClientSession, ) -> Result<Option<T>>

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

Where calling 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().session(&mut session).await?;
let mut resume_token = None;
while change_stream.is_alive() {
    if let Some(event) = change_stream.next_if_any(&mut session).await? {
        // process event
    }
    resume_token = change_stream.resume_token();
}