Using Socket.io with RethinkDB Changefeeds to Build a Reactive JavaScript Stack


RethinkDB is the latest major entrant NoSQL database. Having used it in production to build my current startup, it is in my opinion one of the best databases for a reactive web app. The key feature that makes it so powerful are changefeeds. These are essentially defined ReQL queries tied to an event listener (callbacks). Every time a document is changed, added, or removed that is selected by the query, the changefeed passes the updated document(s) to the event listener. If you're anything like me, I'm sure your mind is already spinning with the possibilities!

Meteor made use of MongoDB oplog tailing to push database changes up to the client (i.e. the client app did not need to regularly poll the backend for updates). Building on this idea, this is how I've used RethinkDB changefeeds to push database changes to the client of a WebSocket, thus, keeping client and server data in seamless sync. Let's see how this works.

Notes:

  • The only language used in this tip is JavaScript (node.js)
  • Assuming you already have a working Node app, with the RethinkDB module installed, and connected to a running RethinkDB database. We'll reference 'connection' as the pointer to the RethinkDB connection, and 'r' as the RethinkDB JS library
  • No Client side code is covered, so you could use React, Angular, or any framework that has a client side immutable data store with getters and setters (i.e. this.setState({}) and this.state in React.js)
  • This post assumes you have some knowledge of Socket.io, and have a socket setup as 'socket'

Let's say we have a table in RethinkDB called Fruit, with the documents:

id type
e73c1635-88a9-4d27-8c2a-ab447449bb4b Apple
86fc5fd7-0ce0-42a5-b9cc-44d2e34c027d Orange

We run the following query to get all of the fruit:


r.table('Fruit').run(connection, loadCallback);

Now, we want to write a callback to emit an initial array of all of the fruit to the client


var loadCallback = function(err, cursor) {
     if (err) // throw something

     // returns an array of all documents (fruit in this case) in the cursor
     cursor.toArray(function(err, fruit) {

          if (err) // throw here too
          socket.emit('load_fruit', fruit);
     }
};

Of course, one more layer and we'll officially be in callback hell. So by all means, use promises here.

But now we have an array of all of the fruit emited to the client (or whoever happens to be listening to 'load_fruit' on this socket). This is where we would write a listener in our client data store to set the initial value.

And now for changefeeds!

We make one additional method call to the above query to get the changefeed for a table:


r.table('Fruit').changes().run(connection, updateCallback);


And finally, to push updates to our fruit table to any socket listeners, our update callback is as follows:


var updateCallback = function(err, cursor) {
     if (err) // throw something

     cursor.each(function(err, fruit) {
          if (err) // throw here too

         // handle a new fruit
         if (fruit && fruit.new_val && fruit.old_val === undefined) {
              // Here, the listener in the client side data
              // store should push this fruit onto the fruit array
              socket.emit('new_fruit', fruit.new_val);
         }

         // handle update fruit
         else if (fruit && fruit.new_val && fruit.old_val) {
              // client side listener should look up fruit by id
              // and replace with this new fruit value
              socket.emit('update_fruit', fruit.new_val);

         // handle deleted fruit
         else if (fruit && fruit.new_val === null) {
              // client side listener should look up fruit by id
              // and delete it
              // *note* we still pass the old value of the fruit
              // so that we have it's id to look it up in the client
              socket.emit('delete_fruit', fruit.old_val);
         }
     }
};

Now, we can add a new Fruit, say a type = 'Banana', and a new_fruit event will be emitted. If we change the Apple into a Pear, an update_fruit event will be emitted. Finally, if we delete the Orange, a delete_fruit event will be emitted with the old values (including the id).

And that's the basics of combining RethinkDB changefeeds with Socket.io, to keep client side data stores in sync with the backend!