Monitor Data Changes

On this page

Overview

Sample Data
Open a Change Stream
Modify the Change Stream Output
Modify watch() Behavior
Include Pre-Images and Post-Images
Additional Information
API Documentation

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 collection, database, or deployment.

When using the MongoDB PHP Library, you can call the watch() method to return an instance of MongoDB\ChangeStream. Then, you can iterate through the MongoDB\ChangeStream instance to monitor data changes, such as updates, insertions, and deletions.

Sample Data

The examples in this guide use the restaurants collection in the sample_restaurants database from the Atlas sample datasets. To access this collection from your PHP application, instantiate a MongoDB\Client that connects to an Atlas cluster and assign the following value to your $collection variable:

$collection = $client->sample_restaurants->restaurants;

Tip

To learn how to create a free MongoDB Atlas cluster and load the sample datasets, see the Get Started with Atlas guide.

Some examples use the toJSON() function to represent change events, which are BSON documents, as Extended JSON. To use this function, paste the following code into your application file:

function toJSON(object $document): string
{
    return MongoDB\BSON\Document::fromPHP($document)->toRelaxedExtendedJSON();
}

Open a Change Stream

To open a change stream, call the watch() method. The instance on which you call the watch() method determines the scope of events that the change stream monitors. You can call the watch() method on instances of the following classes:

MongoDB\Client: Monitor all changes in the MongoDB deployment
MongoDB\Database: Monitor changes in all collections in the database
MongoDB\Collection: Monitor changes in the collection

The following example opens a change stream on the restaurants collection and outputs changes as they occur:

$changeStream = $collection->watch();
for ($changeStream->rewind(); true; $changeStream->next()) {
    if ( ! $changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($event['operationType'] === 'invalidate') {
        break;
    }
}

To begin watching for changes, run the preceding code. Then, in a separate shell, modify the restaurants collection. The following example updates a document that has a name field value of 'Blarney Castle':

$result = $collection->updateOne(
    ['name' => 'Blarney Castle'],
    ['$set' => ['cuisine' => 'Irish']]
);

When you update the collection, the change stream application prints the change as it occurs. The printed change event resembles the following output:

{ "_id" : { "_data" : "..." }, "operationType" : "update", "clusterTime" :
{ "$timestamp" : { ... } }, "wallTime" : { "$date" : "..." }, "ns" : { "db" :
"sample_restaurants", "coll" : "restaurants" }, "documentKey" : { "_id" :
{ "$oid" : "..." } }, "updateDescription" : { "updatedFields" : { "cuisine" : "Irish" },
"removedFields" : [  ], "truncatedArrays" : [  ] } }

Modify the Change Stream Output

To modify the change stream output, you can pass pipeline stages in an array as a parameter to the watch() method. You can include the following stages in the array:

$addFields or $set: Adds new fields to documents
$match: Filters the documents
$project: Projects a subset of the document fields
$replaceWith or $replaceRoot: Replaces the input document with the specified document
$redact: Restricts the contents of the documents
$unset: Removes fields from documents

The following example passes a pipeline that includes the $match stage to the watch() method. This instructs the watch() method to output events only when update operations occur:

$pipeline = [['$match' => ['operationType' => 'update']]];
$changeStream = $collection->watch($pipeline);
for ($changeStream->rewind(); true; $changeStream->next()) {
    if ( ! $changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($event['operationType'] === 'invalidate') {
        break;
    }
}

Modify `watch()` Behavior

To modify the behavior of the watch() method, you can pass an options array as a parameter to watch(). The following table describes useful options you can set in the array:

Option	Description
`fullDocument`	Specifies whether to show the full document after the change, rather than showing only the changes made to the document. To learn more about this option, see the Include Pre-Images and Post-Images section of this guide.
`fullDocumentBeforeChange`	Specifies whether to show the full document as it was before the change, rather than showing only the changes made to the document. To learn more about this option, see Include Pre-Images and Post-Images.
`startAfter`	Instructs `watch()` to start a new change stream after the operation specified in the resume token. This field allows notifications to resume after an invalidate event. Each change stream event document includes a resume token as the `_id` field. Pass the entire `_id` field of the change event document that represents the operation you want to resume after. This option is mutually exclusive with `resumeAfter` and `startAtOperationTime`.
`startAtOperationTime`	Instructs the change stream to only provide changes that occurred at or after the specified timestamp. This option is mutually exclusive with `startAfter` and `resumeAfter`.
`collation`	Sets the collation to use for the change stream cursor.

For a full list of watch() options, see MongoDB\Collection::watch() in the API documentation.

Include Pre-Images and Post-Images

Important

You can enable pre-images and post-images on collections only if your deployment uses MongoDB v6.0 or later.

By default, when you perform an operation on a collection, the corresponding change event includes only the delta of the fields modified by that operation. To see the full document before or after a change, specify the fullDocumentBeforeChange or the fullDocument options in an array parameter to watch().

The pre-image is the full version of a document before a change. To include the pre-image in the change stream event, set the fullDocumentBeforeChange option to one of the following values:

MongoDB\Operation\Watch::FULL_DOCUMENT_BEFORE_CHANGE_WHEN_AVAILABLE: The change event includes a pre-image of the modified document for change events. If the pre-image is not available, this change event field has a null value.
MongoDB\Operation\Watch::FULL_DOCUMENT_BEFORE_CHANGE_REQUIRED: The change event includes a pre-image of the modified document for change events. If the pre-image is not available, the server raises an error.

The post-image is the full version of a document after a change. To include the post-image in the change stream event, set the fullDocument option to one of the following values:

MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP: The change event includes a copy of the entire changed document from some time after the change.
MongoDB\Operation\Watch::FULL_DOCUMENT_WHEN_AVAILABLE: The change event includes a post-image of the modified document for change events. If the post-image is not available, this change event field has a null value.
MongoDB\Operation\Watch::FULL_DOCUMENT_REQUIRED: The change event includes a post-image of the modified document for change events. If the post-image is not available, the server raises an error.

The following example calls the watch() method on a collection and includes the post-image of updated documents by setting the fullDocument option:

$options = ['fullDocument' => MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP];
$changeStream = $collection->watch([], $options);
for ($changeStream->rewind(); true; $changeStream->next()) {
    if ( ! $changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($event['operationType'] === 'invalidate') {
        break;
    }
}

With the change stream application running in a separate shell, updating a document in the restaurants collection by using the preceding update example prints a change event resembling the following output:

{ "_id" : { "_data" : "..." }, "operationType" : "update", "clusterTime" :
{ "$timestamp" : { ... } }, "wallTime" : { "$date" : "..." },
"fullDocument" : { "_id" : { "$oid" : "..." }, "address" : { "building" :
"202-24", "coord" : [ -73.925044200000002093, 40.559546199999999772 ], "street"
: "Rockaway Point Boulevard", "zipcode" : "11697" }, "borough" : "Queens",
"cuisine" : "Irish", "grades" : [ ...], "name" : "Blarney Castle", "restaurant_id" :
"40366356" }, "ns" : { "db" : "sample_restaurants", "coll" : "restaurants" },
"documentKey" : { "_id" : { "$oid" : "..." } }, "updateDescription" :
{ "updatedFields" : { "cuisine" : "Irish" }, "removedFields" : [  ],
"truncatedArrays" : [  ] } }

Tip

To learn more about pre-images and post-images, see Change Streams with Document Pre- and Post-Images in the MongoDB Server manual.

Additional Information

To learn more about change streams, see Change Streams in the MongoDB Server manual.

API Documentation

To learn more about any of the methods or types discussed in this guide, see the following API documentation:

Back

Distinct Field Values

Write Data