Open Change Streams

Overview

In this guide, you can learn how to use a change stream to monitor real-time changes to your data. A change stream is a MongoDB Server feature that allows your application to subscribe to data changes on a single collection, database, or deployment.

You can open a change stream to help perform the following actions:

• Enable devices to track and react to events, such as motion-detecting cameras

• Create an application that monitors changes in stock prices

• Generate a log of employee activity for specific job positions

You can specify a set of aggregation operators to filter and transform the data that your application receives. When connected to a MongoDB deployment v6.0 or later, you can also configure the events to include the document data before and after the change.

To learn how to open and configure a change stream, navigate to the following sections:

• Sample Data

• Open a Change Stream

• Apply Aggregation Operators to your Change Stream

• Include Pre-Images and Post-Images

• Additional Information

Sample Data

The examples in this guide monitor changes to the directors collection. Assume that this collection contains the following documents, which are modelled as structs:

let docs = vec! [
    Director {
        name: "Todd Haynes".to_string(),
        movies: vec! ["Far From Heaven".to_string(), "Carol".to_string()],
        oscar_noms: 1,
    },
    Director {
        name: "Jane Campion".to_string(),
        movies: vec! ["The Piano".to_string(), "Bright Star".to_string()],
        oscar_noms: 5,
    }
];

Open a Change Stream

You can open a change stream to subscribe to specific types of data changes and produce change events in your application.

To open a change stream, call the watch() method on a Collection, Database, or Client instance.

The struct type on which you call the watch() method determines the scope of events that the change stream listens for. The following table describes the behavior of the watch() method based on its calling object:

let mut change_stream = my_coll.watch().await?;
while let Some(event) = change_stream.next().await.transpose()? {
    println!("Operation performed: {:?}", event.operation_type);
    println!("Document: {:?}", event.full_document);
}

Operation performed: Insert
Document: Some(Director { name: "Wes Anderson", movies: [...], oscar_noms: 7 })

Apply Aggregation Operators to your Change Stream

You can chain the pipeline() method to the watch() method to specify which change events the change stream receives. Pass an aggregation pipeline as a parameter to pipeline().

To learn which aggregation operators your MongoDB Server version supports, see Modify Change Stream Output in the Server manual.

let mut update_change_stream = my_coll.watch()
    .pipeline(vec![doc! { "$match" : doc! { "operationType" : "update" } }])
    .await?;
while let Some(event) = update_change_stream.next().await.transpose()? {
    println!("Update performed: {:?}", event.update_description);
}

Update performed: Some(UpdateDescription { updated_fields: Document({
"oscar_noms": Int64(2)}), removed_fields: [], truncated_arrays: Some([]) })

Include Pre-Images and Post-Images

You can configure the change event to contain or omit the following data:

• Pre-image: a document that represents the version of the document before the operation, if it exists

• Post-image: a document that represents the version of the document after the operation, if it exists

To receive change stream events that include a pre-image or post-image, you must perform the following actions:

• Enable pre-images and post-images for the collection on your MongoDB deployment.

  Tip

  To learn how to enable pre- and post-images on your deployment, see Change Streams with Document Pre- and Post-Images in the Server manual.

  To learn how to instruct the driver to create a collection with pre-images and post-images enabled, see the Create a Collection with Pre-Image and Post-Images Enabled section of this page.

• Configure your change stream to retrieve either or both the pre-images and post-images. During this configuration, you can instruct the driver to require pre- and post-images or to only include them when available.

  Tip

  To configure your change stream to record the pre-image in change events, see the Pre-Image Configuration Example on this page.

  To configure your change stream to record the post-image in change events, see the Post-Image Configuration Example on this page.

let enable = ChangeStreamPreAndPostImages::builder().enabled(true).build();
let result = my_db.create_collection("directors")
    .change_stream_pre_and_post_images(enable)
    .await?;

let pre_image = FullDocumentBeforeChangeType::Required;
let mut change_stream = my_coll.watch()
    .full_document_before_change(pre_image)
    .await?;
while let Some(event) = change_stream.next().await.transpose()? {
    println!("Operation performed: {:?}", event.operation_type);
    println!("Pre-image: {:?}", event.full_document_before_change);
}

Operation performed: Update
Pre-image: Some(Director { name: "Jane Campion", movies: ["The Piano",
"Bright Star"], oscar_noms: 5 })

let post_image = FullDocumentType::WhenAvailable;
let mut change_stream = my_coll.watch()
    .full_document(post_image)
    .await?;
while let Some(event) = change_stream.next().await.transpose()? {
    println!("Operation performed: {:?}", event.operation_type);
    println!("Post-image: {:?}", event.full_document);
}

Operation performed: Delete
Post-image: None

Additional Information