add optional authorize callback and update dependencies

Author: bourgeoa

lib/server.js

@@ -16,6 +16,7 @@ function WsServer (server, opts) {
   opts = opts || {}
   this.suffix = opts.suffix || '.changes'
   this.store = opts.store || new InMemory(opts)
+  this.authorize = opts.authorize // Authorization callback
   var toChannel = opts.toChannel || defaultToChannel
 
   // Starting WSS server
@@ -26,14 +27,20 @@ function WsServer (server, opts) {
   })
 
   // Handling a single connection
-  wss.on('connection', function (client) {
+  wss.on('connection', function (client, req) {
     debug('New connection')
-    // var location = url.parse(client.upgradeReq.url, true)
+    // Store the request for authorization checks
+    client.upgradeReq = req
 
     // Handling messages
     client.on('message', function (message) {
       debug('New message: ' + message)
 
+      // ws@8 may send Buffer, convert to string
+      if (Buffer.isBuffer(message)) {
+        message = message.toString()
+      }
+
       if (!message || typeof message !== 'string') {
         return
       }
@@ -47,15 +54,38 @@ function WsServer (server, opts) {
         return
       }
 
-      var channel = toChannel ? toChannel(iri) : iri
-      self.store.subscribe(channel, iri, client, function (err, uuid) {
-        if (err) {
-          // TODO Should return an error
-          return
-        }
-
-        client.send('ack ' + tuple[1])
-      })
+      // Check authorization if callback is provided
+      if (self.authorize) {
+        self.authorize(iri, req, function (err, allowed) {
+          if (err || !allowed) {
+            debug('Subscription denied for ' + iri)
+            client.send('err ' + iri + ' forbidden')
+            return
+          }
+
+          // Authorization passed, proceed with subscription
+          var channel = toChannel ? toChannel(iri) : iri
+          self.store.subscribe(channel, iri, client, function (err, uuid) {
+            if (err) {
+              client.send('err ' + iri + ' error')
+              return
+            }
+
+            client.send('ack ' + tuple[1])
+          })
+        })
+      } else {
+        // No authorization, proceed directly
+        var channel = toChannel ? toChannel(iri) : iri
+        self.store.subscribe(channel, iri, client, function (err, uuid) {
+          if (err) {
+            // TODO Should return an error
+            return
+          }
+
+          client.send('ack ' + tuple[1])
+        })
+      }
     })
 
     // Respond to ping