UCOMmandIt
/
uci-mqtt
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

0.3.0 bump for 3-2020 deployment of light code 

adjust event names to match uci socket module
catch no packet for send and push
master v0.3.0
David Kebler 2020-03-15 15:56:48 -07:00
parent a1a9eb0822
commit f5fde15007
2 changed files with 13 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
package.json
View File

@ -1,6 +1,6 @@
{ {
"name": "@uci/mqtt", "name": "@uci/mqtt",
"version": "0.1.13", "version": "0.3.0",
"description": "mqtt client with UCI json packet payloads", "description": "mqtt client with UCI json packet payloads",
"main": "src", "main": "src",
"scripts": { "scripts": {
@ -40,7 +40,7 @@
"chai": "^4.2.0", "chai": "^4.2.0",
"delay": "^4.3.0", "delay": "^4.3.0",
"esm": "^3.2.25", "esm": "^3.2.25",
"mocha": "^6.2.2", "mocha": "^7.1.0",
"nodemon": "^2.0.0" "nodemon": "^2.0.2"
} }
} }

15
src/mqtt.js
View File

@ -42,6 +42,7 @@ class MQTTClient extends EventEmitter {
this.mqtt = {} // orginal mqtt.js object when created by connect this.mqtt = {} // orginal mqtt.js object when created by connect
// self bindings // self bindings
this._connected = false this._connected = false
this.conPackets = opts.conPackets || [opts.conPacket]
this.connect = this.connect.bind(this) this.connect = this.connect.bind(this)
this.publish = this.publish.bind(this) this.publish = this.publish.bind(this)
this._subscribe = this._subscribe.bind(this) this._subscribe = this._subscribe.bind(this)
@ -77,14 +78,14 @@ class MQTTClient extends EventEmitter {
this._connected=true this._connected=true
this._subscribe(this.topics) // resubscribe this._subscribe(this.topics) // resubscribe
this.emit('status',{level:'info', msg:'reconnected', id:this.id, opts:this.opts, connected:this._connected}) this.emit('status',{level:'info', msg:'reconnected', id:this.id, opts:this.opts, connected:this._connected})
this.emit('consumer-connection', {state:'reconnected', id:this.id}) this.emit('connection:consumer', {state:'connected', id:this.id, mqtt:true})
if (this.opts.conPacket) this.send(this.conPacket) if (this.opts.conPacket) this.send(this.conPacket)
}) })
this.mqtt.on('error', err => { this.mqtt.on('error', err => {
this.emit('status',{level:'fatal', method:'connect', err: err, msg:'connection error to broker'}) this.emit('status',{level:'fatal', method:'connect', err: err, msg:'connection error to broker'})
log.fatal({method:'connect', line:76, err: err, msg:'connection error to broker'}) log.fatal({method:'connect', line:76, err: err, msg:'connection error to broker'})
this.emit('consumer-connection', {state:'disconnected', id:this.id}) this.emit('connection:consumer', {state:'disconnected', id:this.id, mqtt:true})
this._connected=false this._connected=false
}) })
@ -106,7 +107,7 @@ class MQTTClient extends EventEmitter {
if (err) reject({error:'unable to subscribe to topics after connection', err:err}) if (err) reject({error:'unable to subscribe to topics after connection', err:err})
else{ else{
this.emit('status',{level:'info', msg:'connected', id:this.id, opts:this.opts, connected:this._connected}) this.emit('status',{level:'info', msg:'connected', id:this.id, opts:this.opts, connected:this._connected})
this.emit('consumer-connection', {state:'connected', id:this.id}) this.emit('connection:consumer', {state:'connected', id:this.id, mqtt:true})
if (this.opts.conPacket) this.send(this.conPacket) if (this.opts.conPacket) this.send(this.conPacket)
resolve(`mqtt client connected to broker at ${this.mqtt.options.hostname}:${this.mqtt.options.port}`) resolve(`mqtt client connected to broker at ${this.mqtt.options.hostname}:${this.mqtt.options.port}`)
} }
@ -140,6 +141,8 @@ class MQTTClient extends EventEmitter {
// if topic not passed then it will send to the topic given by the cmd // if topic not passed then it will send to the topic given by the cmd
async _send(ipacket, topic, options) { // packet is required async _send(ipacket, topic, options) { // packet is required
if (!ipacket || !Object.keys(ipacket).length) return {error:'send aborted, no packet passed'}
if (isPlainObject(topic)) { // if neither then assume none and what is passed is options if (isPlainObject(topic)) { // if neither then assume none and what is passed is options
options = topic || {} options = topic || {}
topic = '' topic = ''
@ -184,7 +187,9 @@ class MQTTClient extends EventEmitter {
}) // end promise }) // end promise
} }
async _push(packet={}, topic, options) { async _push(packet, topic, options) {
if (!packet || !Object.keys(packet).length) return {error:'push aborted, no packet passed', packet:packet, topic:topic}
if (!this._connected) { if (!this._connected) {
log.error({method:'send', line:105, url:this.url, opts:this.opts, msg:'no active connection to broker'}) log.error({method:'send', line:105, url:this.url, opts:this.opts, msg:'no active connection to broker'})
@ -211,7 +216,7 @@ class MQTTClient extends EventEmitter {
// this is internal UCI publish // this is internal UCI publish
async _publish(packet,topic,options) { async _publish(packet,topic,options) {
if(!topic) return {error: 'no topic to which to publish packet'} if(!topic) return {error: 'no topic to which to publish packet'}
let payload = packet.payload ? (typeof packet.payload ==='string' ? packet.payload : this._serialize(packet)) : this._serialize(packet) // payload is entire packet let payload = packet.payload!=null ? ( (typeof packet.payload ==='string' ? packet.payload : this._serialize(packet.payload)) ) : this._serialize(packet) // payload is entire packet
if (!payload) { if (!payload) {
log.error({method:'send', line:163, msg:'not able to serilaize packet', packet:packet}) log.error({method:'send', line:163, msg:'not able to serilaize packet', packet:packet})
return {error:'not able to serilaize packet', packet:packet} return {error:'not able to serilaize packet', packet:packet}