Nic Raboy

<div>
    <img style="float: right;
        border-radius: 10px;
        margin-bottom: 30px;
        vertical-align: bottom;
        width: 30%;"
        src="https://www.mongodb.com/developer/images/qs-badges/qs-badge-go.png"
        alt="QuickStart Golang Logo"
    />

If you've been keeping up with my getting started with Go and MongoDB tutorial series, you'll remember that we've accomplished quite a bit so far. We've had a look at everything from CRUD interaction with the database to data modeling, and more. To play catch up with everything we've done, you can have a look at the following tutorials in the series:
</div>

-   [How to Get Connected to Your MongoDB Cluster with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--starting-and-setup)
-   [Creating MongoDB Documents with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--how-to-create-documents)
-   [Retrieving and Querying MongoDB Documents with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--how-to-read-documents)
-   [Updating MongoDB Documents with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--how-to-update-documents)
-   [Deleting MongoDB Documents with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--how-to-delete-documents)
-   [Modeling MongoDB Documents with Native Go Data Structures](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--modeling-documents-with-go-data-structures)
-   [Performing Complex MongoDB Data Aggregation Queries with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--data-aggregation-pipeline)

In this tutorial we're going to explore [change streams](https://docs.mongodb.com/manual/changeStreams/) in MongoDB and how they might be useful, all with the Go programming language (Golang).

Before we take a look at the code, let's take a step back and understand what change streams are and why there's often a need for them.

Imagine this scenario, one of many possible:

You have an application that engages with internet of things (IoT) clients. Let's say that this is a geofencing application and the IoT clients are something that can trigger the geofence as they come in and out of range. Rather than having your application constantly run queries to see if the clients are in range, wouldn't it make more sense to watch in real-time and react when it happens?

With MongoDB change streams, you can create a pipeline to watch for changes on a collection level, database level, or deployment level, and write logic within your application to do something as data comes in based on your pipeline.

## Creating a Real-Time MongoDB Change Stream with Golang

While there are many possible use-cases for change streams, we're going to continue with the example that we've been using throughout the scope of this getting started series. We're going to continue working with podcast show and podcast episode data.

Let's assume we have the following code to start:

``` go
package main

import (
   "context"
   "fmt"
   "os"
   "sync"

   "go.mongodb.org/mongo-driver/bson"
   "go.mongodb.org/mongo-driver/mongo"
   "go.mongodb.org/mongo-driver/mongo/options"
)

func main() {
   client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(os.Getenv("ATLAS_URI")))
   if err != nil {
      panic(err)
   }
   defer client.Disconnect(context.TODO())

   database := client.Database("quickstart")
   episodesCollection := database.Collection("episodes")
}
```

The above code is a very basic connection to a MongoDB cluster, something that we explored in the [How to Get Connected to Your MongoDB Cluster with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--starting-and-setup), tutorial.

To watch for changes, we can do something like the following:

``` go
episodesStream, err := episodesCollection.Watch(context.TODO(), mongo.Pipeline{})
if err != nil {
   panic(err)
}
```

The above code will watch for any and all changes to documents within the `episodes` collection. The result is a cursor that we can iterate over indefinitely for data as it comes in.

We can iterate over the curser and make sense of our data using the following code:

``` go
episodesStream, err := episodesCollection.Watch(context.TODO(), mongo.Pipeline{})
if err != nil {
   panic(err)
}

defer episodesStream.Close(context.TODO())

for episodesStream.Next(context.TODO()) {
   var data bson.M
   if err := episodesStream.Decode(&data); err != nil {
      panic(err)
   }
   fmt.Printf("%v\n", data)
}
```

If data were to come in, it might look something like the following:

``` none
map[_id:map[_data:825E4EFCB9000000012B022C0100296E5A1004D960EAE47DBE4DC8AC61034AE145240146645F696400645E3B38511C9D4400004117E80004] clusterTime:{1582234809 1} documentKey:map[_id:ObjectID("5e3b38511c9d
4400004117e8")] fullDocument:map[_id:ObjectID("5e3b38511c9d4400004117e8") description:The second episode duration:30 podcast:ObjectID("5e3b37e51c9d4400004117e6") title:Episode #3] ns:map[coll:episodes 
db:quickstart] operationType:replace]
```

In the above example, I've done a `Replace` on a particular document in the collection. In addition to information about the data, I also receive the full document that includes the change. The results will vary depending on the `operationType` that takes place.

While the code that we used would work fine, it is currently a blocking operation. If we wanted to watch for changes and continue to do other things, we'd want to use a [goroutine](https://tour.golang.org/concurrency/1) for iterating over our change stream cursor.

We could make some changes like this:

``` go
package main

import (
   "context"
   "fmt"
   "os"
   "sync"

   "go.mongodb.org/mongo-driver/bson"
   "go.mongodb.org/mongo-driver/mongo"
   "go.mongodb.org/mongo-driver/mongo/options"
)

func iterateChangeStream(routineCtx context.Context, waitGroup sync.WaitGroup, stream *mongo.ChangeStream) {
   defer stream.Close(routineCtx)
   defer waitGroup.Done()
   for stream.Next(routineCtx) {
      var data bson.M
      if err := stream.Decode(&data); err != nil {
            panic(err)
      }
      fmt.Printf("%v\n", data)
   }
}

func main() {
   client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(os.Getenv("ATLAS_URI")))
   if err != nil {
      panic(err)
   }
   defer client.Disconnect(context.TODO())

   database := client.Database("quickstart")
   episodesCollection := database.Collection("episodes")

   var waitGroup sync.WaitGroup

   episodesStream, err := episodesCollection.Watch(context.TODO(), mongo.Pipeline{})
   if err != nil {
      panic(err)
   }
   waitGroup.Add(1)
   routineCtx, cancelFn := context.WithCancel(context.Background())
   go iterateChangeStream(routineCtx, waitGroup, episodesStream)

   waitGroup.Wait()
}
```

A few things are happening in the above code. We've moved the stream iteration into a separate function to be used in a goroutine. However, running the application would result in it terminating quite quickly because the `main` function will terminate not too longer after creating the goroutine. To resolve this, we are making use of a `WaitGroup`. In our example, the `main` function will wait until the `WaitGroup` is empty and the `WaitGroup` only becomes empty when the goroutine terminates.

