uci-base/src/base.js

// communication modules
import Socket from '@uci/socket'
import MQTT from '@uci/mqtt'  // requires broker
// import MQTT from '../../uci-mqtt/src/client'  // requires broker
import WebSocket from '@uci/websocket' // server only
// TODO

import EventEmitter from 'events'
import pSettle from 'p-settle'
import { bindFuncs } from '@uci/utils/src/function'

import { processor, commands, namespaces } from './processing'

import logger from '@uci/logger'
let log = {}

export default class Base extends EventEmitter {
  constructor(opts={}) {
    super()
    this.id = opts.id || opts.name || 'uci-base:'+ new Date().getTime()
    log = logger({name:'base',id:this.id})
    this.desc = opts.desc // additional details for humans
    this.started = false  // flag to know when instance has been initialized
    this.socket={} // holds all the various communication sockets
    this._processors = { _default: processor }
    this._defaultCmds = commands
    this._namespaces = namespaces
    this.bindFuncs = bindFuncs
    // predefined sockets:
    // comma delimited list of this form  '<name>#<c/p/s>><n=np/t=tcp/m=mqtt/w=web>'
    this.socket = {}
    if(opts.sockets) {
      opts.sockets.split(/[,|\s]+/).forEach( socketStr => {
        let socket = {}
        socketStr.split(/[>#]+/).map(function(prop,index) {
          socket[SOCKET_INFO_KEYS[index]] = prop
        })
        this.addSocket(socket.name, socket.type, socket.transport,opts[socket.name])
      })
    }

  } // end constructor

  // Class Methods

  async init () {
    let sockets = []
    let initSockets = []
    for(let name of Object.keys(this.socket)){
      initSockets.push(this.initSocket(name))
      sockets.push(name)
    }
    this.started = true
    return pSettle(initSockets).then(res=>{
      console.log(res)
      let err = []
      res.forEach((p,index) => {
        if (p.isRejected) {
          err.push({name:sockets[index],err:p.reason})
        }
      })
      return err
    })
    // TODO if a websocket server was working then push status
    // TODO if no mqtt broker then attempt to start one

  } // init

  async addSocket (name, type='c', transport='n', options={}) {
  // transport: (n) named pipe/ unix socket,  (t) tcp socket,  (m) mqtt subscribe, (w) websocket
    log.info({socketName:name, type:type, tranport:transport, options:options},`adding socket ${name}`)
    options.id = this.id +':'+ name
    switch (transport) {
    case 'n':
      options.path = options.path || true
    // falls through
    case 't':
      this.socket[name] = new Socket[TRANSLATE[type]](options)
      break
    case 'm':
      if (type === 'p') type ='c'
      options.connect = options.connect || {}
      options.connect.connectTimeout = options.connect.connectTimeout || 5000
      this.socket[name] =  new MQTT(options)
      break
    case 'w':
      if (type ==='s') this.socket[name] = new WebSocket(options)
      else log.warn({name:name,type:type,transport:transport},'Web socket not created Consumer/Client Web Socket not supported')
    }
    // console.log(name, '====',this.socket[name])
    this.socket[name].name = name
    this.socket[name].type = type
    this.socket[name].transport = transport
    this.socket[name]._packetProcess = this._packetProcess.bind(this,name)
    if (this.started) await this.initSocket(name)
    else return `socket ${name} added`
    // console.log(name, '====',this.socket[name])

  }

  async initSocket(name) {
    let socket = this.socket[name]
    let init = {}
    if (this.socket[name].type ==='s' &&  this.socket[name].transport !=='m') {
      init = socket.create
    } else {
      init = socket.connect
    }
    log.info(`initializing socket ${name}, ${socket.type}, ${socket.transport}`)
    if(this.started) return `socket ${name} added and initialzed, ${await init()}`
    else return init()
  }

  async removeSocket (name) {
    //TODO
  }

  async send (name,packet) {
    if (typeof name !== 'string') {
      packet = name
      let sends = []
      for(let name of Object.keys(this.socket)){
        if (this.socket[name].type ==='c') {
          // console.log(name, this.socket[name])
          sends.push(this.socket[name].send.bind(this.socket[name]))
        }
      }
      // console.log(sends.map(send => {return send(packet)}))
      if (sends.length === 1) return sends[0](packet)
      return Promise.all(sends.map(send => {return send(packet)}))
    }  else {
      if (this.socket[name]) return await this.socket[name].send(packet)
      else return {error: `not consumer socket of name ${name}`}
    }
  }

  async sendTransport(packet,transport) {
    let sends = []
    for(let name of Object.keys(this.socket)){
      if (this.socket[name].type ==='c') {
        if (this.socket[name].transport === transport) {
          sends.push(this.socket[name].send.bind(this.socket[name]))
        }
      }
    }
    if (sends.length === 1) return sends[0](packet)
    return Promise.all(sends.map(send => {return send(packet)}))
  }

