host protocol scheme

Author: alanm

package.json

@@ -1,7 +1,7 @@
 {
   "name": "@dataparty/api",
   "private": false,
-  "version": "1.2.16",
+  "version": "1.2.17",
   "main": "dist/dataparty.js",
   "frontend": "dist/dataparty-browser.js",
   "backend": "dist/dataparty.js",
@@ -76,6 +76,7 @@
     "express": "^4.17.1",
     "express-list-routes": "^0.1.4",
     "git-repo-info": "^2.1.1",
+    "joi-objectid": "^4.0.2",
     "jshashes": "^1.0.8",
     "jsonpath-plus": "^0.20.1",
     "last-eventemitter": "^1.1.1",
@@ -86,6 +87,7 @@
     "morgan": "^1.10.0",
     "multer": "^1.4.5-lts.1",
     "nconf": "^0.10.0",
+    "node-mocks-http": "^1.12.1",
     "node-persist": "^3.0.1",
     "origin-router": "^1.6.4",
     "parse-url": "^5.0.1",

src/comms/host-op.js

@@ -0,0 +1,101 @@
+const Hoek = require('@hapi/hoek')
+const debug = require('debug')('roshub.op')
+const moment = require('moment')
+const EventEmitter = require('eventemitter3')
+
+const OP_STATUS_LEVEL = {
+  Status_Undefined: 'Status_Undefined',
+  Status_None: 'Status_None',
+  Status_Error: 'Status_Error',
+  Status_Warning: 'Status_Warning',
+  Status_Info: 'Status_Info',
+  Status_Debug: 'Status_Debug'
+}
+
+
+const OP_STATES = {
+  Pending: 'Pending',
+  Starting: 'Starting',
+  Running: 'Running',
+  Finished_Success: 'Finished_Success',
+  Finished_Fail: 'Finished_Fail',
+  Finished_Permission_Denied: 'Finished_Permission_Denied',
+  Finished_Invalid_Op: 'Finished_Invalid_Op',
+  Finished_Invalid_Id: 'Finished_Invalid_Id',
+  Finished_Buffer_Full: 'Finished_Buffer_Full'
+}
+
+class Op extends EventEmitter {
+  constructor({msg, input}){
+    super()
+    this.msg = msg      //! original encrypted message
+    this.input = input  //! parsed and validated input
+    this.start = Date.now()
+    this.end = undefined
+    this.state = Op.STATES.Pending
+    this.level = Op.STATUS_LEVELS.Status_Info
+  }
+
+  get op(){
+    return Hoek.reach(this.input, 'op')
+  }
+
+  get id(){
+    return Hoek.reach(this.input, 'id')
+  }
+
+
+  static get STATES(){
+    return OP_STATES
+  }
+
+  static get STATUS_LEVELS(){
+    return OP_STATUS_LEVEL
+  }
+
+  setState(state, level, message){
+
+    this.state = state
+
+    const status = {
+      op: 'status',
+      id: this.id,
+      state: this.state,
+      level: level || Op.STATUS_LEVELS.Status_Info,
+      msg: message || undefined,
+      timestamp: Date.now(),
+      stats: {
+        start: this.start,
+        end: this.end,
+        duration_ms: this.end - this.start
+      }
+    }
+
+    if(this.state == Op.STATES.Finished_Success ||
+      this.state == Op.STATES.Finished_Fail ||
+      this.state == Op.STATES.Finished_Invalid_Op ||
+      this.state == Op.STATES.Finished_Invalid_Id ||
+      this.state == Op.STATES.Finished_Permission_Denied ||
+      this.state == Op.STATES.Finished_Buffer_Full)
+    {
+      debug('set end time')
+      this.end = Date.now()
+      this.emit('finished', true)
+    }
+    else{
+      this.emit('finished', false)
+      this.emit('status', status)
+    }
+  }
+
+  setOpStatusLevel(level){
+    this.level = level
+  }
+
+  setStatus(level, message){
+    this.emit('status', )
+  }
+
+}
+
+module.exports = Op

src/comms/protocol-scheme.js

@@ -0,0 +1,82 @@
+const Joi = require('@hapi/joi')
+Joi.objectId = require('joi-objectid')(Joi)
+
+const ID_SCHEME = Joi.alternatives().try(Joi.string(), Joi.number())
+
+const OP_HEADER = Joi.object().keys({
+  id: ID_SCHEME.required(),
+  op: Joi.string().valid([
+    'auth',
+    'advertise',
+    'subscribe',
+    'unsubscribe',
+    'publish'
+  ]).required()
+})
+
+
+const AUTH_OP = Joi.object().keys({
+  id: ID_SCHEME.required(),
+  op: Joi.string().valid('auth').required(),
+  session: Joi.objectId().required()
+})
+
+const ADVERTISE_OP = Joi.object().keys({
+  id: ID_SCHEME.required(),
+  op: Joi.string().valid('advertise').required(),
+  type: Joi.string().required(),
+  topic: Joi.string().required(),
+  queue_size: Joi.number(),
+  latch: Joi.boolean()
+})
+
+const UNADVERTISE_OP = Joi.object().keys({
+  id: ID_SCHEME.required(),
+  op: Joi.string().valid('unadvertise').required(),
+  topic: Joi.string().required()
+})
+
+
+const SUBSCRIBE_OP = Joi.object().keys({
+  id: ID_SCHEME.required(),
+  op: Joi.string().valid('subscribe').required(),
+  type: Joi.string().required(),
+  topic: Joi.string().required(),
+  compression: Joi.string(),
+  queue_length: Joi.number(),
+  throttle_rate: Joi.number()
+})
+
+const UNSUBSCRIBE_OP = Joi.object().keys({
+  id: ID_SCHEME.required(),
+  op: Joi.string().valid('unsubscribe').required(),
+  topic: Joi.string().required()
+})
+
+const PUBLISH_OP = Joi.object().keys({
+  id: ID_SCHEME.required(),
+  op: Joi.string().valid('publish').required(),
+  topic: Joi.string().required(),
+  latch: Joi.boolean(),
+  msg: Joi.any()
+})
+
+
+const ANY_OP = Joi.alternatives().try(
+  ADVERTISE_OP,
+  UNADVERTISE_OP,
+  SUBSCRIBE_OP,
+  UNSUBSCRIBE_OP,
+  PUBLISH_OP
+)
+
+module.exports = {
+  'OP_HEADER': OP_HEADER,
+  'AUTH_OP': AUTH_OP,
+  'ADVERTISE_OP': ADVERTISE_OP,
+  'UNADVERTISE_OP': UNADVERTISE_OP,
+  'SUBSCRIBE_OP': SUBSCRIBE_OP,
+  'UNSUBSCRIBE_OP': UNSUBSCRIBE_OP,
+  'PUBLISH_OP': PUBLISH_OP,
+  'ANY_OP': ANY_OP
+}