Posted onSep 21, 2020

Realtime Collaborative Drawing (part 4): Redis PubSub + WebRTC Signaling

#javascript #tutorial #webdev #node

In any system that involves real-time communication, open connections, and messages that need to be routed among peers - you tend to run into the problem that not all of your connections will be able to run on a single server. What we need to do, instead, is setup a system that can route messages to any number of servers maintaining any number of connections.

In a previous article, ourdrawing program was recently refactored to useserver sent events by maintaining a connection open. However, if we introduced another web server to create some load balancing, we run into the problem that the client connections may not be accessible across servers.

We can solve this by having a shared communication server/cluster that can handle routing all these messages for us. To do this we will be using thepublisher-subscriber pattern and we will leverageredis to get this job done.

Redis

Redis is an open source (BSD licensed), in-memory data structure store, used as a database, cache and message broker. It supports data structures such as strings, hashes, lists, sets, sorted sets with range queries, bitmaps, hyperloglogs, geospatial indexes with radius queries and streams. Redis has built-in replication, Lua scripting, LRU eviction, transactions and different levels of on-disk persistence, and provides high availability via Redis Sentinel and automatic partitioning with Redis Cluster

Redis is an incredible project, it'sextremely fast and uses minimal cpu. It is a work of art that the project has also remained backwards compatible since version 1. The maintainerantirez (who recently has decided to move on) has created this project over a number of years and built it into something truly incredible. Redis supports just about all the data structures as first-class features and operations.

And it even supportscluster. You can even use it as alast-write-wins CRDT usingroshi. At my work, we've used just about all of these features from queues, hyperloglogs, sorted sets, caching. In a previous project, I once used redis to build aclick stream system using an sort ofevent sourcing model.

Redis PubSub

We're going to use a small feature of redis calledpubsub to route our messages between connections on our server. Assuming you have a redis-server setup. You'll need to addredis as a dependency to our drawing app.

npm install --save redis bluebird

We're going to usebluebird to be able topromisifyAll to all the redis client functions. This will help us write our code withasync/await instead of a number of callbacks.

/connect to SSE and Subscribe to Redis

Recall that our express server was simply keeping an in-memory cache of both connections and channels. We're going to first update our/connect function to insteadsubscribe to messages received from a redispubsub client. To do this, we'll update the client creation code and add aredis.createClient. Then subscribe to messages received to our particular client id viaredis.subscribe('messages:' + client.id). Whenever we receive messages viaredis.on('message', (channel, message) => ...) we can simply emit them back to the server sent event stream.

varredis=require('redis');varbluebird=require('bluebird');bluebird.promisifyAll(redis);app.get('/connect',auth,(req,res)=>{if(req.headers.accept!=='text/event-stream'){returnres.sendStatus(404);}// write the event stream headersres.setHeader('Cache-Control','no-cache');res.setHeader('Content-Type','text/event-stream');res.setHeader("Access-Control-Allow-Origin","*");res.flushHeaders();// setup a clientletclient={id:req.user.id,user:req.user,redis:redis.createClient(),emit:(event,data)=>{res.write(`id:${uuid.v4()}\n`);res.write(`event:${event}\n`);res.write(`data:${JSON.stringify(data)}\n\n`);}};// cache the current connection until it disconnectsclients[client.id]=client;// subscribe to redis events for userclient.redis.on('message',(channel,message)=>{letmsg=JSON.parse(message);client.emit(msg.event,msg.data);});client.redis.subscribe(`messages:${client.id}`);// emit the connected stateclient.emit('connected',{user:req.user});// ping to the client every so oftensetInterval(()=>{client.emit('ping');},10000);req.on('close',()=>{disconnected(client);});});

Also notice that I've added an interval toping the client once every 10 seconds or so. This may not be completely necessary, but I add it to make sure our connection state doesn't inadvertently get cut off for whatever reason.

Peer Join, Peer Signaling

The only other functions we need to change arewhen a peer joins the room,when a peer is sending a signal message to another peer, andwhen a peer disconnects from the server. The other functions likeauth,:roomId remain the same. Let's update the join function below. Note, we will need to keep track of a redis client that the server for general purpose redis communication.

constredisClient=redis.createClient();app.post('/:roomId/join',auth,async(req,res)=>{letroomId=req.params.roomId;awaitredisClient.saddAsync(`${req.user.id}:channels`,roomId);letpeerIds=awaitredisClient.smembersAsync(`channels:${roomId}`);peerIds.forEach(peerId=>{redisClient.publish(`messages:${peerId}`,JSON.stringify({event:'add-peer',data:{peer:req.user,roomId,offer:false}}));redisClient.publish(`messages:${req.user.id}`,JSON.stringify({event:'add-peer',data:{peer:{id:peerId},roomId,offer:true}}));});awaitredisClient.saddAsync(`channels:${roomId}`,req.user.id);returnres.sendStatus(200);});

In order to keep track of who is in a particularroomId, we will make use ofredis sets and add the room id to the current user's set of channels. Following this, we lookup what members are in thechannels:{roomId} and iterate over the peer ids. For each peer id, we effectively will be routing a message to that peer that the current user has joined, and we will route the peer id to therequest.user. Finally, we add ourrequest.user to thechannels:{roomId} set in redis.

Next, let's update the relay code. This will be even simpler since all we have to do is justpublish the message to that peer id.

app.post('/relay/:peerId/:event',auth,(req,res)=>{letpeerId=req.params.peerId;letmsg={event:req.params.event,data:{peer:req.user,data:req.body}};redisClient.publish(`messages:${peerId}`,JSON.stringify(msg));returnres.sendStatus(200);});

Disconnect

Disconnect is a bit more involved, since we have toclean up the rooms that the user is in, then iterate over those rooms toget the list of peers in those rooms, then we mustsignal to each peer in those rooms that the peer has disconnected.

asyncfunctiondisconnected(client){deleteclients[client.id];awaitredisClient.delAsync(`messages:${client.id}`);letroomIds=awaitredisClient.smembersAsync(`${client.id}:channels`);awaitredisClient.delAsync(`${client.id}:channels`);awaitPromise.all(roomIds.map(asyncroomId=>{awaitredisClient.sremAsync(`channels:${roomId}`,client.id);letpeerIds=awaitredisClient.smembersAsync(`channels:${roomId}`);letmsg=JSON.stringify({event:'remove-peer',data:{peer:client.user,roomId:roomId}});awaitPromise.all(peerIds.forEach(asyncpeerId=>{if(peerId!==client.id){awaitredisClient.publish(`messages:${peerId}`,msg);}}));}));}

Success!

Conclusion

Now that we've added support for Redis PubSub, we can scale our service to any number of server nodes (so long as we have a redis server that we can communicate between). Connections will remain open per node process while the messages and channel communication will be routed through redis to ensure that every message is delivered through the proper server sent event stream.

Thanks for following along!

Cheers! 🍻