  // async sendMQTT(topic, packet) {return this.socket.mqtt.send(topic, packet)}
  async sendTCP(packet) {return this.sendTransport(packet,'t')}
  async sendIPC(packet) {return this.sendTransport(packet,'n')}

  getSocket(name) {return this.socket[name]}

  getPacketByName(name, packets) {
    if (!packets.length) packets = [packets]
    let found = {}
    packets.some((packet,index,packets) => {
      if (packet._header.sender.name === name) {
        found = packets[index]
        return true
      }
    })
    return found
  }


  // TODO confirm Object.assign will be ok as it is not a deep copy
  amendConsumerProcessing(funcs,trans) {
    if (trans) {
      if (!this._defaultCmds.c[trans]) this._defaultCmds.c[trans] ={}
      Object.assign(this._defaultCmds.c[trans],funcs)
    }
    Object.assign(this._defaultCmds.c,funcs)
  }

  amendSocketProcessing(funcs,trans) {
    if (trans) {
      if (!this._defaultCmds.c[trans]) this._defaultCmds.c[trans] ={}
      Object.assign(this._defaultCmds.c[trans],funcs)
    }
    Object.assign(this._defaultCmds.c,funcs)
  }

  // use s: and c: keys  TODO need to change this
  addNamedProcessing(name,funcs,type) {
    if (type){
      if(!this._cmds[name][type]) this._cmds[name][type] = {}
      Object.assign(this._cmds[name][type],funcs)
    } else {
      if(!this._cmds[name]) this._cmds[name] ={}
      Object.assign(this._cmds[name],funcs)
    }
  }

  // takes and returns a packet
  beforeSendHook (type,funcs){}  // TODO before packet send
  afterReceiveHook(type,funcs){}  // TODO after receiv
  afterProcessHook(type,funcs){}  // TODO

  // here you can add namespaced functions for packet commands
  consumersProcessor(func) {
    for(let name of Object.keys(this.socket)){
      if (this.socket[name].type ==='c') {
        this.socketNamedProcessor(func,name)
      }
    }
  }

  socketsProcessor(func) {
    for(let name of Object.keys(this.socket)){
      if (this.socket[name].type ==='s') {
        this.socketNamedProcessor(func,name)
      }
    }
  }

  socketNameProcessor(func,socket_name) {
    socket_name = socket_name || '_default'
    this._processors[socket_name]._process = func
  }

  addNamespace(space,type,trans) {
    if (trans) return this._namespaces[type+trans].unshift(space)
    else return this._namespaces[type].unshift(space)
  }

  // registerPacketProcessor(name,func) {
  //   this._packetProcess = func
  // }


  /*
  *
  * Private Methods
  *
  */

  _transport(name) {return this.socket[name].transport} //getter for socket transport
  _type(name) {return this.socket[name].type} //getter for socket type

  _getTransportNamespaces(socket) {
    return this._namespace[this._type(socket)+this._transport(socket)]
  }

  _getCmdFuncNamespace (cmd,namespaces) {
    let cmd_func = null
    namespaces.some( namespace => {
      namespace = namespace ? namespace+'.'+cmd : cmd
      cmd_func = this._getCmdFunc(namespace)
      if (cmd_func) return true
    })
    return cmd_func
  }

  // takes command and returns corresponding function in a hash
  _getCmdFunc (cmd,obj) {
    // console.log('obj',obj)
    if (typeof cmd ==='string') {
      if (!obj) obj = this
      cmd=cmd.split(/[.:/]+/)
      // console.log('===================',cmd)
    }
    var prop=cmd.shift()
    if (cmd.length === 0) return obj[prop]
    if(!obj[prop]) return null
    // console.log(cmd.length,cmd,prop, obj[prop])
    return this._getCmdFunc(cmd, obj[prop])
  }

  async _callCmdFunc(packet,socket) {
    let cmd_func = this._getCmdFuncNamespace(packet.cmd,this._namespaces[this._type(socket)+this._transport(socket)])
    if (cmd_func) return await cmd_func.bind(this)(packet)
    cmd_func = this._getCmdFuncNamespace(packet.cmd,this._namespaces[this._type(socket)])
    if (cmd_func) return await cmd_func.bind(this)(packet)
    return 'failed'
  }

  /*
  **********default packet processor for all sockets
  */

  async _packetProcess (socket_name,packet) {
    // console.log(socket_name,packet)
    let processor = packet._processor || this._processors[socket_name] || '_default'
    return await this._processors[processor].bind(this)(packet,socket_name,this._processors[processor])
  }


} // end class

const SOCKET_INFO_KEYS = ['name','type','transport']
const TRANSLATE= {n:'Named Pipe',t:'TCP',s:'Socket',c:'Consumer', m:'MQTT', w:'Web Socket'}