Using MongoDB as a realtime database with change streams

This article was originally published on Pusher’s blog .

Pusher, our weekly sponsor, makes communication and collaboration APIs that power apps all over the world, supported by easy to integrate SDKs for web, mobile, as well as most popular backend stacks. Get started.

Getting data changes from a database in realtime is not as easy as you may think.

In a previous tutorial, I mentioned there are three main approaches to do this:

• Poll the database every X seconds and determine if something has changed using a timestamp, version number or status field.
• Use database or application-level triggers to execute a piece of code when something changes.
• Use the database transaction/replication log, which records every change to the database.

However, in MongoDB, change streams allows you to listen for changes in collections without any complexity.

Change streams are available since MongoDB 3.6 and they work by reading the oplog, a capped collection where all the changes to the data are written and functions as the database replication log.

In this tutorial, you’re going to learn how to stream, in realtime, the changes made to a collection in a MongoDB database to a React app using a Node.js server.

The application that you’ll be building allows you to add and delete tasks. It looks like this:

Under the hood, it communicates to an API implemented in Node.js that saves the changes to a database. The Node.js script also receives these changes using change streams, parsing them and publishing them to a Pusher channel so the React application can consume them.

Here’s the diagram that describes the above process:

Of course, a scenario where multiple applications are writing to the same database could be more realistic, but for learning purposes, I’ll use a simple application.

In addition, you’ll see how a solution like this one, could be a good alternative to the realtime database capabilities of Firebase.

Prerequisites

Here’s what you need to have installed to follow this tutorial:

• MongoDB (version 3.6 or superior)
• Node.js (6 or superior)
• Optionally, a JavaScript editor.

You’ll need to have knowledge of:

• JavaScript (intermediate level), in particular, Node.js and React.
• Basic MongoDB management tasks

For reference, here is a GitHub repository with all the code shown in this tutorial and instructions to run it.

Now let’s start by creating a Pusher application.

Creating a Pusher application

If you haven’t already, create a free account at Pusher.

Then, go to your dashboard and create a Channels app, choosing a name, the cluster closest to your location, and optionally, React as the frontend tech and Node.js as the backend tech:

This will give you some sample code to get started:

Save your app id, key, secret and cluster values. We’ll need them later.

Configuring MongoDB

Since change streams use MongoDB’s operations log, and the oplog is used to support the replication features of this database, you can only use change streams with replica sets or sharded clusters.

It’s easier to use replica sets, so let’s go that way.

A replica set is a group of mongod processes that maintain the same data set. However, you can create a replica set with only one server, just execute this command:

mongod --replSet "rs"

Remember that if you do not use the default data directory (/data/db or c:\data\db), specify the path to the data directory using the --dbpath option:

mongod --dbpath <DATA_PATH> --replSet "rs"

Next, in a separate terminal window, run mongo, the MongoDB client.

If this is the first time you create a replica set, execute rs.initiate():

eh@eh:~/Documents/mongodb-linux-x86_64-3.6.4$ bin/mongoMongoDB shell version v3.6.4connecting to: mongodb://127.0.0.1:27017MongoDB server version: 3.6.4...> rs.initiate(){    "info2" : "no configuration specified. Using a default configuration for the set",    "me" : "localhost:27017",    "ok" : 1,    "operationTime" : Timestamp(1527258648, 1),    "$clusterTime" : {        "clusterTime" : Timestamp(1527258648, 1),        "signature" : {            "hash" : BinData(0,"AAAAAAAAAAAAAAAAAAAAAAAAAAA="),            "keyId" : NumberLong(0)        }    }}rs:OTHER>

The application is going to watch the collection tasks in a database called tasksDb.

Usually, the database and the collection are created by the MongoDB driver when the application performs the first operation upon them, but for change streams, they must exist before opening the stream.

So while you are at mongo , create the database and the collection with the commands use and db.createCollection, like this:

rs:OTHER> use tasksDbswitched to db tasksDbrs:OTHER> db.createCollection('tasks'){    "ok" : 1,    "operationTime" : Timestamp(1527266976, 1),    "$clusterTime" : {        "clusterTime" : Timestamp(1527266976, 1),        "signature" : {          "hash" : BinData(0,"AAAAAAAAAAAAAAAAAAAAAAAAAAA="),          "keyId" : NumberLong(0)        }    }}rs:OTHER>

