Monitor Data with Change Streams

Sample Data

The examples in this guide use the following Course struct as a model for documents in the courses collection:

type Course struct {
	Title      string
	Enrollment int32
}

To run the examples in this guide, load these documents into the courses collection in the db database by using the following snippet:

coll := client.Database("db").Collection("courses")
docs := []interface{}{
	Course{Title: "World Fiction", Enrollment: 35},
	Course{Title: "Abstract Algebra", Enrollment: 60},
}
result, err := coll.InsertMany(context.TODO(), docs)

Each document contains a description of a university course that includes the course title and maximum enrollment, corresponding to the title and enrollment fields in each document.

Example

The following example opens a change stream on the courses collection and prints the change stream events as they occur:

changeStream, err := coll.Watch(context.TODO(), mongo.Pipeline{})
if err != nil {
	panic(err)
}
defer changeStream.Close(context.TODO())
// Iterates over the cursor to print the change stream events
for changeStream.Next(context.TODO()) {
	fmt.Println(changeStream.Current)
}

If you modify the courses collection in a separate program or shell, this code prints your changes as they occur. Inserting a document with a title value of "Advanced Screenwriting" and an enrollment value of 20 results in the following change event:

map[_id:map[_data:...] clusterTime: {...}
documentKey:map[_id:ObjectID("...")] fullDocument:map[_id:ObjectID("...")
enrollment:20 title:Advanced Screenwriting] ns: map[coll:courses db:db]
operationType:insert]

To view a fully runnable example, see Open a Change Stream Example: Full File section in this guide.

Filter Change Events

Use the pipeline parameter to modify the change stream output. This parameter allows you to only watch for certain change events. Format the pipeline parameter as an array of documents, with each document representing an aggregation stage.

You can use the following pipeline stages in this parameter:

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

The following example opens a change stream on the db database but only watches for new delete operations:

db := client.Database("db")
pipeline := bson.D{{"$match", bson.D{{"operationType", "delete"}}}}
changeStream, err := db.Watch(context.TODO(), mongo.Pipeline{pipeline})
if err != nil {
	panic(err)
}
defer changeStream.Close(context.TODO())
// Iterates over the cursor to print the delete operation change events
for changeStream.Next(context.TODO()) {
	fmt.Println(changeStream.Current)
}

Open a Change Stream Example: Full File

The following example opens a change stream on the restaurants collection and prints inserted documents:

coll := client.Database("sample_restaurants").Collection("restaurants")
// Creates instructions to watch for insert operations
pipeline := mongo.Pipeline{bson.D{{"$match", bson.D{{"operationType", "insert"}}}}}
// Creates a change stream that receives change events
cs, err := coll.Watch(context.TODO(), pipeline)
if err != nil {
	panic(err)
}
defer cs.Close(context.TODO())
fmt.Println("Waiting For Change Events. Insert something in MongoDB!")
// Prints a message each time the change stream receives an event
for cs.Next(context.TODO()) {
	var event bson.M
	if err := cs.Decode(&event); err != nil {
		panic(err)
	}
	output, err := json.MarshalIndent(event["fullDocument"], "", "    ")
	if err != nil {
		panic(err)
	}
	fmt.Printf("%s\n", output)
}
if err := cs.Err(); err != nil {
	panic(err)
}

View a fully runnable example.

// results truncated {
  "_id": ..., "name": "8282", "cuisine": "Korean"
}

Pre- and Post-Images

When you perform any CRUD operation on a collection, by default, the corresponding change event document contains only the delta of the fields modified by the operation. You can see the full document before and after a change, in addition to the delta, by specifying settings in the options parameter of the Watch() method.

If you want to see a document's post-image, the full version of the document after a change, set the FullDocument field of the options parameter to one of the following values:

• UpdateLookup: The change event document includes a copy of the entire changed document.

• WhenAvailable: The change event document includes a post-image of the modified document for change events if the post-image is available.

• Required: The output is the same as for WhenAvailable, but the driver raises a server-side error if the post-image is not available.

If you want to see a document's pre-image, the full version of the document before a change, set the FullDocumentBeforeChange field of the options parameter to one of the following values:

• WhenAvailable: The change event document includes a pre-image of the modified document for change events if the pre-image is available.

• Required: The output is the same as for WhenAvailable, but the driver raises a server-side error if the pre-image is not available.

Example

The following example calls the Watch() method on the courses collection. It specifies a value for the FullDocument field of the options parameter to output a copy of the entire modified document, instead of only the changed fields:

opts := options.ChangeStream().SetFullDocument(options.UpdateLookup)
changeStream, err := coll.Watch(context.TODO(), mongo.Pipeline{}, opts)
if err != nil {
	panic(err)
}
defer changeStream.Close(context.TODO())
for changeStream.Next(context.TODO()) {
	fmt.Println(changeStream.Current)
}

Updating the enrollment value of the document with the title of "World Fiction" from 35 to 30 results in the following change event:

{"_id": {"_data": "..."},"operationType": "update","clusterTime":
{"$timestamp": {"t":"...","i":"..."}},"fullDocument": {"_id":
{"$oid":"..."},"title": "World Fiction","enrollment": {"$numberInt":"30"}},
"ns": {"db": "db","coll": "courses"},"documentKey": {"_id": {"$oid":"..."}},
"updateDescription": {"updatedFields": {"enrollment": {"$numberInt":"30"}},
"removedFields": [],"truncatedArrays": []}}

Without specifying the FullDocument option, the same update operation no longer outputs the "fullDocument" value in the change event document.

API Documentation

To learn more about the Watch() method, see the following API documentation:

• Watch() for collections

• Watch() for databases

• Watch() for clients