'use strict'
/**
* Module dependencies
*/
const EventEmitter = require('events').EventEmitter
const Store = require('./store')
const TopicAliasRecv = require('./topic-alias-recv')
const TopicAliasSend = require('./topic-alias-send')
const mqttPacket = require('mqtt-packet')
const DefaultMessageIdProvider = require('./default-message-id-provider')
const Writable = require('readable-stream').Writable
const inherits = require('inherits')
const reInterval = require('reinterval')
const clone = require('rfdc/default')
const validations = require('./validations')
const xtend = require('xtend')
const debug = require('debug')('mqttjs:client')
const nextTick = process ? process.nextTick : function (callback) { setTimeout(callback, 0) }
const setImmediate = global.setImmediate || function (callback) {
// works in node v0.8
nextTick(callback)
}
const defaultConnectOptions = {
keepalive: 60,
reschedulePings: true,
protocolId: 'MQTT',
protocolVersion: 4,
reconnectPeriod: 1000,
connectTimeout: 30 * 1000,
clean: true,
resubscribe: true
}
const socketErrors = [
'ECONNREFUSED',
'EADDRINUSE',
'ECONNRESET',
'ENOTFOUND'
]
// Other Socket Errors: EADDRINUSE, ECONNRESET, ENOTFOUND.
const errors = {
0: '',
1: 'Unacceptable protocol version',
2: 'Identifier rejected',
3: 'Server unavailable',
4: 'Bad username or password',
5: 'Not authorized',
16: 'No matching subscribers',
17: 'No subscription existed',
128: 'Unspecified error',
129: 'Malformed Packet',
130: 'Protocol Error',
131: 'Implementation specific error',
132: 'Unsupported Protocol Version',
133: 'Client Identifier not valid',
134: 'Bad User Name or Password',
135: 'Not authorized',
136: 'Server unavailable',
137: 'Server busy',
138: 'Banned',
139: 'Server shutting down',
140: 'Bad authentication method',
141: 'Keep Alive timeout',
142: 'Session taken over',
143: 'Topic Filter invalid',
144: 'Topic Name invalid',
145: 'Packet identifier in use',
146: 'Packet Identifier not found',
147: 'Receive Maximum exceeded',
148: 'Topic Alias invalid',
149: 'Packet too large',
150: 'Message rate too high',
151: 'Quota exceeded',
152: 'Administrative action',
153: 'Payload format invalid',
154: 'Retain not supported',
155: 'QoS not supported',
156: 'Use another server',
157: 'Server moved',
158: 'Shared Subscriptions not supported',
159: 'Connection rate exceeded',
160: 'Maximum connect time',
161: 'Subscription Identifiers not supported',
162: 'Wildcard Subscriptions not supported'
}
function defaultId () {
return 'mqttjs_' + Math.random().toString(16).substr(2, 8)
}
function applyTopicAlias (client, packet) {
if (client.options.protocolVersion === 5) {
if (packet.cmd === 'publish') {
let alias
if (packet.properties) {
alias = packet.properties.topicAlias
}
const topic = packet.topic.toString()
if (client.topicAliasSend) {
if (alias) {
if (topic.length !== 0) {
// register topic alias
debug('applyTopicAlias :: register topic: %s - alias: %d', topic, alias)
if (!client.topicAliasSend.put(topic, alias)) {
debug('applyTopicAlias :: error out of range. topic: %s - alias: %d', topic, alias)
return new Error('Sending Topic Alias out of range')
}
}
} else {
if (topic.length !== 0) {
if (client.options.autoAssignTopicAlias) {
alias = client.topicAliasSend.getAliasByTopic(topic)
if (alias) {
packet.topic = ''
packet.properties = { ...(packet.properties), topicAlias: alias }
debug('applyTopicAlias :: auto assign(use) topic: %s - alias: %d', topic, alias)
} else {
alias = client.topicAliasSend.getLruAlias()
client.topicAliasSend.put(topic, alias)
packet.properties = { ...(packet.properties), topicAlias: alias }
debug('applyTopicAlias :: auto assign topic: %s - alias: %d', topic, alias)
}
} else if (client.options.autoUseTopicAlias) {
alias = client.topicAliasSend.getAliasByTopic(topic)
if (alias) {
packet.topic = ''
packet.properties = { ...(packet.properties), topicAlias: alias }
debug('applyTopicAlias :: auto use topic: %s - alias: %d', topic, alias)
}
}
}
}
} else if (alias) {
debug('applyTopicAlias :: error out of range. topic: %s - alias: %d', topic, alias)
return new Error('Sending Topic Alias out of range')
}
}
}
}
function removeTopicAliasAndRecoverTopicName (client, packet) {
let alias
if (packet.properties) {
alias = packet.properties.topicAlias
}
let topic = packet.topic.toString()
if (topic.length === 0) {
// restore topic from alias
if (typeof alias === 'undefined') {
return new Error('Unregistered Topic Alias')
} else {
topic = client.topicAliasSend.getTopicByAlias(alias)
if (typeof topic === 'undefined') {
return new Error('Unregistered Topic Alias')
} else {
packet.topic = topic
}
}
}
if (alias) {
delete packet.properties.topicAlias
}
}
function sendPacket (client, packet, cb) {
debug('sendPacket :: packet: %O', packet)
debug('sendPacket :: emitting `packetsend`')
client.emit('packetsend', packet)
debug('sendPacket :: writing to stream')
const result = mqttPacket.writeToStream(packet, client.stream, client.options)
debug('sendPacket :: writeToStream result %s', result)
if (!result && cb && cb !== nop) {
debug('sendPacket :: handle events on `drain` once through callback.')
client.stream.once('drain', cb)
} else if (cb) {
debug('sendPacket :: invoking cb')
cb()
}
}
function flush (queue) {
if (queue) {
debug('flush: queue exists? %b', !!(queue))
Object.keys(queue).forEach(function (messageId) {
if (typeof queue[messageId].cb === 'function') {
queue[messageId].cb(new Error('Connection closed'))
// This is suspicious. Why do we only delete this if we have a callbck?
// If this is by-design, then adding no as callback would cause this to get deleted unintentionally.
delete queue[messageId]
}
})
}
}
function flushVolatile (queue) {
if (queue) {
debug('flushVolatile :: deleting volatile messages from the queue and setting their callbacks as error function')
Object.keys(queue).forEach(function (messageId) {
if (queue[messageId].volatile && typeof queue[messageId].cb === 'function') {
queue[messageId].cb(new Error('Connection closed'))
delete queue[messageId]
}
})
}
}
function storeAndSend (client, packet, cb, cbStorePut) {
debug('storeAndSend :: store packet with cmd %s to outgoingStore', packet.cmd)
let storePacket = packet
let err
if (storePacket.cmd === 'publish') {
// The original packet is for sending.
// The cloned storePacket is for storing to resend on reconnect.
// Topic Alias must not be used after disconnected.
storePacket = clone(packet)
err = removeTopicAliasAndRecoverTopicName(client, storePacket)
if (err) {
return cb && cb(err)
}
}
client.outgoingStore.put(storePacket, function storedPacket (err) {
if (err) {
return cb && cb(err)
}
cbStorePut()
sendPacket(client, packet, cb)
})
}
function nop (error) {
debug('nop ::', error)
}
/**
* MqttClient constructor
*
* @param {Stream} stream - stream
* @param {Object} [options] - connection options
* (see Connection#connect)
*/
function MqttClient (streamBuilder, options) {
let k
const that = this
if (!(this instanceof MqttClient)) {
return new MqttClient(streamBuilder, options)
}
this.options = options || {}
// Defaults
for (k in defaultConnectOptions) {
if (typeof this.options[k] === 'undefined') {
this.options[k] = defaultConnectOptions[k]
} else {
this.options[k] = options[k]
}
}
debug('MqttClient :: options.protocol', options.protocol)
debug('MqttClient :: options.protocolVersion', options.protocolVersion)
debug('MqttClient :: options.username', options.username)
debug('MqttClient :: options.keepalive', options.keepalive)
debug('MqttClient :: options.reconnectPeriod', options.reconnectPeriod)
debug('MqttClient :: options.rejectUnauthorized', options.rejectUnauthorized)
debug('MqttClient :: options.topicAliasMaximum', options.topicAliasMaximum)
this.options.clientId = (typeof options.clientId === 'string') ? options.clientId : defaultId()
debug('MqttClient :: clientId', this.options.clientId)
this.options.customHandleAcks = (options.protocolVersion === 5 && options.customHandleAcks) ? options.customHandleAcks : function () { arguments[3](0) }
this.streamBuilder = streamBuilder
this.messageIdProvider = (typeof this.options.messageIdProvider === 'undefined') ? new DefaultMessageIdProvider() : this.options.messageIdProvider
// Inflight message storages
this.outgoingStore = options.outgoingStore || new Store()
this.incomingStore = options.incomingStore || new Store()
// Should QoS zero messages be queued when the connection is broken?
this.queueQoSZero = options.queueQoSZero === undefined ? true : options.queueQoSZero
// map of subscribed topics to support reconnection
this._resubscribeTopics = {}
// map of a subscribe messageId and a topic
this.messageIdToTopic = {}
// Ping timer, setup in _setupPingTimer
this.pingTimer = null
// Is the client connected?
this.connected = false
// Are we disconnecting?
this.disconnecting = false
// Packet queue
this.queue = []
// connack timer
this.connackTimer = null
// Reconnect timer
this.reconnectTimer = null
// Is processing store?
this._storeProcessing = false
// Packet Ids are put into the store during store processing
this._packetIdsDuringStoreProcessing = {}
// Store processing queue
this._storeProcessingQueue = []
// Inflight callbacks
this.outgoing = {}
// True if connection is first time.
this._firstConnection = true
if (options.topicAliasMaximum > 0) {
if (options.topicAliasMaximum > 0xffff) {
debug('MqttClient :: options.topicAliasMaximum is out of range')
} else {
this.topicAliasRecv = new TopicAliasRecv(options.topicAliasMaximum)
}
}
// Send queued packets
this.on('connect', function () {
const queue = this.queue
function deliver () {
const entry = queue.shift()
debug('deliver :: entry %o', entry)
let packet = null
if (!entry) {
that._resubscribe()
return
}
packet = entry.packet
debug('deliver :: call _sendPacket for %o', packet)
let send = true
if (packet.messageId && packet.messageId !== 0) {
if (!that.messageIdProvider.register(packet.messageId)) {
send = false
}
}
if (send) {
that._sendPacket(
packet,
function (err) {
if (entry.cb) {
entry.cb(err)
}
deliver()
}
)
} else {
debug('messageId: %d has already used. The message is skipped and removed.', packet.messageId)
deliver()
}
}
debug('connect :: sending queued packets')
deliver()
})
this.on('close', function () {
debug('close :: connected set to `false`')
this.connected = false
debug('close :: clearing connackTimer')
clearTimeout(this.connackTimer)
debug('close :: clearing ping timer')
if (that.pingTimer !== null) {
that.pingTimer.clear()
that.pingTimer = null
}
if (this.topicAliasRecv) {
this.topicAliasRecv.clear()
}
debug('close :: calling _setupReconnect')
this._setupReconnect()
})
EventEmitter.call(this)
debug('MqttClient :: setting up stream')
this._setupStream()
}
inherits(MqttClient, EventEmitter)
/**
* setup the event handlers in the inner stream.
*
* @api private
*/
MqttClient.prototype._setupStream = function () {
const that = this
const writable = new Writable()
const parser = mqttPacket.parser(this.options)
let completeParse = null
const packets = []
debug('_setupStream :: calling method to clear reconnect')
this._clearReconnect()
debug('_setupStream :: using streamBuilder provided to client to create stream')
this.stream = this.streamBuilder(this)
parser.on('packet', function (packet) {
debug('parser :: on packet push to packets array.')
packets.push(packet)
})
function nextTickWork () {
if (packets.length) {
nextTick(work)
} else {
const done = completeParse
completeParse = null
done()
}
}
function work () {
debug('work :: getting next packet in queue')
const packet = packets.shift()
if (packet) {
debug('work :: packet pulled from queue')
that._handlePacket(packet, nextTickWork)
} else {
debug('work :: no packets in queue')
const done = completeParse
completeParse = null
debug('work :: done flag is %s', !!(done))
if (done) done()
}
}
writable._write = function (buf, enc, done) {
completeParse = done
debug('writable stream :: parsing buffer')
parser.parse(buf)
work()
}
function streamErrorHandler (error) {
debug('streamErrorHandler :: error', error.message)
if (socketErrors.includes(error.code)) {
// handle error
debug('streamErrorHandler :: emitting error')
that.emit('error', error)
} else {
nop(error)
}
}
debug('_setupStream :: pipe stream to writable stream')
this.stream.pipe(writable)
// Suppress connection errors
this.stream.on('error', streamErrorHandler)
// Echo stream close
this.stream.on('close', function () {
debug('(%s)stream :: on close', that.options.clientId)
flushVolatile(that.outgoing)
debug('stream: emit close to MqttClient')
that.emit('close')
})
// Send a connect packet
debug('_setupStream: sending packet `connect`')
const connectPacket = Object.create(this.options)
connectPacket.cmd = 'connect'
if (this.topicAliasRecv) {
if (!connectPacket.properties) {
connectPacket.properties = {}
}
if (this.topicAliasRecv) {
connectPacket.properties.topicAliasMaximum = this.topicAliasRecv.max
}
}
// avoid message queue
sendPacket(this, connectPacket)
// Echo connection errors
parser.on('error', this.emit.bind(this, 'error'))
// auth
if (this.options.properties) {
if (!this.options.properties.authenticationMethod && this.options.properties.authenticationData) {
that.end(() =>
this.emit('error', new Error('Packet has no Authentication Method')
))
return this
}
if (this.options.properties.authenticationMethod && this.options.authPacket && typeof this.options.authPacket === 'object') {
const authPacket = xtend({ cmd: 'auth', reasonCode: 0 }, this.options.authPacket)
sendPacket(this, authPacket)
}
}
// many drain listeners are needed for qos 1 callbacks if the connection is intermittent
this.stream.setMaxListeners(1000)
clearTimeout(this.connackTimer)
this.connackTimer = setTimeout(function () {
debug('!!connectTimeout hit!! Calling _cleanUp with force `true`')
that._cleanUp(true)
}, this.options.connectTimeout)
}
MqttClient.prototype._handlePacket = function (packet, done) {
const options = this.options
if (options.protocolVersion === 5 && options.properties && options.properties.maximumPacketSize && options.properties.maximumPacketSize < packet.length) {
this.emit('error', new Error('exceeding packets size ' + packet.cmd))
this.end({ reasonCode: 149, properties: { reasonString: 'Maximum packet size was exceeded' } })
return this
}
debug('_handlePacket :: emitting packetreceive')
this.emit('packetreceive', packet)
switch (packet.cmd) {
case 'publish':
this._handlePublish(packet, done)
break
case 'puback':
case 'pubrec':
case 'pubcomp':
case 'suback':
case 'unsuback':
this._handleAck(packet)
done()
break
case 'pubrel':
this._handlePubrel(packet, done)
break
case 'connack':
this._handleConnack(packet)
done()
break
case 'auth':
this._handleAuth(packet)
done()
break
case 'pingresp':
this._handlePingresp(packet)
done()
break
case 'disconnect':
this._handleDisconnect(packet)
done()
break
default:
// do nothing
// maybe we should do an error handling
// or just log it
break
}
}
MqttClient.prototype._checkDisconnecting = function (callback) {
if (this.disconnecting) {
if (callback && callback !== nop) {
callback(new Error('client disconnecting'))
} else {
this.emit('error', new Error('client disconnecting'))
}
}
return this.disconnecting
}
/**
* publish - publish <message> to <topic>
*
* @param {String} topic - topic to publish to
* @param {String, Buffer} message - message to publish
* @param {Object} [opts] - publish options, includes:
* {Number} qos - qos level to publish on
* {Boolean} retain - whether or not to retain the message
* {Boolean} dup - whether or not mark a message as duplicate
* {Function} cbStorePut - function(){} called when message is put into `outgoingStore`
* @param {Function} [callback] - function(err){}
* called when publish succeeds or fails
* @returns {MqttClient} this - for chaining
* @api public
*
* @example client.publish('topic', 'message');
* @example
* client.publish('topic', 'message', {qos: 1, retain: true, dup: true});
* @example client.publish('topic', 'message', console.log);
*/
MqttClient.prototype.publish = function (topic, message, opts, callback) {
debug('publish :: message `%s` to topic `%s`', message, topic)
const options = this.options
// .publish(topic, payload, cb);
if (typeof opts === 'function') {
callback = opts
opts = null
}
// default opts
const defaultOpts = { qos: 0, retain: false, dup: false }
opts = xtend(defaultOpts, opts)
if (this._checkDisconnecting(callback)) {
return this
}
const that = this
const publishProc = function () {
let messageId = 0
if (opts.qos === 1 || opts.qos === 2) {
messageId = that._nextId()
if (messageId === null) {
debug('No messageId left')
return false
}
}
const packet = {
cmd: 'publish',
topic: topic,
payload: message,
qos: opts.qos,
retain: opts.retain,
messageId: messageId,
dup: opts.dup
}
if (options.protocolVersion === 5) {
packet.properties = opts.properties
}
debug('publish :: qos', opts.qos)
switch (opts.qos) {
case 1:
case 2:
// Add to callbacks
that.outgoing[packet.messageId] = {
volatile: false,
cb: callback || nop
}
debug('MqttClient:publish: packet cmd: %s', packet.cmd)
that._sendPacket(packet, undefined, opts.cbStorePut)
break
default:
debug('MqttClient:publish: packet cmd: %s', packet.cmd)
that._sendPacket(packet, callback, opts.cbStorePut)
break
}
return true
}
if (this._storeProcessing || this._storeProcessingQueue.length > 0 || !publishProc()) {
this._storeProcessingQueue.push(
{
invoke: publishProc,
cbStorePut: opts.cbStorePut,
callback: callback
}
)
}
return this
}
/**
* subscribe - subscribe to <topic>
*
* @param {String, Array, Object} topic - topic(s) to subscribe to, supports objects in the form {'topic': qos}
* @param {Object} [opts] - optional subscription options, includes:
* {Number} qos - subscribe qos level
* @param {Function} [callback] - function(err, granted){} where:
* {Error} err - subscription error (none at the moment!)
* {Array} granted - array of {topic: 't', qos: 0}
* @returns {MqttClient} this - for chaining
* @api public
* @example client.subscribe('topic');
* @example client.subscribe('topic', {qos: 1});
* @example client.subscribe({'topic': {qos: 0}, 'topic2': {qos: 1}}, console.log);
* @example client.subscribe('topic', console.log);
*/
MqttClient.prototype.subscribe = function () {
const that = this
const args = new Array(arguments.length)
for (let i = 0; i < arguments.length; i++) {
args[i] = arguments[i]
}
const subs = []
let obj = args.shift()
const resubscribe = obj.resubscribe
let callback = args.pop() || nop
let opts = args.pop()
const version = this.options.protocolVersion
delete obj.resubscribe
if (typeof obj === 'string') {
obj = [obj]
}
if (typeof callback !== 'function') {
opts = callback
callback = nop
}
const invalidTopic = validations.validateTopics(obj)
if (invalidTopic !== null) {
setImmediate(callback, new Error('Invalid topic ' + invalidTopic))
return this
}
if (this._checkDisconnecting(callback)) {
debug('subscribe: discconecting true')
return this
}
const defaultOpts = {
qos: 0
}
if (version === 5) {
defaultOpts.nl = false
defaultOpts.rap = false
defaultOpts.rh = 0
}
opts = xtend(defaultOpts, opts)
if (Array.isArray(obj)) {
obj.forEach(function (topic) {
debug('subscribe: array topic %s', topic)
if (!Object.prototype.hasOwnProperty.call(that._resubscribeTopics, topic) ||
that._resubscribeTopics[topic].qos < opts.qos ||
resubscribe) {
const currentOpts = {
topic: topic,
qos: opts.qos
}
if (version === 5) {
currentOpts.nl = opts.nl
currentOpts.rap = opts.rap
currentOpts.rh = opts.rh
currentOpts.properties = opts.properties
}
debug('subscribe: pushing topic `%s` and qos `%s` to subs list', currentOpts.topic, currentOpts.qos)
subs.push(currentOpts)
}
})
} else {
Object
.keys(obj)
.forEach(function (k) {
debug('subscribe: object topic %s', k)
if (!Object.prototype.hasOwnProperty.call(that._resubscribeTopics, k) ||
that._resubscribeTopics[k].qos < obj[k].qos ||
resubscribe) {
const currentOpts = {
topic: k,
qos: obj[k].qos
}
if (version === 5) {
currentOpts.nl = obj[k].nl
currentOpts.rap = obj[k].rap
currentOpts.rh = obj[k].rh
currentOpts.properties = opts.properties
}
debug('subscribe: pushing `%s` to subs list', currentOpts)
subs.push(currentOpts)
}
})
}
if (!subs.length) {
callback(null, [])
return this
}
const subscribeProc = function () {
const messageId = that._nextId()
if (messageId === null) {
debug('No messageId left')
return false
}
const packet = {
cmd: 'subscribe',
subscriptions: subs,
qos: 1,
retain: false,
dup: false,
messageId: messageId
}
if (opts.properties) {
packet.properties = opts.properties
}
// subscriptions to resubscribe to in case of disconnect
if (that.options.resubscribe) {
debug('subscribe :: resubscribe true')
const topics = []
subs.forEach(function (sub) {
if (that.options.reconnectPeriod > 0) {
const topic = { qos: sub.qos }
if (version === 5) {
topic.nl = sub.nl || false
topic.rap = sub.rap || false
topic.rh = sub.rh || 0
topic.properties = sub.properties
}
that._resubscribeTopics[sub.topic] = topic
topics.push(sub.topic)
}
})
that.messageIdToTopic[packet.messageId] = topics
}
that.outgoing[packet.messageId] = {
volatile: true,
cb: function (err, packet) {
if (!err) {
const granted = packet.granted
for (let i = 0; i < granted.length; i += 1) {
subs[i].qos = granted[i]
}
}
callback(err, subs)
}
}
debug('subscribe :: call _sendPacket')
that._sendPacket(packet)
return true
}
if (this._storeProcessing || this._storeProcessingQueue.length > 0 || !subscribeProc()) {
this._storeProcessingQueue.push(
{
invoke: subscribeProc,
callback: callback
}
)
}
return this
}
/**
* unsubscribe - unsubscribe from topic(s)
*
* @param {String, Array} topic - topics to unsubscribe from
* @param {Object} [opts] - optional subscription options, includes:
* {Object} properties - properties of unsubscribe packet
* @param {Function} [callback] - callback fired on unsuback
* @returns {MqttClient} this - for chaining
* @api public
* @example client.unsubscribe('topic');
* @example client.unsubscribe('topic', console.log);
*/
MqttClient.prototype.unsubscribe = function () {
const that = this
const args = new Array(arguments.length)
for (let i = 0; i < arguments.length; i++) {
args[i] = arguments[i]
}
let topic = args.shift()
let callback = args.pop() || nop
let opts = args.pop()
if (typeof topic === 'string') {
topic = [topic]
}
if (typeof callback !== 'function') {
opts = callback
callback = nop
}
const invalidTopic = validations.validateTopics(topic)
if (invalidTopic !== null) {
setImmediate(callback, new Error('Invalid topic ' + invalidTopic))
return this
}
if (that._checkDisconnecting(callback)) {
return this
}
const unsubscribeProc = function () {
const messageId = that._nextId()
if (messageId === null) {
debug('No messageId left')
return false
}
const packet = {
cmd: 'unsubscribe',
qos: 1,
messageId: messageId
}
if (typeof topic === 'string') {
packet.unsubscriptions = [topic]
} else if (Array.isArray(topic)) {
packet.unsubscriptions = topic
}
if (that.options.resubscribe) {
packet.unsubscriptions.forEach(function (topic) {
delete that._resubscribeTopics[topic]
})
}
if (typeof opts === 'object' && opts.properties) {
packet.properties = opts.properties
}
that.outgoing[packet.messageId] = {
volatile: true,
cb: callback
}
debug('unsubscribe: call _sendPacket')
that._sendPacket(packet)
return true
}
if (this._storeProcessing || this._storeProcessingQueue.length > 0 || !unsubscribeProc()) {
this._storeProcessingQueue.push(
{
invoke: unsubscribeProc,
callback: callback
}
)
}
return this
}
/**
* end - close connection
*
* @returns {MqttClient} this - for chaining
* @param {Boolean} force - do not wait for all in-flight messages to be acked
* @param {Object} opts - added to the disconnect packet
* @param {Function} cb - called when the client has been closed
*
* @api public
*/
MqttClient.prototype.end = function (force, opts, cb) {
const that = this
debug('end :: (%s)', this.options.clientId)
if (force == null || typeof force !== 'boolean') {
cb = opts || nop
opts = force
force = false
if (typeof opts !== 'object') {
cb = opts
opts = null
if (typeof cb !== 'function') {
cb = nop
}
}
}
if (typeof opts !== 'object') {
cb = opts
opts = null
}
debug('end :: cb? %s', !!cb)
cb = cb || nop
function closeStores () {
debug('end :: closeStores: closing incoming and outgoing stores')
that.disconnected = true
that.incomingStore.close(function (e1) {
that.outgoingStore.close(function (e2) {
debug('end :: closeStores: emitting end')
that.emit('end')
if (cb) {
const err = e1 || e2
debug('end :: closeStores: invoking callback with args')
cb(err)
}
})
})
if (that._deferredReconnect) {
that._deferredReconnect()
}
}
function finish () {
// defer closesStores of an I/O cycle,
// just to make sure things are
// ok for websockets
debug('end :: (%s) :: finish :: calling _cleanUp with force %s', that.options.clientId, force)
that._cleanUp(force, () => {
debug('end :: finish :: calling process.nextTick on closeStores')
// const boundProcess = nextTick.bind(null, closeStores)
nextTick(closeStores.bind(that))
}, opts)
}
if (this.disconnecting) {
cb()
return this
}
this._clearReconnect()
this.disconnecting = true
if (!force && Object.keys(this.outgoing).length > 0) {
// wait 10ms, just to be sure we received all of it
debug('end :: (%s) :: calling finish in 10ms once outgoing is empty', that.options.clientId)
this.once('outgoingEmpty', setTimeout.bind(null, finish, 10))
} else {
debug('end :: (%s) :: immediately calling finish', that.options.clientId)
finish()
}
return this
}
/**
* removeOutgoingMessage - remove a message in outgoing store
* the outgoing callback will be called withe Error('Message removed') if the message is removed
*
* @param {Number} messageId - messageId to remove message
* @returns {MqttClient} this - for chaining
* @api public
*
* @example client.removeOutgoingMessage(client.getLastAllocated());
*/
MqttClient.prototype.removeOutgoingMessage = function (messageId) {
const cb = this.outgoing[messageId] ? this.outgoing[messageId].cb : null
delete this.outgoing[messageId]
this.outgoingStore.del({ messageId: messageId }, function () {
cb(new Error('Message removed'))
})
return this
}
/**
* reconnect - connect again using the same options as connect()
*
* @param {Object} [opts] - optional reconnect options, includes:
* {Store} incomingStore - a store for the incoming packets
* {Store} outgoingStore - a store for the outgoing packets
* if opts is not given, current stores are used
* @returns {MqttClient} this - for chaining
*
* @api public
*/
MqttClient.prototype.reconnect = function (opts) {
debug('client reconnect')
const that = this
const f = function () {
if (opts) {
that.options.incomingStore = opts.incomingStore
that.options.outgoingStore = opts.outgoingStore
} else {
that.options.incomingStore = null
that.options.outgoingStore = null
}
that.incomingStore = that.options.incomingStore || new Store()
that.outgoingStore = that.options.outgoingStore || new Store()
that.disconnecting = false
that.disconnected = false
that._deferredReconnect = null
that._reconnect()
}
if (this.disconnecting && !this.disconnected) {
this._deferredReconnect = f
} else {
f()
}
return this
}
/**
* _reconnect - implement reconnection
* @api privateish
*/
MqttClient.prototype._reconnect = function () {
debug('_reconnect: emitting reconnect to client')
this.emit('reconnect')
if (this.connected) {
this.end(() => { this._setupStream() })
debug('client already connected. disconnecting first.')
} else {
debug('_reconnect: calling _setupStream')
this._setupStream()
}
}
/**
* _setupReconnect - setup reconnect timer
*/
MqttClient.prototype._setupReconnect = function () {
const that = this
if (!that.disconnecting && !that.reconnectTimer && (that.options.reconnectPeriod > 0)) {
if (!this.reconnecting) {
debug('_setupReconnect :: emit `offline` state')
this.emit('offline')
debug('_setupReconnect :: set `reconnecting` to `true`')
this.reconnecting = true
}
debug('_setupReconnect :: setting reconnectTimer for %d ms', that.options.reconnectPeriod)
that.reconnectTimer = setInterval(function () {
debug('reconnectTimer :: reconnect triggered!')
that._reconnect()
}, that.options.reconnectPeriod)
} else {
debug('_setupReconnect :: doing nothing...')
}
}
/**
* _clearReconnect - clear the reconnect timer
*/
MqttClient.prototype._clearReconnect = function () {
debug('_clearReconnect : clearing reconnect timer')
if (this.reconnectTimer) {
clearInterval(this.reconnectTimer)
this.reconnectTimer = null
}
}
/**
* _cleanUp - clean up on connection end
* @api private
*/
MqttClient.prototype._cleanUp = function (forced, done) {
const opts = arguments[2]
if (done) {
debug('_cleanUp :: done callback provided for on stream close')
this.stream.on('close', done)
}
debug('_cleanUp :: forced? %s', forced)
if (forced) {
if ((this.options.reconnectPeriod === 0) && this.options.clean) {
flush(this.outgoing)
}
debug('_cleanUp :: (%s) :: destroying stream', this.options.clientId)
this.stream.destroy()
} else {
const packet = xtend({ cmd: 'disconnect' }, opts)
debug('_cleanUp :: (%s) :: call _sendPacket with disconnect packet', this.options.clientId)
this._sendPacket(
packet,
setImmediate.bind(
null,
this.stream.end.bind(this.stream)
)
)
}
if (!this.disconnecting) {
debug('_cleanUp :: client not disconnecting. Clearing and resetting reconnect.')
this._clearReconnect()
this._setupReconnect()
}
if (this.pingTimer !== null) {
debug('_cleanUp :: clearing pingTimer')
this.pingTimer.clear()
this.pingTimer = null
}
if (done && !this.connected) {
debug('_cleanUp :: (%s) :: removing stream `done` callback `close` listener', this.options.clientId)
this.stream.removeListener('close', done)
done()
}
}
/**
* _sendPacket - send or queue a packet
* @param {Object} packet - packet options
* @param {Function} cb - callback when the packet is sent
* @param {Function} cbStorePut - called when message is put into outgoingStore
* @api private
*/
MqttClient.prototype._sendPacket = function (packet, cb, cbStorePut) {
debug('_sendPacket :: (%s) :: start', this.options.clientId)
cbStorePut = cbStorePut || nop
cb = cb || nop
const err = applyTopicAlias(this, packet)
if (err) {
cb(err)
return
}
if (!this.connected) {
// allow auth packets to be sent while authenticating with the broker (mqtt5 enhanced auth)
if (packet.cmd === 'auth') {
this._shiftPingInterval()
sendPacket(this, packet, cb)
return
}
debug('_sendPacket :: client not connected. Storing packet offline.')
this._storePacket(packet, cb, cbStorePut)
return
}
// When sending a packet, reschedule the ping timer
this._shiftPingInterval()
switch (packet.cmd) {
case 'publish':
break
case 'pubrel':
storeAndSend(this, packet, cb, cbStorePut)
return
default:
sendPacket(this, packet, cb)
return
}
switch (packet.qos) {
case 2:
case 1:
storeAndSend(this, packet, cb, cbStorePut)
break
/**
* no need of case here since it will be caught by default
* and jshint comply that before default it must be a break
* anyway it will result in -1 evaluation
*/
case 0:
/* falls through */
default:
sendPacket(this, packet, cb)
break
}
debug('_sendPacket :: (%s) :: end', this.options.clientId)
}
/**
* _storePacket - queue a packet
* @param {Object} packet - packet options
* @param {Function} cb - callback when the packet is sent
* @param {Function} cbStorePut - called when message is put into outgoingStore
* @api private
*/
MqttClient.prototype._storePacket = function (packet, cb, cbStorePut) {
debug('_storePacket :: packet: %o', packet)
debug('_storePacket :: cb? %s', !!cb)
cbStorePut = cbStorePut || nop
let storePacket = packet
if (storePacket.cmd === 'publish') {
// The original packet is for sending.
// The cloned storePacket is for storing to resend on reconnect.
// Topic Alias must not be used after disconnected.
storePacket = clone(packet)
const err = removeTopicAliasAndRecoverTopicName(this, storePacket)
if (err) {
return cb && cb(err)
}
}
// check that the packet is not a qos of 0, or that the command is not a publish
if (((storePacket.qos || 0) === 0 && this.queueQoSZero) || storePacket.cmd !== 'publish') {
this.queue.push({ packet: storePacket, cb: cb })
} else if (storePacket.qos > 0) {
cb = this.outgoing[storePacket.messageId] ? this.outgoing[storePacket.messageId].cb : null
this.outgoingStore.put(storePacket, function (err) {
if (err) {
return cb && cb(err)
}
cbStorePut()
})
} else if (cb) {
cb(new Error('No connection to broker'))
}
}
/**
* _setupPingTimer - setup the ping timer
*
* @api private
*/
MqttClient.prototype._setupPingTimer = function () {
debug('_setupPingTimer :: keepalive %d (seconds)', this.options.keepalive)
const that = this
if (!this.pingTimer && this.options.keepalive) {
this.pingResp = true
this.pingTimer = reInterval(function () {
that._checkPing()
}, this.options.keepalive * 1000)
}
}
/**
* _shiftPingInterval - reschedule the ping interval
*
* @api private
*/
MqttClient.prototype._shiftPingInterval = function () {
if (this.pingTimer && this.options.keepalive && this.options.reschedulePings) {
this.pingTimer.reschedule(this.options.keepalive * 1000)
}
}
/**
* _checkPing - check if a pingresp has come back, and ping the server again
*
* @api private
*/
MqttClient.prototype._checkPing = function () {
debug('_checkPing :: checking ping...')
if (this.pingResp) {
debug('_checkPing :: ping response received. Clearing flag and sending `pingreq`')
this.pingResp = false
this._sendPacket({ cmd: 'pingreq' })
} else {
// do a forced cleanup since socket will be in bad shape
debug('_checkPing :: calling _cleanUp with force true')
this._cleanUp(true)
}
}
/**
* _handlePingresp - handle a pingresp
*
* @api private
*/
MqttClient.prototype._handlePingresp = function () {
this.pingResp = true
}
/**
* _handleConnack
*
* @param {Object} packet
* @api private
*/
MqttClient.prototype._handleConnack = function (packet) {
debug('_handleConnack')
const options = this.options
const version = options.protocolVersion
const rc = version === 5 ? packet.reasonCode : packet.returnCode
clearTimeout(this.connackTimer)
delete this.topicAliasSend
if (packet.properties) {
if (packet.properties.topicAliasMaximum) {
if (packet.properties.topicAliasMaximum > 0xffff) {
this.emit('error', new Error('topicAliasMaximum from broker is out of range'))
return
}
if (packet.properties.topicAliasMaximum > 0) {
this.topicAliasSend = new TopicAliasSend(packet.properties.topicAliasMaximum)
}
}
if (packet.properties.serverKeepAlive && options.keepalive) {
options.keepalive = packet.properties.serverKeepAlive
this._shiftPingInterval()
}
if (packet.properties.maximumPacketSize) {
if (!options.properties) { options.properties = {} }
options.properties.maximumPacketSize = packet.properties.maximumPacketSize
}
}
if (rc === 0) {
this.reconnecting = false
this._onConnect(packet)
} else if (rc > 0) {
const err = new Error('Connection refused: ' + errors[rc])
err.code = rc
this.emit('error', err)
}
}
MqttClient.prototype._handleAuth = function (packet) {
const options = this.options
const version = options.protocolVersion
const rc = version === 5 ? packet.reasonCode : packet.returnCode
if (version !== 5) {
const err = new Error('Protocol error: Auth packets are only supported in MQTT 5. Your version:' + version)
err.code = rc
this.emit('error', err)
return
}
const that = this
this.handleAuth(packet, function (err, packet) {
if (err) {
that.emit('error', err)
return
}
if (rc === 24) {
that.reconnecting = false
that._sendPacket(packet)
} else {
const error = new Error('Connection refused: ' + errors[rc])
err.code = rc
that.emit('error', error)
}
})
}
/**
* @param packet the packet received by the broker
* @return the auth packet to be returned to the broker
* @api public
*/
MqttClient.prototype.handleAuth = function (packet, callback) {
callback()
}
/**
* _handlePublish
*
* @param {Object} packet
* @api private
*/
/*
those late 2 case should be rewrite to comply with coding style:
case 1:
case 0:
// do not wait sending a puback
// no callback passed
if (1 === qos) {
this._sendPacket({
cmd: 'puback',
messageId: messageId
});
}
// emit the message event for both qos 1 and 0
this.emit('message', topic, message, packet);
this.handleMessage(packet, done);
break;
default:
// do nothing but every switch mus have a default
// log or throw an error about unknown qos
break;
for now i just suppressed the warnings
*/
MqttClient.prototype._handlePublish = function (packet, done) {
debug('_handlePublish: packet %o', packet)
done = typeof done !== 'undefined' ? done : nop
let topic = packet.topic.toString()
const message = packet.payload
const qos = packet.qos
const messageId = packet.messageId
const that = this
const options = this.options
const validReasonCodes = [0, 16, 128, 131, 135, 144, 145, 151, 153]
if (this.options.protocolVersion === 5) {
let alias
if (packet.properties) {
alias = packet.properties.topicAlias
}
if (typeof alias !== 'undefined') {
if (topic.length === 0) {
if (alias > 0 && alias <= 0xffff) {
const gotTopic = this.topicAliasRecv.getTopicByAlias(alias)
if (gotTopic) {
topic = gotTopic
debug('_handlePublish :: topic complemented by alias. topic: %s - alias: %d', topic, alias)
} else {
debug('_handlePublish :: unregistered topic alias. alias: %d', alias)
this.emit('error', new Error('Received unregistered Topic Alias'))
return
}
} else {
debug('_handlePublish :: topic alias out of range. alias: %d', alias)
this.emit('error', new Error('Received Topic Alias is out of range'))
return
}
} else {
if (this.topicAliasRecv.put(topic, alias)) {
debug('_handlePublish :: registered topic: %s - alias: %d', topic, alias)
} else {
debug('_handlePublish :: topic alias out of range. alias: %d', alias)
this.emit('error', new Error('Received Topic Alias is out of range'))
return
}
}
}
}
debug('_handlePublish: qos %d', qos)
switch (qos) {
case 2: {
options.customHandleAcks(topic, message, packet, function (error, code) {
if (!(error instanceof Error)) {
code = error
error = null
}
if (error) { return that.emit('error', error) }
if (validReasonCodes.indexOf(code) === -1) { return that.emit('error', new Error('Wrong reason code for pubrec')) }
if (code) {
that._sendPacket({ cmd: 'pubrec', messageId: messageId, reasonCode: code }, done)
} else {
that.incomingStore.put(packet, function () {
that._sendPacket({ cmd: 'pubrec', messageId: messageId }, done)
})
}
})
break
}
case 1: {
// emit the message event
options.customHandleAcks(topic, message, packet, function (error, code) {
if (!(error instanceof Error)) {
code = error
error = null
}
if (error) { return that.emit('error', error) }
if (validReasonCodes.indexOf(code) === -1) { return that.emit('error', new Error('Wrong reason code for puback')) }
if (!code) { that.emit('message', topic, message, packet) }
that.handleMessage(packet, function (err) {
if (err) {
return done && done(err)
}
that._sendPacket({ cmd: 'puback', messageId: messageId, reasonCode: code }, done)
})
})
break
}
case 0:
// emit the message event
this.emit('message', topic, message, packet)
this.handleMessage(packet, done)
break
default:
// do nothing
debug('_handlePublish: unknown QoS. Doing nothing.')
// log or throw an error about unknown qos
break
}
}
/**
* Handle messages with backpressure support, one at a time.
* Override at will.
*
* @param Packet packet the packet
* @param Function callback call when finished
* @api public
*/
MqttClient.prototype.handleMessage = function (packet, callback) {
callback()
}
/**
* _handleAck
*
* @param {Object} packet
* @api private
*/
MqttClient.prototype._handleAck = function (packet) {
/* eslint no-fallthrough: "off" */
const messageId = packet.messageId
const type = packet.cmd
let response = null
const cb = this.outgoing[messageId] ? this.outgoing[messageId].cb : null
const that = this
let err
// Checking `!cb` happens to work, but it's not technically "correct".
//
// Why? This code assumes that "no callback" is the same as that "we're not
// waiting for responses" (puback, pubrec, pubcomp, suback, or unsuback).
//
// It would be better to check `if (!this.outgoing[messageId])` here, but
// there's no reason to change it and risk (another) regression.
//
// The only reason this code works is becaues code in MqttClient.publish,
// MqttClinet.subscribe, and MqttClient.unsubscribe ensures that we will
// have a callback even if the user doesn't pass one in.)
if (!cb) {
debug('_handleAck :: Server sent an ack in error. Ignoring.')
// Server sent an ack in error, ignore it.
return
}
// Process
debug('_handleAck :: packet type', type)
switch (type) {
case 'pubcomp':
// same thing as puback for QoS 2
case 'puback': {
const pubackRC = packet.reasonCode
// Callback - we're done
if (pubackRC && pubackRC > 0 && pubackRC !== 16) {
err = new Error('Publish error: ' + errors[pubackRC])
err.code = pubackRC
cb(err, packet)
}
delete this.outgoing[messageId]
this.outgoingStore.del(packet, cb)
this.messageIdProvider.deallocate(messageId)
this._invokeStoreProcessingQueue()
break
}
case 'pubrec': {
response = {
cmd: 'pubrel',
qos: 2,
messageId: messageId
}
const pubrecRC = packet.reasonCode
if (pubrecRC && pubrecRC > 0 && pubrecRC !== 16) {
err = new Error('Publish error: ' + errors[pubrecRC])
err.code = pubrecRC
cb(err, packet)
} else {
this._sendPacket(response)
}
break
}
case 'suback': {
delete this.outgoing[messageId]
this.messageIdProvider.deallocate(messageId)
for (let grantedI = 0; grantedI < packet.granted.length; grantedI++) {
if ((packet.granted[grantedI] & 0x80) !== 0) {
// suback with Failure status
const topics = this.messageIdToTopic[messageId]
if (topics) {
topics.forEach(function (topic) {
delete that._resubscribeTopics[topic]
})
}
}
}
this._invokeStoreProcessingQueue()
cb(null, packet)
break
}
case 'unsuback': {
delete this.outgoing[messageId]
this.messageIdProvider.deallocate(messageId)
this._invokeStoreProcessingQueue()
cb(null)
break
}
default:
that.emit('error', new Error('unrecognized packet type'))
}
if (this.disconnecting &&
Object.keys(this.outgoing).length === 0) {
this.emit('outgoingEmpty')
}
}
/**
* _handlePubrel
*
* @param {Object} packet
* @api private
*/
MqttClient.prototype._handlePubrel = function (packet, callback) {
debug('handling pubrel packet')
callback = typeof callback !== 'undefined' ? callback : nop
const messageId = packet.messageId
const that = this
const comp = { cmd: 'pubcomp', messageId: messageId }
that.incomingStore.get(packet, function (err, pub) {
if (!err) {
that.emit('message', pub.topic, pub.payload, pub)
that.handleMessage(pub, function (err) {
if (err) {
return callback(err)
}
that.incomingStore.del(pub, nop)
that._sendPacket(comp, callback)
})
} else {
that._sendPacket(comp, callback)
}
})
}
/**
* _handleDisconnect
*
* @param {Object} packet
* @api private
*/
MqttClient.prototype._handleDisconnect = function (packet) {
this.emit('disconnect', packet)
}
/**
* _nextId
* @return unsigned int
*/
MqttClient.prototype._nextId = function () {
return this.messageIdProvider.allocate()
}
/**
* getLastMessageId
* @return unsigned int
*/
MqttClient.prototype.getLastMessageId = function () {
return this.messageIdProvider.getLastAllocated()
}
/**
* _resubscribe
* @api private
*/
MqttClient.prototype._resubscribe = function () {
debug('_resubscribe')
const _resubscribeTopicsKeys = Object.keys(this._resubscribeTopics)
if (!this._firstConnection &&
(this.options.clean || (this.options.protocolVersion === 5 && !this.connackPacket.sessionPresent)) &&
_resubscribeTopicsKeys.length > 0) {
if (this.options.resubscribe) {
if (this.options.protocolVersion === 5) {
debug('_resubscribe: protocolVersion 5')
for (let topicI = 0; topicI < _resubscribeTopicsKeys.length; topicI++) {
const resubscribeTopic = {}
resubscribeTopic[_resubscribeTopicsKeys[topicI]] = this._resubscribeTopics[_resubscribeTopicsKeys[topicI]]
resubscribeTopic.resubscribe = true
this.subscribe(resubscribeTopic, { properties: resubscribeTopic[_resubscribeTopicsKeys[topicI]].properties })
}
} else {
this._resubscribeTopics.resubscribe = true
this.subscribe(this._resubscribeTopics)
}
} else {
this._resubscribeTopics = {}
}
}
this._firstConnection = false
}
/**
* _onConnect
*
* @api private
*/
MqttClient.prototype._onConnect = function (packet) {
if (this.disconnected) {
this.emit('connect', packet)
return
}
const that = this
this.connackPacket = packet
this.messageIdProvider.clear()
this._setupPingTimer()
this.connected = true
function startStreamProcess () {
let outStore = that.outgoingStore.createStream()
function clearStoreProcessing () {
that._storeProcessing = false
that._packetIdsDuringStoreProcessing = {}
}
that.once('close', remove)
outStore.on('error', function (err) {
clearStoreProcessing()
that._flushStoreProcessingQueue()
that.removeListener('close', remove)
that.emit('error', err)
})
function remove () {
outStore.destroy()
outStore = null
that._flushStoreProcessingQueue()
clearStoreProcessing()
}
function storeDeliver () {
// edge case, we wrapped this twice
if (!outStore) {
return
}
that._storeProcessing = true
const packet = outStore.read(1)
let cb
if (!packet) {
// read when data is available in the future
outStore.once('readable', storeDeliver)
return
}
// Skip already processed store packets
if (that._packetIdsDuringStoreProcessing[packet.messageId]) {
storeDeliver()
return
}
// Avoid unnecessary stream read operations when disconnected
if (!that.disconnecting && !that.reconnectTimer) {
cb = that.outgoing[packet.messageId] ? that.outgoing[packet.messageId].cb : null
that.outgoing[packet.messageId] = {
volatile: false,
cb: function (err, status) {
// Ensure that the original callback passed in to publish gets invoked
if (cb) {
cb(err, status)
}
storeDeliver()
}
}
that._packetIdsDuringStoreProcessing[packet.messageId] = true
if (that.messageIdProvider.register(packet.messageId)) {
that._sendPacket(packet)
} else {
debug('messageId: %d has already used.', packet.messageId)
}
} else if (outStore.destroy) {
outStore.destroy()
}
}
outStore.on('end', function () {
let allProcessed = true
for (const id in that._packetIdsDuringStoreProcessing) {
if (!that._packetIdsDuringStoreProcessing[id]) {
allProcessed = false
break
}
}
if (allProcessed) {
clearStoreProcessing()
that.removeListener('close', remove)
that._invokeAllStoreProcessingQueue()
that.emit('connect', packet)
} else {
startStreamProcess()
}
})
storeDeliver()
}
// start flowing
startStreamProcess()
}
MqttClient.prototype._invokeStoreProcessingQueue = function () {
if (this._storeProcessingQueue.length > 0) {
const f = this._storeProcessingQueue[0]
if (f && f.invoke()) {
this._storeProcessingQueue.shift()
return true
}
}
return false
}
MqttClient.prototype._invokeAllStoreProcessingQueue = function () {
while (this._invokeStoreProcessingQueue()) { /* empty */ }
}
MqttClient.prototype._flushStoreProcessingQueue = function () {
for (const f of this._storeProcessingQueue) {
if (f.cbStorePut) f.cbStorePut(new Error('Connection closed'))
if (f.callback) f.callback(new Error('Connection closed'))
}
this._storeProcessingQueue.splice(0)
}
module.exports = MqttClient