Now you're ready to start building the application.

Let’s start with the Node.js server.

Building the Node.js server

Create a new directory and in a terminal window, inside that directory, initialize a Node.js project with the command:

npm init -y

Next, install the dependencies the application is going to use with:

npm install --save body-parser express mongoose pusher

• body-parser is a middleware for parsing the body of the request.
• express to create the web server for the REST API that the React app is going to use.
• mongoose is a schema-based library for working with MongoDB.
• pusher to publish the database changes in realtime.

Now the first thing we’re going to do is create a schema for the task collection. Create the file models/task.jsand copy the following code:

const mongoose = require('mongoose');  const Schema   = mongoose.Schema;const taskSchema = new Schema({   task: { type: String },});module.exports = mongoose.model('Task', taskSchema);

As you can see, the collection is only going to store the task as text.

Next, create the file routes/api.js and require the task schema and Express to create a router:

const Task  = require('../models/task');const express = require('express');const router = express.Router();

Create a POST endpoint with the /new path to save task:

router.post('/new', (req, res) => {    Task.create({        task: req.body.task,    }, (err, task) => {        if (err) {          console.log('CREATE Error: ' + err);          res.status(500).send('Error');        } else {          res.status(200).json(task);        }    });});

And another one to delete tasks, passing the ID of the task using a DELETE method:

router.route('/:id')    /* DELETE */    .delete((req, res) => {        Task.findById(req.params.id, (err, task) => {          if (err) {             console.log('DELETE Error: ' + err);            res.status(500).send('Error');          } else if (task) {            task.remove( () => {              res.status(200).json(task);            });          } else {            res.status(404).send('Not found');          }        });    });module.exports = router;

Now, in the root directory, create the file server.js and require the following modules:

const express = require('express');const bodyParser = require('body-parser');const mongoose = require('mongoose');const api = require('./routes/api');const Pusher = require('pusher');

Configure the Pusher object entering your app information:

const pusher = new Pusher({  appId      : '<INSERT_APP_ID>',  key        : '<INSERT_APP_KEY>',  secret     : '<INSERT_APP_SECRET>',  cluster    : '<INSERT_APP_CLUSTER>',  encrypted  : true,});const channel = 'tasks';

And configure an Express server with CORS headers (because the React app is going to be published in a different port), JSON requests, and /api as the path:

const app = express();app.use((req, res, next) => {  res.header("Access-Control-Allow-Origin", "*");  res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");  res.header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS");  next();});app.use(bodyParser.json());app.use(bodyParser.urlencoded({ extended: true }));app.use('/api', api);

This way, you can connect to the database passing the name of the replica set you configured before:

mongoose.connect('mongodb://localhost/tasksDb?replicaSet=rs');

And set two callbacks, one for connections errors and another one if the connection is successful:

const db = mongoose.connection;db.on('error', console.error.bind(console, 'Connection Error:'));db.once('open', () => {});

If the connection is successful, let’s start listening for connections on port 9000 and watch for changes on the tasks collection:

db.once('open', () => {  app.listen(9000, () => {    console.log('Node server running on port 9000');  });  const taskCollection = db.collection('tasks');  const changeStream = taskCollection.watch();  changeStream.on('change', (change) => {  });});

Here comes the interesting part.

When there’s a change in the collection, a change event is received. In particular, the following changes are supported:

• Insert
• Update
• Replace
• Delete
• Invalidate

Here’s an example of an insert event:

{  _id:   {    _data: Binary {      _bsontype: 'Binary',      sub_type: 0,      position: 49,      buffer: <Buffer 82 5b 08 8a 2a 00 00 00 01 46 64 5f 69 64 00 64 5b 08 8a 2a 99 a1 c5 0d 65 f4 c4 4f 00 5a 10 04 13 79 9a 22 35 5b 45 76 ba 45 6a f0 69 81 60 af 04>     }  },  operationType: 'insert',  fullDocument: {     _id: 5b088a2a99a1c50d65f4c44f,     task: 'my task',     __v: 0  },  ns: { db: 'tasksDb', coll: 'tasks' },  documentKey: { _id: 5b088a2a99a1c50d65f4c44f } }

