Tutorial: How to use MongoDB Change Streams [Part 2]
Tags: mongodb, database, go, tutorial
Published: 2020-07-09 15:09:09

This is the second part of a blog series that covers MongoDB Change Streams and how it can be used with `Azure Cosmos DB` which [has wire protocol support for MongoDB server version 3.6](https://docs.microsoft.com/azure/cosmos-db/mongodb-feature-support-36?WT.mc_id=devto-blog-abhishgu) (including the [Change Streams](https://docs.mongodb.com/manual/changeStreams/) feature). Part 1 covered the introduction, overview of the Change streams processor service and walked you through how to run the application so that you can witness Changes streams in action.

In this instalment, we will go over the code and see how things work behind the scenes.

> As always, the code is [available on GitHub](https://github.com/abhirockzz/mongodb-changestreams-processor)

## Overview: Change processor app

Before moving on, here is a quick refresher of the Change Processor service (copied over from part 1):

The application is a change processor service that uses the Change stream feature. It's a [`Go`](https://golang.org/) application that uses the [official MongoDB Go driver](https://godoc.org/go.mongodb.org/mongo-driver/mongo) but the concepts should be applicable to any other language whose native driver supports Change Streams. 

It uses the [`Watch` API](https://godoc.org/go.mongodb.org/mongo-driver/mongo#Collection.Watch) to subscribe to the change events feed in a specific `Collection` so that it is notified of documents being created, updated and deleted. It extracts the relevant information from the change event payload i.e. the document which was affected and saves it locally to a file. It also demonstrates how to use `Resume Tokens` to save processing progress.


## Code walkthrough

![](https://media.giphy.com/media/ZVik7pBtu9dNS/giphy.gif)

Here is the project layout: the `main.go` file consists of most of the logic around subscribing and processing change streams while `token/resume_token.go` handles saving/retrieving resume tokens.

```
.
├── go.mod
├── go.sum
├── main.go
└── token
    └── resume_token.go
```

The application starts by connecting to the Azure Cosmos DB MongoDB API and bails out if that fails for some reason. If the connectivity succeeds, we get a handle to the MongoDB Collection we want to watch

```
	client, err := mongo.NewClient(options.Client().ApplyURI(mongoURI))
	if err != nil {
		log.Fatal("failed to create client: ", err)
	}

	ctx, cancel := context.WithCancel(context.Background())

	err = client.Connect(ctx)
	if err != nil {
		log.Fatal("failed to connect", err)
	}
  
  coll := client.Database(mongoDBName).Collection(mongoCollectionName)
	defer func() {
		err = client.Disconnect(context.Background())
		if err != nil {
			fmt.Println("failed to close connection")
		}
	}()
```

It's time to start "tracking" the collection for change events. Notice the `pipeline` and `opts` arguments

```
cs, err := coll.Watch(ctx, pipeline, opts)
```

`pipeline` is a [`*mongo.Pipeline`](https://godoc.org/go.mongodb.org/mongo-driver/mongo#Pipeline). We specify a couple of stages as a part of the `Pipeline` - `$match` and `$project` along with the `fullDocument` option

> At the time of writing, [these options are mandatory](https://docs.microsoft.com/azure/cosmos-db/mongodb-change-streams?WT.mc_id=devto-blog-abhishgu#current-limitations) due to constraints specific to Azure Cosmos DB. 

This is how these are manifested in the code:

```
	matchStage := bson.D{{"$match", bson.D{{"operationType", bson.D{{"$in", bson.A{"insert", "update", "replace"}}}}}}}
	//matchStage := bson.D{{"$match", bson.D{{"operationType", "insert"}}}}
	projectStage := bson.D{{"$project", bson.M{"_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1}}}
	pipeline := mongo.Pipeline{matchStage, projectStage}
	opts := options.ChangeStream().SetFullDocument(options.UpdateLookup)
```

You have a choice of using the resume token (specified by an environment variable - more on this in the next section). The resume token is set as an [`*options.ChangeStreamOptions`](https://godoc.org/go.mongodb.org/mongo-driver/mongo/options#ChangeStreamOptions) using the `ResumeAfter` field. The resume token (if present) is stored locally in a file named `token` - we check for the token before starting our change stream and use it if it's present

```
	if resumeSupported {
		t, err := token.RetrieveToken()
		if err != nil {
			log.Fatal("failed to fetch resume token: ", err)
		}
		if t != nil {
			opts.SetResumeAfter(t)
		}
	}
```

As mentioned earlier, token operations (save and retrieve) are part of `token/resume_token.go`. Here is how an existing token is fetched:

```
func RetrieveToken() (bson.Raw, error) {
	tf, err := os.Open(tokenFileName)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, nil
		}
		return nil, err
	}
	token, err := bson.NewFromIOReader(tf)
	if err != nil {
		return nil, err
	}
	return token, nil
}
```

We create the output file to store the change events and start watching the collection

```
  op, err := os.OpenFile(outputfileName,
		os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
  ...

	cs, err := coll.Watch(ctx, pipeline, opts)
	if err != nil {
		log.Fatal("failed to start change stream watch: ", err)
	}

```

A different goroutine is used to process the change events 

```
	go func() {
		fmt.Println("started change stream...")
		for cs.Next(ctx) {
			re := cs.Current.Index(1)
			_, err := op.WriteString(re.Value().String() + "\n")
			if err != nil {
				fmt.Println("failed to save change event", err)
			}
		}
	}()
```

The goroutine is the heart of this (rather simple) processing service and runs as a tight `for` loop which uses [`Next`](https://godoc.org/go.mongodb.org/mongo-driver/mongo#ChangeStream.Next) to fetch the change event. Since `Next` is a blocking call and we need to think about clean/graceful exit mechanism. The `ctx` argument which is passed into `Next` plays an important role - it is a [`cancellable context`](https://godoc.org/context#WithCancel) (more in how this is used in just a bit)

```
ctx, cancel := context.WithCancel(context.Background())
```

An `exit` channel is used to detect program termination (e.g. `ctrl+c`)

```
exit := make(chan os.Signal)
signal.Notify(exit, syscall.SIGINT, syscall.SIGTERM)
```

While the change stream detection and processing is running in a separate goroutine, the `main` goroutine is blocked using the `exit` channel

```
	<-exit
	fmt.Println("exit signalled. cancelling context")
	cancel()
	if resumeSupported {
		token.SaveToken(cs.ResumeToken())
	}
```


It does a couple of things:

Invokes `cancel()` which is nothing but the [`Cancel function`](https://godoc.org/context#CancelFunc) returned as a by-product of creating the `cancellable` Context (which is passed into `Next`). Calling the `CancelFunc` propagates program termination to the change stream for loop, allowing it to exit.

> Please note that I have used user-initiated (manual) program termination (e.g. pressing ctrl+c) as an example. The same concept applies if the processor service was executing as a long-running server component

Another key bit is to save the Resume token to the local file:

```
if resumeSupported {
  token.SaveToken(cs.ResumeToken())
}
```

Here is how the resume token is saved to a local file:

```
func SaveToken(token []byte) {

	if len(token) == 0 {
		return
	}
	tf, err := os.Create(tokenFileName)

	if err != nil {
		fmt.Println("token file creation failed", err)
		return
	}
	_, err = tf.Write(token)
	if err != nil {
		fmt.Println("failed to save token", err)
		return
	}
	fmt.Println("saved token to file")
}
```

That concludes the two-part series on MongoDB change streams. I hope this was useful and helped understand the feature with the help of a practical example!