Manage forward communication and queue system with @scorpion9979

Author: marcocastignoli

migrations/20190210201120-add-queue.js

@@ -0,0 +1,18 @@
+'use strict'
+
+exports.up = function (db) {
+  return db.runSql(`
+    CREATE TABLE queue (
+      queue_id   INTEGER PRIMARY KEY,
+      sender       INTEGER    NOT NULL,
+      recipient    INTEGER,
+      message TEXT
+    );
+  `)
+}
+
+exports.down = function (db) {
+  return db.runSql(`
+    DROP TABLE queue;
+`)
+}

src/Instance.js

@@ -102,9 +102,46 @@ module.exports = class Instance extends EventEmitter {
           channel: 'instance.to.client',
           action: 'pinged'
         }
+      },
+      {
+        channel: 'client.to.node',
+        callback: this.forwardClientToNode.bind(this)
       }
     ]
   }
+  async forwardClientToNode (protocol, sender, parameters, reply, forwardObject) {
+    let onlineRecipient = this.online.nodes.find(n => n.id === forwardObject.recipient)
+    if (onlineRecipient) {
+      this.protocols[onlineRecipient.protocol].to(onlineRecipient.resource, 'client.to.node', forwardObject.action, parameters)
+      reply({ msg: 'Message forwarded' })
+    } else {
+      const { success, results } = await this.db.run(` SELECT resource FROM nodes where node_id = ?`, forwardObject.recipient)
+      if (success) {
+        const resource = results[0].resource
+        // TODO: find a better regex to split ws://923u10hdashdsh into ['ws', '923u10hdashdsh']
+        const [, recipientProtocol, recipientResource] = resource.match(/(^\w+):\/\/([a-z0-9]+)/)
+        if (this.protocols[recipientProtocol].needsQueue) {
+          const { success, results } = await this.isClientTokenValid(parameters.token)
+          // we don't want to pass the token to the recipient !!! TODO: change the authentication system, put the token at the level of action, properties, recipient
+          parameters.token = ''
+          if (success) {
+            const client = results[0]
+            const { success } = await this.db.run(`
+              INSERT INTO queue 
+                (sender, recipient, message)
+              VALUES
+                (?, ?, ?)`, [client.client_id, forwardObject.recipient, parameters])
+            if (success) {
+              reply({ msg: 'Message stored in the queue' })
+            }
+          }
+        } else {
+          this.protocols[recipientProtocol].to(recipientResource, 'client.to.node', forwardObject.action, parameters)
+          reply({ msg: 'Message forwarded' })
+        }
+      }
+    }
+  }
   loadListeners () {
     for (let protocolId in this.protocols) {
       let protocol = this.protocols[protocolId]

src/protocols/Http.js

@@ -3,6 +3,7 @@ class Http {
   constructor (http) {
     this.resource = http
     this.ID = 'http'
+    this.needsQueue = false
   }
   to (recipient, channel, action, parameters, response) {
     fetch(recipient + channel + '/' + action, {

src/protocols/WebSockets.js

@@ -2,13 +2,22 @@ class WebSockets {
   constructor (ws) {
     this.resource = ws
     this.ID = 'ws'
+    this.needsQueue = true
   }
-  to (recipient, channel, action, parameters, response) {
-    if (recipient) {
-      this.resource.to(recipient).emit(channel, {
-        action: action,
-        parameters: parameters
-      })
+  to (delivery, channel, action, parameters, response) {
+    if (delivery) {
+      if (typeof delivery === 'object') {
+        this.resource.to(delivery.mediator).emit(channel, {
+          action: action,
+          parameters: parameters,
+          recipient: delivery.recipient
+        })
+      } else if (typeof delivery === 'string') {
+        this.resource.to(delivery).emit(channel, {
+          action: action,
+          parameters: parameters
+        })
+      }
     } else {
       this.resource.emit(channel, {
         action: action,
@@ -22,12 +31,13 @@ class WebSockets {
   on (channel, action, resource, callback, response) {
     let resourceId = resource.id
     resource.on(channel, data => {
-      if (data.action === action) {
+      if (!action || data.action === action) {
         callback(
           this.ID,
           resourceId,
           data.parameters,
-          (parameters) => this.to(resourceId, response.channel, response.action, parameters, { listen: null, resource })
+          (parameters) => this.to(resourceId, response.channel, response.action, parameters, { listen: null, resource }),
+          { recipient: data.recipient, action: action }
         )
       }
     })