You can use the _id property to resume a change stream, in other words, to start receiving events from the operation represented by that property.

Here’s an example of a delete event:

{   _id:   {     _data: Binary {      _bsontype: 'Binary',      sub_type: 0,      position: 49,      buffer: <Buffer 82 5b 08 8b f6 00 00 00 01 46 64 5f 69 64 00 64 5b 08 8a 2a 99 a1 c5 0d 65 f4 c4 4f 00 5a 10 04 13 79 9a 22 35 5b 45 76 ba 45 6a f0 69 81 60 af 04>     }  },  operationType: 'delete',  ns: { db: 'tasksDb', coll: 'tasks' },  documentKey: { _id: 5b088a2a99a1c50d65f4c44f }}

Notice that in this case, the deleted object is not returned, just its ID in the documentKey property.

You can learn more about these change events here.

With this information, back to server.js, you can extract the relevant data from the object and publish it to Pusher in the following way:

changeStream.on('change', (change) => {    console.log(change);

    if(change.operationType === 'insert') {      const task = change.fullDocument;      pusher.trigger(        channel,        'inserted',         {          id: task._id,          task: task.task,        }      );     } else if(change.operationType === 'delete') {      pusher.trigger(        channel,        'deleted',         change.documentKey._id      );    }});

And that’s the code for the server. Now let’s build the React app.

Building the React app

Let’s use create-react-app to bootstrap a React app.

In another directory, execute the following command in a terminal window to create a new app:

npx create-react-app my-app

Now go into the app directory and install all the Pusher dependency with npm:

cd my-appnpm install --save pusher-js

Open the file src/App.css and replace its content with the following CSS styles:

*{  box-sizing: border-box;}