Making use of the `WaitGroup` isn't an absolute requirement as there are other ways to keep the application running while watching for changes.  However, given the simplicity of this example, it made sense in order to see any changes in the stream.

To keep the `iterateChangeStream` function from running indefinitely, we are creating and passing a context that can be canceled. While we don't demonstrate canceling the function, at least we know it can be done.

## Complicating the Change Stream with the Aggregation Pipeline

In the previous example, the aggregation pipeline that we used was as basic as you can get. In other words, we were looking for any and all changes that were happening to our particular collection. While this might be good in a lot of scenarios, you'll probably get more out of using a better defined aggregation pipeline.

Take the following for example:

``` go
matchPipeline := bson.D{
   {
      "$match", bson.D{
            {"operationType", "insert"},
            {"fullDocument.duration", bson.D{
               {"$gt", 30},
            }},
      },
   },
}

episodesStream, err := episodesCollection.Watch(context.TODO(), mongo.Pipeline{matchPipeline})
```

In the above example, we're still watching for changes to the `episodes` collection. However, this time we're only watching for new documents that have a `duration` field greater than 30. Any other insert or other change stream operation won't be detected.

The results of the above code, when a match is found, might look like the following:

``` none
map[_id:map[_data:825E4F03CF000000012B022C0100296E5A1004D960EAE47DBE4DC8AC61034AE145240146645F696400645E4F03A01C9D44000063CCBD0004] clusterTime:{1582236623 1} documentKey:map[_id:ObjectID("5e4f03a01c9d
44000063ccbd")] fullDocument:map[_id:ObjectID("5e4f03a01c9d44000063ccbd") description:a quick start into mongodb duration:35 podcast:1234 title:getting started with mongodb] ns:map[coll:episodes db:qui
ckstart] operationType:insert]
```

