You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

170 lines
7.6 KiB
JavaScript

const ConnectorRegistry = require("./ConnectorRegistry")
const config = require('../config.json')
const md5 = require('md5')
class MessageManager {
constructor() {
this.messages = {}
ConnectorRegistry.events.on('msg:status', this.msgStatus.bind(this))
}
async New(type, routingParams, payload) {
if (!routingParams.device) {
routingParams.device = 'generic'
}
//console.log(routingParams)
const msgObj = {
type,
routingParams,
payload,
_payload: payload,
date: new Date()
}
await require("./DeviceRegistry").Devices[ routingParams.device ].formatTX(msgObj)
// console.log('finished msg obj is ', msgObj)
this.messages[ msgObj.id ]._routerData = this.messages[ msgObj.id ]._routerData || {}
const hwDuplexSupport = require("./DeviceRegistry").Devices[ routingParams.device ].duplex || false
Object.assign(this.messages[ msgObj.id ]._routerData, hwDuplexSupport ? {
duplexCapable: true,
recvAck: false, // aka "delivered"
readAck: false, // "read"
response: false, // "resp"
deliveryLog: {},
} : {
duplexCapable: false,
deliveryLog: {},
})
console.log(`Type:\t\t${ type }\nDevice:\t\t${ routingParams.device }`)
console.log(`Message UUID:\t${ msgObj.id } HEX: ${ Buffer.from(msgObj.id, 'utf-8').toString('hex') }`)
console.log(`Connectors:\t${ JSON.parse(JSON.stringify(routingParams.connectors)).map(x=>`${x[0]}=${x.splice(1).join(',')}`).join('&') }`)
console.log(`Message Original Payload:\t"${ payload }"`)
console.log(`Processed Device Payload:\t"${ msgObj.payload }"`)
return msgObj.id
}
async msgStatus(msgId, uuid, status) {
this.messages[ msgId ]._routerData.deliveryLog[ uuid ] = status
if (status === 'delivered') this.messages[ msgId ]._routerData.recvAck = true
//this.Deliver(msgId)
console.log(msgId, uuid, 'status is', status)
}
async Deliver(msgId) {
if (this.messages[ msgId ].type === 'duplex')
this.DeliverDuplex(msgId)
else
this.DeliverOneWay(msgId)
}
attachMetadata(msgId, metadata) {
if (!this.messages[ msgId ]._routerData.metadata) {
this.messages[ msgId ]._routerData.metadata = []
}
this.messages[ msgId ]._routerData.metadata.push(metadata)
}
markMessageRead(msgId) {
this.messages[ msgId ]._routerData.readAck = true
}
respondToMessage(msgId, response) {
this.messages[ msgId ]._routerData.response = response
}
_clearEventHandlers4MsgID(msgId) {
ConnectorRegistry.events.removeAllListeners(`msg:status:${ msgId }:delivered`)
ConnectorRegistry.events.removeAllListeners(`msg:status:${ msgId }:failed`)
}
async DeliverDuplex(msgId) {
const msg = this.messages[ msgId ]
if (!msg._routerData) throw `No Routerdata attached to msg with id ${ msgId }`
if (!!msg.locked) throw 'message is locked'
//console.log(msg.routingParams.connectors)
let deliveryChain = msg.routingParams.connectors.map((connectorDeliveryTry) => {
const connectorName = connectorDeliveryTry[0],
connectorArgs = connectorDeliveryTry.slice(1),
connectorConfig = config.connectors[connectorName]
//const UUID = md5(JSON.stringify(connectorDeliveryTry))
const chainPromise = (res) => {
ConnectorRegistry.events.removeAllListeners(`msg:status:${ msgId }:failed`)
setTimeout(() => {
res([false, 'timeout'])
}, !!connectorConfig && !!connectorConfig.duplexTimeout ? connectorConfig.duplexTimeout*1e3 : 30e3)
ConnectorRegistry.events.once(`msg:status:${ msgId }:failed`, () => {
console.log(`${ msgId } failed via ${ connectorName }:${ connectorArgs.join(',') }, continuing...`)
res([false, 'failed'])
})
console.log(`Trying to deliver msg#${ msg.id } with ${ connectorName }:${ connectorArgs.join(',') }`)
ConnectorRegistry.transmit(connectorName, msg, connectorArgs)
}
return chainPromise
})
Promise.race([
new Promise(res => { // Delivery Event for this message
ConnectorRegistry.events.once(`msg:status:${ msgId }:delivered`, (_, uuid) => {
console.log(`${ msgId } delivered via ${ uuid }`)
return res()
})
}),
new Promise(async (res, rej) => {
for(let deliveryFunction of deliveryChain) {
//when a different verification channel is used for ACK
if (this.messages[ msgId ]._routerData.recvAck === true) break;
let result = await new Promise(deliveryFunction)
if (result[0] === true) { res(); break; }
}
rej()
})
])
.then(($) => {
this._clearEventHandlers4MsgID(msgId)
const dLog = this.messages[ msgId ]._routerData.deliveryLog
console.log('DELIVERY WAS A SUCCESS', Object.keys(dLog).map(x=>`${ x } is ${ dLog[x] }`).join('\n'))
})
.catch(($) => {
this._clearEventHandlers4MsgID(msgId)
console.log('DELIVERY WAS A TOTAL FAILURE', $)
})
return true
}
async DeliverOneWay(msgId) {
const msg = this.messages[ msgId ]
if (!msg._routerData) throw `No Routerdata attached to msg with id ${ msgId }`
if (!!msg.locked) throw 'message is locked'
//console.log(msg.routingParams.connectors)
let deliveryChain = msg.routingParams.connectors.map((connectorDeliveryTry) => {
const connectorName = connectorDeliveryTry[0],
connectorArgs = connectorDeliveryTry.slice(1),
connectorConfig = config.connectors[connectorName]
const chainPromise = (res, rej) => {
this._clearEventHandlers4MsgID(msgId)
setTimeout(() => {
res([false, 'timeout'])
}, !!connectorConfig && !!connectorConfig.simplexTimeout ? connectorConfig.simplexTimeout*1e3 : 2e3)
ConnectorRegistry.events.once(`msg:status:${ msgId }:failed`, () => {
console.log(`${ msgId } failed, continuing`)
res([false, 'failed'])
})
console.log(`Trying to deliver msg#${ msg.id } with ${ JSON.stringify(connectorDeliveryTry) }`)
//console.log(this.messages[ msgId ].deliveryLog)
ConnectorRegistry.transmit(connectorName, msg, connectorArgs)
}
return chainPromise
})
//console.log(deliveryChain)
new Promise(async (res, rej) => {
for(let deliveryFunction of deliveryChain) {
let result = await new Promise(deliveryFunction)
if (result[0] === true) { res(result); break; }
}
rej()
})
.then(($) => {
this._clearEventHandlers4MsgID(msgId)
console.log('ROUTING WAS A SUCCESS', $, this.messages[ msgId ]._routerData.deliveryLog)
})
.catch(($) => {
this._clearEventHandlers4MsgID(msgId)
console.log('ROUTING WAS A TOTAL FAILURE', $)
})
}
async BindMsg(msg) {
this.messages[ msg.id ] = msg
}
}
module.exports = new MessageManager()