body {  font-size: 15px;  font-family: 'Open Sans', sans-serif;  color: #444;  background-color: #300d4f;  padding: 50px 20px;  margin: 0;  min-height: 100vh;  position: relative;}

.todo-wrapper {  width: 400px;  max-width: 100%;  min-height: 500px;  margin: 20px auto 40px;  border: 1px solid #eee;  border-radius: 4px;  padding: 40px 20px;  -webkit-box-shadow: 0 0 15px 0 rgba(0,0,0,0.05);  box-shadow: 0 0 15px 0 rgba(0,0,0,0.05);  background-color: #e9edf6;  overflow: hidden;  position: relative;}

form {  overflow: overlay;}

.btn, input {  line-height: 2em;  border-radius: 3px;  border: 0;  display: inline-block;  margin: 15px 0;  padding: 0.2em 1em;  font-size: 1em;}

input[type='text'] {  border: 1px solid #ddd;   min-width: 80%;}

input:focus {  outline: none;  border: 1px solid #a3b1ff;}

.btn {  text-align: center;  font-weight: bold;   cursor: pointer;  border-width: 1px;  border-style: solid;}

.btn-add {  background: #00de72;  color: #fefefe;  min-width: 17%;  font-size: 2.2em;  line-height: 0.5em;  padding: 0.3em 0.3em;  float: right;}

ul {  list-style: none;  padding: 0;}

li {    display: flex;    justify-content: space-between;    align-items: center;    margin-bottom: 5px;    background-color: #dee2eb;}

.text {  padding: 0.7em;}

.delete {  padding: 0.3em 0.7em;  min-width: 17%;  background: #f56468;  color: white;  font-weight: bold;  cursor: pointer;  font-size: 2.2em;  line-height: 0.5em;}

Next, open the file src/App.js and at the top, import the Pusher library:

import Pusher from 'pusher-js';

Define a constant for the API URL:

const API_URL = 'http://localhost:9000/api/';

In the constructor of the class, define an array for the tasks and a property for the text of a task as the state, and bind the methods to update the text and add and delete tasks:

class App extends Component {    constructor(props) {    super(props);    this.state = {      tasks: [],      task: ''    };    this.updateText = this.updateText.bind(this);    this.postTask = this.postTask.bind(this);    this.deleteTask = this.deleteTask.bind(this);    this.addTask = this.addTask.bind(this);    this.removeTask = this.removeTask.bind(this);  }  ...}

Let’s review each method. Add them after the constructor, before the render() method.

The updateText method will update the state every time the input text for the task changes:

updateText(e) {    this.setState({ task: e.target.value });}

The postTask method will post to task entered by the user to the API:

postTask(e) {    e.preventDefault();    if (!this.state.task.length) {      return;    }    const newTask = {      task: this.state.task    };    fetch(API_URL + 'new', {      method: 'post',      headers: {        'Content-Type': 'application/json'      },      body: JSON.stringify(newTask)    }).then(console.log);}

And the method deleteTask will call the API to delete a task using its ID:

deleteTask(id) {    fetch(API_URL + id, {      method: 'delete'    }).then(console.log);}

On the other hand, you’ll also need methods to add and delete a task from the state so the changes can be reflected in the UI. That’s the job of the methods addTask and removeTask:

addTask(newTask) {    this.setState(prevState => ({      tasks: prevState.tasks.concat(newTask),      task: ''    }));}

removeTask(id) {    this.setState(prevState => ({      tasks: prevState.tasks.filter(el => el.id !== id)    }));}

The app will call these methods when the corresponding event from Pusher is received.

You can set up Pusher and bind these methods to the inserted and deleted events in the method componentDidMount, entering your Pusher app key and cluster:

componentDidMount() { this.pusher = new Pusher('<INSERT_APP_KEY>', {     cluster: '<INSERT_APP_CLUSTER>',     encrypted: true, }); this.channel = this.pusher.subscribe('tasks');

 this.channel.bind('inserted', this.addTask); this.channel.bind('deleted', this.removeTask);}

This way, the render method just renders the tasks from the state using a Task component and a form to enter new tasks.

Replace the render() method with the following:

render() {    let tasks = this.state.tasks.map(item =>        <Task key={item.id} task={item} onTaskClick={this.deleteTask} />    );

    return (        <div className="todo-wrapper">            <form>              <input type="text" className="input-todo" placeholder="New task" onChange={this.updateText} value={this.state.task} />              <div className="btn btn-add" onClick={this.postTask}>+</div>            </form>

            <ul>              {tasks}            </ul>        </div>    );}

And the code of the Task component (which you can place after the App class):

class Task extends Component {    constructor(props) {      super(props);      this._onClick = this._onClick.bind(this);    }    _onClick() {      this.props.onTaskClick(this.props.task.id);    }    render() {      return (        <li key={this.props.task.id}>          <div className="text">{this.props.task.task}</div>          <div className="delete" onClick={this._onClick}>-</div>        </li>      );    }}

And that’s it. Let’s test the complete application.

Testing the application

Make sure the MongoDB database is running with the replica set configured on the server:

mongod --dbpath <DATA_PATH> --replSet "rs"

In a terminal window, go to the directory where the Node.js server resides and execute:

node server.js

For the React app, inside the app directory, execute:

npm start

A browser window will open http://localhost:3000/, and from there, you can start entering and deleting tasks:

You can also see in the output of the Node.js server how change events are received from MongoDB:

Or on Pusher’s dashboard, select your app, and in the Debug section, you’ll see how the messages are received:

Conclusion

In this tutorial, you have learned how to persist data in MongoDB and propagate the changes in realtime using change streams and Pusher channels

This is equivalent to the functionality provided by Firebase and its realtime database. The advantage is that a solution like the one presented in this tutorial is more flexible and gives you more control.

From here, the application can be extended in many ways, for example:

• Support for more collections
• Implement an update functionality for the tasks (for example, the status) and replicate this event.
• Use the resume token to receiving the events from the last one registered, after a connection failure.

Remember that in this GitHub repository you can find the code of the Node.js server and the React app.

For more information about change streams, here are some good resources:

• Using Change Streams to Keep Up with Your Data
• An Introduction to Change Streams
• MongoDB 3.6 change streams example with Node.js
• MongoDB Data Change
• MongoDB manual: Change Streams

Pusher, our weekly sponsor, makes communication and collaboration APIs that power apps all over the world, supported by easy to integrate SDKs for web, mobile, as well as most popular backend stacks. Get started.