With change streams, you'll have access to a subset of the MongoDB aggregation pipeline and its operators. You can learn more about what's available in the [official documentation](http://docs.mongodb.com/manual/changeStreams/#modify-change-stream-output).

## Conclusion

You just saw how to use MongoDB change streams in a Golang application using the MongoDB Go driver. As previously pointed out, change streams make it very easy to react to database, collection, and deployment changes without having to constantly query the cluster. This allows you to efficiently plan out aggregation pipelines to respond to as they happen in real-time.

If you're looking to catch up on the other tutorials in the MongoDB with Go quick start series, you can find them below:

-   [How to Get Connected to Your MongoDB Cluster with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--starting-and-setup)
-   [Creating MongoDB Documents with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--how-to-create-documents)
-   [Retrieving and Querying MongoDB Documents with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--how-to-read-documents)
-   [Updating MongoDB Documents with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--how-to-update-documents)
-   [Deleting MongoDB Documents with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--how-to-delete-documents)
-   [Modeling MongoDB Documents with Native Go Data Structures](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--modeling-documents-with-go-data-structures)
-   [Performing Complex MongoDB Data Aggregation Queries with Go](https://www.mongodb.com/blog/post/quick-start-golang--mongodb--data-aggregation-pipeline)

To bring the series to a close, the next tutorial will focus on transactions with the MongoDB Go driver.

Learn how to use change streams to react to changes to MongoDB documents, databases, and clusters in real-time using the Go programming language.

Reacting to Database Changes with MongoDB Change Streams and Go

Maxime Beugnet

Learn how to use the Change Streams using the MongoDB Java Driver.

Java - Change Streams

Lauren Schaefer

Discover how to react to changes in your MongoDB database using change streams implemented in Node.js and Atlas triggers.

Change Streams & Triggers with Node.js Tutorial

Naomi Pentrel

## Introduction

<div>
    <img
        style="float: right; width: 30%;"
        src="https://www.mongodb.com/developer/images/qs-badges/qs-badge-python.png"
        alt="QuickStart Python Logo" />

Change streams allow you to listen to changes that occur in your MongoDB database. On MongoDB 3.6 or above, this functionality allows you to build applications that can immediately respond to real time data changes. In this tutorial, we'll show you how to use change streams with Python. In particular you will:
</div>

-   Learn about change streams
-   Create a program that listens to inserts
-   Change the program to listen to other event types
-   Change the program to listen to specific notifications

To follow along, you can create a test environment using the steps below. This is optional but highly encouraged as it will allow you to test usage of the change stream functionality with the examples provided. You will be given all commands, but some familiarity with MongoDB is needed.

## Learn about Change Streams

The ability to listen to specific changes in the data allows an application to be much faster in responding to change. If a user of your system updates their information, the system can listen to and propagate these changes right away. For example, this could mean users no longer have to click refresh to see when changes have been applied. Or if a user's changes in one system need approval by someone, another system could listen to changes and send notifications requesting approvals instantaneously.

Before change streams, applications that needed to know about the addition of new data in real-time had to continuously poll data or rely on other update mechanisms. One common, if complex, technique for monitoring changes was tailing MongoDB's Operation Log (Oplog). The Oplog is part of the replication system of MongoDB and as such already tracks modifications to the database but is not easy to use for business logic. Change streams are built on top of the Oplog but they provide a native API that improves efficiency and usability. Note that you cannot open a change stream against a collection in a standalone MongoDB server because the feature relies on the Oplog which is only used on replica sets.

When registering a change stream you need to specify the collection and what types of changes you want to listen to. You can do this by using the `$match` and a few other aggregation pipeline stages which limit the amount of data you will receive. If your database enforces authentication and authorization, change streams provide the same access control as for normal queries.

## Test the Change Stream Features

The best way to understand how change streams operate is to work with them. In the next section, we'll show you how to set up a server and scripts. After completing the setup, you will get two scripts: One Python script will listen to notifications from the change stream and print them. The other script will mimic an application by performing insert, update, replace, and delete operations so that you can see the notifications in the output of the first script. You will also learn how to limit the notifications to the ones you are interested in.

## Set up PyMongo

To get started, set up a virtual environment using [Virtualenv](https://virtualenv.pypa.io/en/stable/). Virtualenv allows you to isolate dependencies of your project from other projects. Create a directory for this project and copy the following into a file called requirements.txt in your new directory:

``` none
pymongo==3.8.0
dnspython
```

To create and activate your virtual environment, run the following commands in your terminal:

``` bash
virtualenv venv  # sets up the environment
source venv/bin/activate  # activates the environment
pip3 install -r requirements.txt  # installs our dependencies
```

>For ease of reading, we assume you are running Python 3 with the python3 and pip3 commands. If you are running Python 2.7, substitute python and pip for those commands.

## Set up your Cluster

We will go through two options for setting up a test MongoDB Replica Set for us to connect to. If you have MongoDB 3.6 or later installed and are comfortable making changes to your local setup choose this option and follow [the guide in the appendix](#how-to-set-up-a-local-cluster) and skip to the next section.

If you do not have MongoDB installed, would prefer not to mess with your local setup or if you are fairly new to MongoDB then we recommend that you set up a MongoDB Atlas cluster; there's a free tier which gives you a three node replica set which is ideal for experimenting and learning with. Simply follow [these steps](https://docs.mongodb.com/manual/tutorial/atlas-free-tier-setup/) until you get the URI connection string in step 8. Take that URI connection string, insert the password where it says `<password>`, and add it to your environment by running

``` bash
export CHANGE_STREAM_DB="mongodb+srv://user:<password>@example-xkfzv.mongodb.net/test?retryWrites=true"
```

in your terminal. The string you use as a value will be different.

## Listen to Inserts from an Application

Before continuing, quickly test your setup. Create a file `test.py` with the following contents:

``` python
import os
import pymongo

client = pymongo.MongoClient(os.environ['CHANGE_STREAM_DB'])
print(client.changestream.collection.insert_one({"hello": "world"}).inserted_id)
```

When you run `python3 test.py` you should see an `ObjectId` being printed.

Now that you've confirmed your setup, let's create the small program that will listen to changes in the database using a change stream.  Create a different file `change_streams.py` with the following content:

``` python
import os
import pymongo
from bson.json_util import dumps

client = pymongo.MongoClient(os.environ['CHANGE_STREAM_DB'])
change_stream = client.changestream.collection.watch()
for change in change_stream:
    print(dumps(change))
    print('') # for readability only
```

Go ahead and run `python3 change_streams.py`, you will notice that the program doesn't print anything and just waits for operations to happen on the specified collection. While keeping the `change_streams` program running, open up another terminal window and run `python3 test.py`. You will have to run the same export command you ran in the *Set up your Cluster* section to add the environment variable to the new terminal window.

Checking the terminal window that is running the `change_streams` program, you will see that the insert operation was logged. It should look like the output below but with a different `ObjectId` and with a different value for `$binary`.

``` json
➜ python3 change_streams.py
{"_id": {"_data": {"$binary": "glsIjGUAAAABRmRfaWQAZFsIjGXiJuWPOIv2PgBaEAQIaEd7r8VFkazelcuRgfgeBA==", "$type": "00"}}, "operationType": "insert", "fullDocument": {"_id": {"$oid": "5b088c65e226e58f388bf63e"}, "hello": "world"}, "ns": {"db": "changestream", "coll": "collection"}, "documentKey": {"_id": {"$oid": "5b088c65e226e58f388bf63e"}}}
```

## Listen to Different Event Types

You can listen to four types of document-based events:

-   Insert
-   Update
-   Replace
-   Delete

Depending on the type of event the document structure you will receive will differ slightly but you will always receive the following:

``` json
{
    _id: <resume_token>,
    operationType: "<type>",
    ns: {db: "<db name>", coll: "<collection name>"},
    documentKey: { <unique identifier> }
}
```

In the case of inserts and replace operations the `fullDocument` is provided by default as well. In the case of update operations the extra field provided is `updateDescription` and it gives you the document delta (i.e. the difference between the document before and after the operation). By default update operations only include the delta between the document before and after the operation. To get the full document with each update you can [pass in "updateLookup" to the full document option](https://docs.mongodb.com/manual/changeStreams/#lookup-full-document-for-update-operations).  If an update operation ends up changing multiple documents, there will be one notification for each updated document. This transformation occurs to ensure that statements in the oplog are [idempotent](https://docs.mongodb.com/manual/core/replica-set-oplog/).

There is one further type of event that can be received which is the invalidate event. This tells the driver that the change stream is no longer valid. The driver will then close the stream. Potential reasons for this include the collection being dropped or renamed.

To see this in action update your `test.py` and run it while also running the `change_stream` program:

``` python
import os
import pymongo

client = pymongo.MongoClient(os.environ['CHANGE_STREAM_DB'])
client.changestream.collection.insert_one({"_id": 1, "hello": "world"})
client.changestream.collection.update_one({"_id": 1}, {"$set": {"hello": "mars"}})
client.changestream.collection.replace_one({"_id": 1} , {"bye": "world"})
client.changestream.collection.delete_one({"_id": 1})
client.changestream.collection.drop()
```

The output should be similar to:

``` json
➜ python3 change_streams.py
{"fullDocument": {"_id": 1, "hello": "world"}, "documentKey": {"_id": 1}, "_id": {"_data": {"$binary": "glsIjuEAAAABRh5faWQAKwIAWhAECGhHe6/FRZGs3pXLkYH4HgQ=", "$type": "00"}}, "ns": {"coll": "collection", "db": "changestream"}, "operationType": "insert"}

{"documentKey": {"_id": 1}, "_id": {"_data": {"$binary": "glsIjuEAAAACRh5faWQAKwIAWhAECGhHe6/FRZGs3pXLkYH4HgQ=", "$type": "00"}}, "updateDescription": {"removedFields": [], "updatedFields": {"hello": "mars"}}, "ns": {"coll": "collection", "db": "changestream"}, "operationType": "update"}

{"fullDocument": {"bye": "world", "_id": 1}, "documentKey": {"_id": 1}, "_id": {"_data": {"$binary": "glsIjuEAAAADRh5faWQAKwIAWhAECGhHe6/FRZGs3pXLkYH4HgQ=", "$type": "00"}}, "ns": {"coll": "collection", "db": "changestream"}, "operationType": "replace"}

{"documentKey": {"_id": 1}, "_id": {"_data": {"$binary": "glsIjuEAAAAERh5faWQAKwIAWhAECGhHe6/FRZGs3pXLkYH4HgQ=", "$type": "00"}}, "ns": {"coll": "collection", "db": "changestream"}, "operationType": "delete"}

{"_id": {"_data": {"$binary": "glsIjuEAAAAFFFoQBAhoR3uvxUWRrN6Vy5GB+B4E", "$type": "00"}}, "operationType": "invalidate"}
```

## Listen to Specific Notifications

So far, your program has been listening to *all* operations. In a real application this would be overwhelming and often unnecessary as each part of your application will generally want to listen only to specific operations. To limit the amount of operations, you can use certain aggregation stages when setting up the stream. These stages are: `$match`, `$project`, `$addfields`, `$replaceRoot`, and `$redact`. All other aggregation stages are not available.

You can test this functionality by changing your `change_stream.py` file with the code below and running the `test.py` script. The output should now only contain insert notifications.

``` python
import os
import pymongo
from bson.json_util import dumps

client = pymongo.MongoClient(os.environ['CHANGE_STREAM_DB'])
change_stream = client.changestream.collection.watch([{
    '$match': {
        'operationType': { '$in': ['insert'] }
    }
}])

for change in change_stream:
    print(dumps(change))
    print('')
```

You can also *match* on document fields and thus limit the stream to certain `DocumentIds` or to documents that have a certain document field, etc.

## Resume your Change Streams

No matter how good your network, there will be situations when connections fail. To make sure that no changes are missed in such cases, you need to add some code for storing and handling `resumeToken`s. Each event contains a `resumeToken`, for example:

``` json
"_id": {"_data": {"$binary": "glsIj84AAAACRh5faWQAKwIAWhAEvyfcy4djS8CUKRZ8tvWuOgQ=", "$type": "00"}}
```

When a failure occurs, the driver should automatically make one attempt to reconnect. The application has to handle further retries as needed.  This means that the application should take care of always persisting the `resumeToken`.

To retry connecting, the `resumeToken` has to be passed into the optional field resumeAfter when creating the new change stream. This does not guarantee that we can always resume the change stream. MongoDB's oplog is a capped collection that keeps a rolling record of the most recent operations. Resuming a change stream is only possible if the oplog has not rolled yet (that is if the changes we are interested in are still in the oplog).

## Caveats

-   **Change Streams in Production**: If you plan to use change streams in production, please read [MongoDB's recommendations](https://docs.mongodb.com/manual/administration/change-streams-production-recommendations/).
-   **Ordering and Rollbacks**: MongoDB guarantees that the received events will be in the order they occurred (thus providing a total ordering of changes across shards if you use shards). On top of that only durable, i.e. majority committed changes will be sent to listeners. This means that listeners do not have to consider rollbacks in their applications.
-   **Reading from Secondaries**: Change streams can be opened against any data-bearing node in a cluster regardless whether it's primary or secondary. However, it is generally not recommended to read from secondaries as failovers can lead to increased load and failures in this setup.
-   **Updates with the fullDocument Option**: The fullDocument option for Update Operations does not guarantee the returned document does not include further changes. In contrast to the document deltas that are guaranteed to be sent in order with update notifications, there is no guarantee that the *fullDocument* returned represents the document as it was exactly after the operation. `updateLookup` will poll the current version of the document. If changes happen quickly it is possible that the document was changed before the `updateLookup` finished. This means that the fullDocument might not represent the document at the time of the event thus potentially giving the impression events took place in a different order.
-   **Impact on Performance**: Up to 1,000 concurrent change streams to each node are supported with negligible impact on the overall performance. However, on [sharded clusters](https://docs.mongodb.com/manual/administration/change-streams-production-recommendations/#sharded-clusters), the guarantee of total ordering could cause response times of the change stream to be slower.
-   **WiredTiger**: Change streams are a MongoDB 3.6 and later feature.  It is not available for older versions, MMAPv1 storage or pre [pv1](https://docs.mongodb.com/manual/reference/replica-configuration/#rsconf.protocolVersion) replications.

## Learn More

To read more about this check out the [Change Streams documentation](https://docs.mongodb.com/manual/changeStreams/).

If you're interested in more MongoDB tips, follow us on Twitter [@mongodb](http://www.twitter.com/mongodb).

## Appendix

### How to set up a Cluster in the Cloud

>If you haven't yet set up your free cluster on [MongoDB Atlas](http://bit.ly/mongodbatlas), now is a great time to do so. You have all the instructions in this [blog post](/quickstart/free-atlas-cluster).

### How to set up a Local Cluster

Before setting up the instances please confirm that you are running version 3.6 or later of the MongoDB Server (mongod) and the MongoDB shell (mongo). You can do this by running `mongod --version` and `mongo --version`. If either of these do not satisfy our requirements, please [upgrade to a more recent version](https://docs.mongodb.com/manual/tutorial/upgrade-revision/#upgrade-replace-binaries) before continuing.

In the following you will set up a single-node replica-set named `test-change-streams`. For a production replica-set, at least three nodes are recommended.

1.  Run the following commands in your terminal to create a directory for the database files and start the mongod process on port `27017`:

    ``` bash
    mkdir -p /data/test-change-streams
    mongod --replSet test-change-streams --logpath "mongodb.log" --dbpath /data/test-change-streams --port 27017 --fork
    ```

2.  Now open up a mongo shell on port `27017`:

    ``` bash
    mongo --port 27017
    ```

3.  Within the mongo shell you just opened, configure your replica set:

    ``` javascript
    config = {
        _id: "test-change-streams",
        members: [{ _id : 0, host : "localhost:27017"}]
    };
    rs.initiate(config);
    ```

4.  Still within the mongo shell, you can now check that your replica set is working by running: `rs.status();`. The output should indicate that your node has become primary. It may take a few seconds to show this so if you are not seeing this immediately, run the command again after a few seconds.

5.  Run

    ``` bash
    export CHANGE_STREAM_DB=mongodb://localhost:27017
    ```

    in your shell and [continue](#listen-to-inserts-from-an-application).

Change streams allow you to listen to changes that occur in your MongoDB database.

MongoDB Change Streams with Python



<div class="introduction">

>
>
>Please note: This article discusses Stitch. Stitch is now MongoDB Realm. All the same features and functionality, now with a new name. Learn more [here](https://www.mongodb.com/realm). We will be updating this article in due course.

When it comes to modern web applications, interactions often need to be done in real-time. This means that instead of periodically checking in for changes, watching or listening for changes often makes more sense.

Take the example of tracking something on a map. When it comes to package shipments, device tracking, or anything else where you need to know the real-time location, watching for those changes in location is great. Imagine needing to know where your fleet is so that you can dispatch them to a nearby incident?

When it comes to MongoDB, watching for changes can be done through change streams. These change streams can be used in any of the drivers, including front-end applications with MongoDB Stitch.

In this tutorial, we're going to leverage [MongoDB Stitch](https://www.mongodb.com/cloud/stitch) change streams. When the location data in our NoSQL documents change, we're going to update the information on an interactive map powered by [Mapbox](https://www.mapbox.com).

Take the following animated image for example:

![mapbox automatically updated by a change stream](https://www.mongodb.com/developer/images/how-to/mapbox-mongodb-change-streams.gif)

Rather than building an Internet of Things (IoT) device to track and submit GPS data, we're going to simulate the experience by directly changing our documents in MongoDB. When the update operations are complete, the front-end application with the interactive map is watching for those changes and responding appropriately.

## The Requirements

To be successful with this example, you'll need to have a few things ready to go prior:

-   A MongoDB Atlas cluster
-   A MongoDB Stitch application
-   A Mapbox account

For this example, the data will exist in MongoDB Atlas. Since we're planning on interacting with our data using a front-end application, we'll be using MongoDB Stitch. A Stitch application should be created within the MongoDB Cloud and connected to the MongoDB Atlas cluster prior to exploring this tutorial.

>Get started with MongoDB Atlas and Stitch for FREE in the [MongoDB Cloud](https://cloud.mongodb.com).

Mapbox will be used as our interactive map. Since Mapbox is a service, you'll need to have created an account and have access to your access token.

In the animated image, I'm using the [MongoDB Visual Studio Code plugin](https://marketplace.visualstudio.com/items?itemName=mongodb.mongodb-vscode) for interacting with the documents in my collection. You can do the same or use another tool such as Compass, the CLI, or the data explorer within Atlas to get the job done.

## Understanding the Document Model for the Location Tracking Example

Because we're only planning on moving a marker around on a map, the data model that we use doesn't need to be extravagant. For this example, the following is more than acceptable:

``` json
{
    "_id": "5ec44f70fa59d66ba0dd93ae",
    "coordinates": [
        -121.4252,
        37.7397
    ],
    "username": "nraboy"
}
```

In the above example, the coordinates array has the first item representing the longitude and the second item representing the latitude. We're including a username to show that we are going to watch for changes based on a particular document field. In a polished application, all users probably wouldn't be watching for changes for all documents. Instead they'd probably be watching for changes of documents that belong to them.

While we could put authorization rules in place for users to access certain documents, it is out of the scope of this example. Instead, we're going to mock it.

## Building a Real-Time Location Tracking Application with Mapbox and the Stitch SDK

Now we're going to build our client-facing application which consists of Mapbox, some basic HTML and JavaScript, and MongoDB Stitch.

Let's start by adding the following boilerplate code:

``` xml
<!DOCTYPE html>
<head>
    <script src="https://api.mapbox.com/mapbox-gl-js/v1.10.0/mapbox-gl.js"></script>
    <link href="https://api.mapbox.com/mapbox-gl-js/v1.10.0/mapbox-gl.css" rel="stylesheet" />
    <script src="https://s3.amazonaws.com/stitch-sdks/js/bundles/4.6.0/stitch.js"></script>
</head>
<html>
    <body style="margin: 0">
        <div id="map" style="width: 100vw; height: 100vh"></div>
        <script>
            // Logic here ...
        </script>
    </body>
</html>
```

The above code sets us up by including the Mapbox and MongoDB Stitch SDKs. When it comes to querying MongoDB and interacting with the map, we're going to be doing that from within the `<script>` tags.

Within the `<script>` tags, let's get started by adding the following:

``` javascript
const client = stitch.Stitch.initializeDefaultAppClient('STITCH_APP_ID_HERE');
const db = client.getServiceClient(stitch.RemoteMongoClient.factory, "mongodb-atlas").db("location_services");
mapboxgl.accessToken = "MAPBOX_ACCESS_TOKEN_HERE";

var currentLocationMarker;
```

The above lines of code are useful for the initialization of our services. You'll want to swap the app id with your actual Stitch app id and the Mapbox access token with your actual Mapbox access token. Both of these can be found within each of the services dashboards.

For this example, I'm going to assume you're using a **location_services** database within MongoDB Atlas and a **tracking** collection within that database.

The `currentLocationMarker` variable will represent our changing marker that moves around on the map as new data comes in from the MongoDB change stream.

Within the same `<script>` tags, we can initialize the map for displaying:

``` javascript
let map = new mapboxgl.Map({
    container: "map",
    style: "mapbox://styles/mapbox/streets-v11",
    center: [-121.4252, 37.7397],
    zoom: 9
});
```

Since the map tiles that compose the map come from a service, we need to wait until the map is ready before we start interacting with it. We can make use of the Mapbox `load` event to let us know the map is ready:

``` javascript
map.on("load", async () => {
    await client.auth.loginWithCredential(new stitch.AnonymousCredential());
    let currentLocation = (await db.collection("tracking").findOne({ "username": "nraboy" })).coordinates;
    currentLocationMarker = new mapboxgl.Marker().setLngLat(currentLocation).addTo(map);
    const stream = await db.collection("tracking").watch({
        "fullDocument.username": "nraboy"
    });
    stream.onNext(event => {
        currentLocationMarker.setLngLat(event.fullDocument.coordinates);
    });
});
```

Inside the `load` event, we are doing anonymous authentication with MongoDB Stitch. Remember, we could very easily use a stronger authentication method and have authorization rules, but for this example it's out of the scope.

Once we're authenticated to Stitch, we execute a `findOne` operation based on the mock `username` field on our document. The only data we care about is the coordinates, but we want to make sure it comes from the correct `username` field.

With the latitude and longitude coordinate information in hand, we can update the marker position.

Up until now, we are just setting the marker to the last known position. This is because when we start watching with a change stream, we won't receive an initial document. This is why we are doing the `findOne` first.

This brings us to the change stream:

``` javascript
const stream = await db.collection("tracking").watch({
    "fullDocument.username": "nraboy"
});
stream.onNext(event => {
    currentLocationMarker.setLngLat(event.fullDocument.coordinates);
});
```

For this particular change stream, we are only watching for documents where the `username` field matches. In a more polished and realistic example, you could use this to watch for only your own documents. We're mocking this by hard-coding the value.

When documents with the `username` field experience some kind of change within Atlas, the marker location will update. This is done without us having to constantly query for updates based on a timer.

## Conclusion

You just saw how to use change streams in a client-facing application using the MongoDB Stitch SDK. To make our example more attractive, we went with a location tracking scenario, whereas when latitude and longitude locations change in the database, the positions are updated in real-time on a map powered by [Mapbox](https://www.mapbox.com).

>If you'd like to give MongoDB Atlas and MongoDB Stitch a try, there's a forever FREE tier available through the [MongoDB Cloud](https://cloud.mongodb.com).

When it comes to location data and MongoDB, there's so much more you can do. If you'd like to learn how to create and manage geofences, check out my previous tutorial titled, [Location Geofencing with MongoDB, Stitch, and Mapbox](https://developer.mongodb.com/how-to/location-geofencing-stitch-mapbox).

Learn how to use change streams with MongoDB Stitch to update location on a Mapbox map in real-time.

Real-Time Location Updates with MongoDB Stitch, Change Streams, and Mapbox

Aaron Bassett



<div class="introduction">

[Change streams](https://docs.mongodb.com/manual/changeStreams/) allow applications to access real-time data changes without the complexity and risk of tailing the [oplog](https://docs.mongodb.com/manual/reference/glossary/#term-oplog). Applications can use change streams to subscribe to all data changes on a single collection, a database, or an entire deployment, and immediately react to them. Because change streams use the aggregation framework, applications can also filter for specific changes or transform the notifications at will.

## Prerequisites

The example code in this article will assume you have a cluster running in [MongoDB Atlas](https://www.mongodb.com/cloud/atlas), but it will work with any MongoDB version from 3.6 onwards.

You will also need [Python 3.6+](https://docs.python.org/3/whatsnew/3.6.html#whatsnew36-pep498) with [Motor](https://motor.readthedocs.io/en/stable/index.html) and [Tornado](https://www.tornadoweb.org/en/stable/) installed.

``` bash
pip install motor
pip install tornado
pip install dnspython
pip install logzero
```

## Creating a WebSocket Server

To allow clients to subscribe to your change stream via WebSockets, you must first create a WebSocket server. This WebSocket server, written in Python and using Tornado, proxies any new data from the change stream to your connected clients.

``` python
class ChangesHandler(tornado.websocket.WebSocketHandler):

    connected_clients = set()

    def open(self):
        ChangesHandler.connected_clients.add(self)

    def on_close(self):
        ChangesHandler.connected_clients.remove(self)

    @classmethod
    def send_updates(cls, message):
        for connected_client in cls.connected_clients:
            connected_client.write_message(message)

    @classmethod
    def on_change(cls, change):
        message = f"{change['operationType']}: {change['fullDocument']['name']}"
        ChangesHandler.send_updates(message)
```

### Opening and Closing WebSocket Connections

As clients connect and disconnect from the WebSocket server, they trigger the `open` and `on_close` methods.

``` python
connected_clients = set()

def open(self):
    ChangesHandler.connected_clients.add(self)

def on_close(self):
    ChangesHandler.connected_clients.remove(self)
```

When a client connects, your server stores a reference to it in the `connected_clients` set. This allows it to push new data to the client when it is received from the change stream. Likewise, when a client disconnects from the server, it is removed from the set of connected clients, so your server does not try to push updates on a connection which no longer exists.

It is worth noting that the server does not have a `on_message` handler. As WebSockets are bi-directional, typically a WebSocket server has a `on_message` method. When the client sends data to the server, it invokes this method to handle the incoming message. However, as you are only using the WebSocket connection to push change stream data to the connected clients, your WebSocket connection is essentially mono-directional, and your server does not have a method for handling inbound data.

### Pushing Messages to Connected Clients

``` python
@classmethod
def send_updates(cls, message):
    for connected_client in cls.connected_clients:
        connected_client.write_message(message)

@classmethod
def on_change(cls, change):
    message = f"{change['operationType']}: {change['fullDocument']['name']}"
    ChangesHandler.send_updates(message)
```

When you have new data from the change stream, you pass it to the WebSocket server using the `on_change` method. This method formats the change stream data into a string ready to be pushed out to the connected clients. This push occurs in the `send_updates` method. Within this method, you loop over all clients in the `connected_clients` set and use the `write_message` action to write the data to the client's WebSocket connection.

## Monitoring a Change Stream for Changes

In Motor, [MongoDB Collections](https://docs.mongodb.com/manual/core/databases-and-collections/) have a `watch()` method which you can use to monitor the collection for any changes. Then, whenever there is a new change on the stream, you can use the WebSocket server's `on_change` method to proxy the data to the WebSocket clients.

``` python
change_stream = None

async def watch(collection):
    global change_stream

    async with collection.watch(full_document='updateLookup') as change_stream:
        async for change in change_stream:
            ChangesHandler.on_change(change)
```

This `watch` function is attached to your Tornado loop as a callback.

``` python
def main():
    client = MotorClient(os.environ["MONGO_SRV"])
    collection = client["sample_airbnb"]["listingsAndReviews"]

    loop = tornado.ioloop.IOLoop.current()
    loop.add_callback(watch, collection)

    try:
        loop.start()
    except KeyboardInterrupt:
        pass
    finally:
        if change_stream is not None:
            change_stream.close()
```

## Subscribing to Changes in the Browser Via WebSockets

For this example, your WebSocket client is a simple web page, and it logs any WebSocket messages to the browser's JavaScript console.

``` javascript
<html>
    <head>
        <script>
            const websocket = new WebSocket('ws://127.0.0.1:8000/socket')
            websocket.onmessage = function(evt) {
                console.log(evt.data)
            }
        </script>
    </head>
    <body></body>
</html>
```

You can use Tornado to serve this web page as well.

``` python
class WebpageHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("templates/index.html")
```

## Putting it All Together

<img src="https://www.mongodb.com/developer/images/how-to/subscribing-changes-browser-websockets/collection-watch-websocket.gif" class="align-center" alt="Screencast showing change being sent in real-time via a WebSocket" />

To try the example for yourself:

-   clone the example code from [GitHub](https://github.com/aaronbassett/mongodb-changestreams-tornado-example).
-   install the requirements.
-   set the required environmental variables.
-   run the Python script.

``` shell
git clone git@github.com:aaronbassett/mongodb-changestreams-tornado-example.git
cd mongodb-changestreams-tornado-example
pip install -r requirements.txt
export MONGO_SRV=
python server.py
```

Once the WebSocket server is running in your terminal, open a browser window to [localhost:8000](http://127.0.0.1:8000) and view your JavaScript console. Then, make some changes to your Collection via [Compass](https://www.mongodb.com/products/compass) or the [MongoDB Shell](https://docs.mongodb.com/manual/mongo/).

## Wrap-Up

In this article, you have subscribed to all changes on a single collection. However, you can use change streams to subscribe to all data changes on a single collection, a database, or even an entire deployment. And, because change streams use the [aggregation framework](https://docs.mongodb.com/manual/aggregation/), applications can also filter for specific changes or transform the notifications.

For more information, see the [MotorChangeStream documentation](https://motor.readthedocs.io/en/stable/api-tornado/motor_change_stream.html).

Subscribe to MongoDB Change Streams via WebSockets using Python and Tornado.

Subscribe to MongoDB Change Streams Via WebSockets

Mark Smith

As a Developer Advocate for MongoDB, I have quite a few conversations with developers. Many of these developers have never used MongoDB, and so the conversation is often around what kind of data MongoDB is particularly good for. (Spoiler: Nearly *all* of them! MongoDB is a general purpose database that just happens to be centered around documents instead of tables.)

But there are lots of developers out there who already use MongoDB every day, and in those situations, my job is to make sure they know how to use MongoDB effectively. I make sure, first and foremost, that these developers know about MongoDB's [Aggregation Framework](https://developer.mongodb.com/quickstart/introduction-aggregation-framework), which is, in my opinion, MongoDB's most powerful feature. It is relatively underused. If you're not using the Aggregation Framework in your projects, then either your project is very simple, or you could probably be doing things more efficiently by adding some aggregation pipelines.

But this article is not about the Aggregation Framework! This article is about three *other* features of MongoDB that deserve to be better known: [TTL Indexes](#ttl-indexes), [Capped Collections](#capped-collections), and [Change Streams](#change-streams).

## TTL Indexes

One of the great things about MongoDB is that it's so *easy* to store data in it, without having to go through complex steps to map your data to the model expected by your database's schema expectations.

Because of this, it's quite common to use MongoDB as a cache as well as a database, to store things like session information, authentication data for third-party services, and other things that are relatively short-lived.

A common idiom is to store an expiry date in the document, and then when retreiving the document, to compare the expiry date to the current time and only use it if it's still valid. In some cases, as with OAuth access tokens, if the token has expired, a new one can be obtained from the OAuth provider and the document can be updated.

``` 
coll.insert_one(
    {
        "name": "Professor Bagura",
        # This document will disappear before 2022:
        "expires_at": datetime.fromisoformat("2021-12-31 23:59:59"),
    }
)

# Retrieve a valid document by filtering on docs where `expires_at` is in the future:
if (doc := coll.find_one({"expires_at": {"$gt": datetime.now()}})) is None:
    # If no valid documents exist, create one (and probably store it):
    doc = create_document()

# Code to use the retrieved or created document goes here.
print(doc)
```

Another common idiom also involves storing an expiry date in the document, and then running code periodically that either deletes or refreshes expired documents, depending on what's correct for the use-case.

``` python
while True:
    # Delete all documents where `expires_at` is in the past:
    coll.delete_many({"expires_at": {"$lt": datetime.now()}})
    time.sleep(60)
```

An alternative way to manage data that has an expiry, either absolute or relative to the time the document is stored, is to use a [TTL index](https://docs.mongodb.com/manual/core/index-ttl/).

To use the definition from the documentation: "TTL indexes are special single-field indexes that MongoDB can use to automatically remove documents from a collection after a certain amount of time or at a specific clock time." TTL indexes are why I like to think of MongoDB as
a platform for building data applications, not just a database. If you apply a TTL index to your documents' expiry field, MongoDB will automatically remove the document for you! This means that you don't need to write your own code for removing expired documents, and you don't need to remember to always filter documents based on whether their expiry is earlier than the current time. You also don't need to calculate the absolute expiry time if all you have is the number of seconds a document remains valid!

Let me show you how this works. The code below demonstrates how to create an index on the `created_at` field. Because `expiresAfterSeconds` is set to 3600 (which is one hour), any documents in the collection with `created_at` set to a date will be deleted one hour after that point in time.

``` python
coll = db.get_collection("ttl_collection")

# Creates a new index on the `created_at`.
# The document will be deleted when current time reaches one hour (3600 seconds)
# after the date stored in `created_at`:
coll.create_index([("expires_at", 1)], expireAfterSeconds=3600)

coll.insert_one(
    {
        "name": "Professor Bagura",
        "created_at": datetime.now(),  # Document will disappear after one hour.
    }
)
```

Another common idiom is to explicitly set the expiry time, when the document should be deleted. This is done by setting `expireAfterSeconds` to 0:

``` python
coll = db.get_collection("expiry_collection")

# Creates a new index on the `expires_at`.
# The document will be deleted when
# the current time reaches the date stored in `expires_at`:
coll.create_index([("expires_at", 1)], expireAfterSeconds=0)

coll.insert_one(
    {
        "name": "Professor Bagura",
        # This document will disappear before 2022:
        "expires_at": datetime.fromisoformat("2021-12-31 23:59:59"),
    }
)
```

Bear in mind that the background process that removes expired documents only runs every 60 seconds, and on a cluster under heavy load, maybe less frequently than that. So, if you're working with documents with very short-lived expiry durations, then this feature probably isn't for you. An alternative is to continue to filter by the expiry in your code, to benefit from finer-grained control over document validity, but allow the TTL expiry service to maintain the collection over time, removing documents that have very obviously expired.

If you're working with data that has a lifespan, then TTL indexes are a great feature for maintaining the documents in a collection.

## Capped Collections

Capped collections are an interesting feature of MongoDB, useful if you wish to efficiently store a ring buffer of documents.

A capped collection has a maximum size in bytes and optionally a maximum number of documents. (The lower of the two values is used at any time, so if you want to reach the maximum number of documents, make sure you set the byte size large enough to handle the number of documents you wish to store.) Documents are stored in insertion order, without the need for a specific index to maintain that order, and so can handle higher throughput than an indexed collection. When either the collection reaches the set byte `size`, or the `max` number of documents, then the oldest documents in the collection are purged.

Capped collections can be useful for buffering recent operations (application-level operations - MongoDB's oplog is a different kind ofthing), and these can be queried when an error state occurs, in order to have a log of recent operations leading up to the error state.

Or, if you just wish to efficiently store a fixed number of documents in insertion order, then capped collections are the way to go.

Capped collections are created with the [createCollection](https://docs.mongodb.com/manual/reference/method/db.createCollection/#create-a-capped-collection) method, by setting the `capped`, `size`, and optionally the `max` parameters:

``` python
# Create acollection with a large size value that will store a max of 3 docs:
coll = db.create_collection("capped", capped=True, size=1000000, max=3)

# Insert 3 docs:
coll.insert_many([{"name": "Chico"}, {"name": "Harpo"}, {"name": "Groucho"}])

# Insert a fourth doc! This will evict the oldest document to make space (Zeppo):
coll.insert_one({"name": "Zeppo"})

# Print out the docs in the collection:
for doc in coll.find():
    print(doc)

# {'_id': ObjectId('600e8fcf36b07f77b6bc8ecf'), 'name': 'Harpo'}
# {'_id': ObjectId('600e8fcf36b07f77b6bc8ed0'), 'name': 'Groucho'}
# {'_id': ObjectId('600e8fcf36b07f77b6bc8ed1'), 'name': 'Zeppo'}
```

If you want a rough idea of how big your bson documents are in bytes, for calculating the value of `size`, you can either use your driver's [bsonSize](https://docs.mongodb.com/manual/reference/mongo-shell/#miscellaneous) method in the `mongo` shell, on a document constructed in code, or you can use MongoDB 4.4's new [bsonSize](https://docs.mongodb.com/master/reference/operator/aggregation/bsonSize/) aggregation operator, on documents already stored in MongoDB.

Note that with the improved efficiency that comes with capped collections, there are also some limitations. It is not possible to explicitly delete a document from a capped collection, although documents will eventually be replaced by newly inserted documents. Updates in a capped collection also cannot change a document's size. You can't shard a capped collection. There are some other limitations around replacing and updating documents and transactions. Read the [documentation](https://docs.mongodb.com/manual/core/capped-collections/#restrictions-and-recommendations) for more details.

It's worth noting that this pattern is similar in feel to the [Bucket Pattern](https://www.mongodb.com/blog/post/building-with-patterns-the-bucket-pattern), which allows you to store a capped number of items in an array, and automatically creates a new document for storing subsequent values when that cap is reached.

## Change Streams and the `watch` method

And finally, the biggest lesser-known feature of them all! [Change streams](https://docs.mongodb.com/manual/changeStreams/) are a live stream of changes to your database. The `watch` method, implemented in most MongoDB drivers, streams the changes made to [a collection](https://pymongo.readthedocs.io/en/stable/api/pymongo/collection.html#pymongo.collection.Collection.watch),
[a database](https://pymongo.readthedocs.io/en/stable/api/pymongo/database.html#pymongo.database.Database.watch), or even your [entire MongoDB replicaset or cluster](https://pymongo.readthedocs.io/en/stable/api/pymongo/mongo_client.html#pymongo.mongo_client.MongoClient.watch), to your application in real-time. I'm always surprised by how few people have not heard of it, given that it's one of the first MongoDB features that really excited me. Perhaps it's just luck that I stumbled across it earlier.

In Python, if I wanted to print all of the changes to a collection as they're made, the code would look a bit like this:

``` python
with my_database.my_collection.watch() as stream:
    for change in stream:
        print(change)
```

In this case, `watch` returns an iterator which blocks until a change is made to the collection, at which point it will yield a BSON document describing the change that was made.

You can also filter the types of events that will be sent to the change stream, so if you're only interested in insertions or deletions, then those are the only events you'll receive.

I've used change streams (which is what the `watch` method returns) to implement a chat app, where changes to a collection which represented a conversation were streamed to the browser using WebSockets.

But fundamentally, change streams allow you to implement the equivalent of a database trigger, but in your favourite programming language, using all the libraries you prefer, running on the servers you specify. It's a super-powerful feature and deserves to be better known.

## Further Resources

If you don't already use the [Aggregation Framework](https://developer.mongodb.com/quickstart/introduction-aggregation-framework), definitely check out the documentation on that. It'll blow your mind (in a good way)!

Further documentation on the topics discussed here:

-   [TTL Index Documentation](https://docs.mongodb.com/manual/core/index-ttl/),
-   [Capped Collection Documentation](https://docs.mongodb.com/manual/core/capped-collections/)
-   [Change Streams](https://docs.mongodb.com/manual/changeStreams/)

>If you have questions, please head to our [developer community website](https://community.mongodb.com/) where the MongoDB engineers and the MongoDB community will help you build your next big idea with MongoDB.


Change Streams

Featured

MongoDB Change Streams with Python

Change Streams & Triggers with Node.js Tutorial

Three Underused MongoDB Features

All Change Streams Content