2314212754

Build Type

github

Committed by GitHub

Commit Message

Merge fe405620f into fa0c314d4

Pull Request Pull Request #76: chore: move to es6, bump deps and drop nodejs 12

Run Details

108 of 149 branches covered (72.48%)

Branch coverage included in aggregate %.

231 of 270 new or added lines in 1 file covered. (85.56%)

5 existing lines in 1 file now uncovered.

242 of 286 relevant lines covered (84.62%)

183.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.46

/persistence.js

'use strict'

const urlModule = require('native-url')
const escape = require('escape-string-regexp')
const CachedPersistence = require('aedes-cached-persistence')
const Packet = CachedPersistence.Packet
const mongodb = require('mongodb')
const pump = require('pump')
const through = require('through2')
const parallel = require('fastparallel')()
const { Qlobber } = require('qlobber')
const qlobberOpts = {
  separator: '/',
  wildcard_one: '+',
  wildcard_some: '#',
  match_empty_levels: true
}

function toStream (op) {
  return op.stream ? op.stream() : op
}

class MongoPersistence extends CachedPersistence {
  constructor (opts = {}) {
    super(opts)

    opts.ttl = opts.ttl || {}

    if (typeof opts.ttl.packets === 'number') {
      opts.ttl.packets = {
        retained: opts.ttl.packets,
        will: opts.ttl.packets,
        outgoing: opts.ttl.packets,
        incoming: opts.ttl.packets
      }
    }

    this._opts = opts
    this._db = null
    this._cl = null
    this.packetsQueue = [] // used for storing retained packets with ordered bulks
    this.executing = false // used as lock while a bulk is executing

    CachedPersistence.call(this, opts)
  }

  _connect (cb) {
    if (this._opts.db) {
      cb(null, this._opts.db)
      return
    }

    const conn = this._opts.url || 'mongodb://127.0.0.1/aedes'

    const options = { useNewUrlParser: true, useUnifiedTopology: true }

    if (this._opts.mongoOptions) {
      Object.assign(options, this._opts.mongoOptions)
    }

    mongodb.MongoClient.connect(conn, options, cb)
  }

  _setup () {
    if (this.ready) {
      return
    }

    const that = this

    this._connect(function (err, client) {
      if (err) {
        that.emit('error', err)
        return
      }

      that._client = client

      let db
      if (that._opts.db) {
        db = that._opts.db
      } else {
        const urlParsed = urlModule.parse(that._opts.url)
        const databaseName = that._opts.database || (urlParsed.pathname ? urlParsed.pathname.substr(1) : undefined)
        db = that._db = client.db(databaseName)
      }

      const subscriptions = db.collection('subscriptions')
      const retained = db.collection('retained')
      const will = db.collection('will')
      const outgoing = db.collection('outgoing')
      const incoming = db.collection('incoming')

      that._cl = {
        subscriptions,
        retained,
        will,
        outgoing,
        incoming
      }

      function initCollections () {
        function finishInit () {
          toStream(subscriptions.find({
            qos: { $gte: 0 }
          })).on('data', function (chunk) {
            that._trie.add(chunk.topic, chunk)
          }).on('end', function () {
            that.emit('ready')
          }).on('error', function (err) {
            that.emit('error', err)
          })
        }

        function createIndex (opts, cb) {
          const indexOpts = { name: opts.name }
          if (typeof opts.expireAfterSeconds === 'number') {
            indexOpts.expireAfterSeconds = opts.expireAfterSeconds
          }

          return this._cl[opts.collection].createIndex(opts.key, indexOpts, cb)
        }

        const indexes = [
          {
            collection: 'outgoing',
            key: { clientId: 1, 'packet.brokerId': 1, 'packet.brokerCounter': 1 },
            name: 'query_clientId_brokerId'
          },
          {
            collection: 'outgoing',
            key: { clientId: 1, 'packet.messageId': 1 },
            name: 'query_clientId_messageId'
          },
          {
            collection: 'incoming',
            key: { clientId: 1, 'packet.brokerId': 1, 'packet.brokerCounter': 1 },
            name: 'query_clientId_brokerId'
          },
          {
            collection: 'incoming',
            key: { clientId: 1, 'packet.messageId': 1 },
            name: 'query_clientId_messageId'
          }
        ]

        if (that._opts.ttl.subscriptions >= 0) {
          indexes.push({
            collection: 'subscriptions',
            key: 'added',
            name: 'ttl',
            expireAfterSeconds: that._opts.ttl.subscriptions
          })
        }

        if (that._opts.ttl.packets) {
          if (that._opts.ttl.packets.retained >= 0) {
            indexes.push({
              collection: 'retained',
              key: 'added',
              name: 'ttl',
              expireAfterSeconds: that._opts.ttl.packets.retained
            })
          }

          if (that._opts.ttl.packets.will >= 0) {
            indexes.push({
              collection: 'will',
              key: 'packet.added',
              name: 'ttl',
              expireAfterSeconds: that._opts.ttl.packets.will
            })
          }

          if (that._opts.ttl.packets.outgoing >= 0) {
            indexes.push({
              collection: 'outgoing',
              key: 'packet.added',
              name: 'ttl',
              expireAfterSeconds: that._opts.ttl.packets.outgoing
            })
          }

          if (that._opts.ttl.packets.incoming >= 0) {
            indexes.push({
              collection: 'incoming',
              key: 'packet.added',
              name: 'ttl',
              expireAfterSeconds: that._opts.ttl.packets.incoming
            })
          }
        }

        parallel(that, createIndex, indexes, finishInit)
      }

      // drop existing indexes (if exists)
      if (that._opts.dropExistingIndexes) {
        that._dropIndexes(db, initCollections)
      } else {
        initCollections()
      }
    })
  }

  _dropIndexes (db, cb) {
    db.collections(function (err, collections) {
      if (err) { throw err }

      let done = 0

      function finish (err) {
        if (err) { throw err }

        done++
        if (done >= collections.length && typeof cb === 'function') {
          cb()
        }
      }

      if (collections.length === 0) {
        finish()
      } else {
        for (let i = 0; i < collections.length; i++) {
          collections[i].indexExists('ttl', function (err, exists) {
            if (err) { throw err }

            if (exists) {
              collections[i].dropIndex('ttl', finish)
            } else {
              finish()
            }
          })
        }
      }
    })
  }

  storeRetained (packet, cb) {
    if (!this.ready) {
      this.once('ready', this.storeRetained.bind(this, packet, cb))
      return
    }

    this.packetsQueue.push({ packet, cb })
    executeBulk(this)
  }

  createRetainedStream (pattern) {
    return this.createRetainedStreamCombi([pattern])
  }

  createRetainedStreamCombi (patterns) {
    let regex = []

    const instance = through.obj(filterPattern)
    instance.matcher = new Qlobber(qlobberOpts)

    for (let i = 0; i < patterns.length; i++) {
      instance.matcher.add(patterns[i], true)
      regex.push(escape(patterns[i]).replace(/(\/*#|\\\+).*$/, ''))
    }

    regex = regex.join('|')

    return pump(
      toStream(this._cl.retained.find({
        topic: new RegExp(regex)
      })),
      instance
    )
  }

  addSubscriptions (client, subs, cb) {
    if (!this.ready) {
      this.once('ready', this.addSubscriptions.bind(this, client, subs, cb))
      return
    }

    const that = this
    let published = 0
    let errored = false
    const bulk = this._cl.subscriptions.initializeOrderedBulkOp()
    subs
      .forEach(function (sub) {
        const subscription = {
          clientId: client.id,
          topic: sub.topic,
          qos: sub.qos
        }
        bulk.find({
          clientId: client.id,
          topic: sub.topic
        }).upsert().updateOne({
          $set: decorateSubscription(subscription, that._opts)
        })
      })

    bulk.execute(finish)
    this._addedSubscriptions(client, subs, finish)

    function finish (err) {
      errored = err
      published++
      if (published === 2) {
        cb(errored, client)
      }
    }
  }

  removeSubscriptions (client, subs, cb) {
    if (!this.ready) {
      this.once('ready', this.removeSubscriptions.bind(this, client, subs, cb))
      return
    }

    let published = 0
    let errored = false
    const bulk = this._cl.subscriptions.initializeOrderedBulkOp()
    subs
      .forEach(function (topic) {
        bulk.find({
          clientId: client.id,
          topic
        }).deleteOne()
      })

    bulk.execute(finish)
    this._removedSubscriptions(client, subs.map(toSub), finish)

    function finish (err) {
      if (err && !errored) {
        errored = true
        cb(err, client)
        return
      }
      published++
      if (published === 2 && !errored) {
        cb(null, client)
      }
    }
  }

  subscriptionsByClient (client, cb) {
    if (!this.ready) {
      this.once('ready', this.subscriptionsByClient.bind(this, client, cb))
      return
    }

    this._cl.subscriptions.find({ clientId: client.id }).toArray(function (err, subs) {
      if (err) {
        cb(err)
        return
      }

      const toReturn = subs.map(function (sub) {
        return {
          topic: sub.topic,
          qos: sub.qos
        }
      })

      cb(null, toReturn.length > 0 ? toReturn : null, client)
    })
  }

  countOffline (cb) {
    let clientsCount = 0
    const that = this
    toStream(this._cl.subscriptions.aggregate([{
      $group: {
        _id: '$clientId'
      }
    }])).on('data', function () {
      clientsCount++
    }).on('end', function () {
      cb(null, that._trie.subscriptionsCount, clientsCount)
    }).on('error', cb)
  }

  destroy (cb) {
    if (!this.ready) {
      this.once('ready', this.destroy.bind(this, cb))
      return
    }

    if (this._destroyed) {
      throw new Error('destroyed called twice!')
    }

    this._destroyed = true

    cb = cb || noop

    if (this._opts.db) {
      cb()
    } else {
      this._client.close(function () {
        // swallow err in case of close
        cb()
      })
    }
  }

  outgoingEnqueue (sub, packet, cb) {
    this.outgoingEnqueueCombi([sub], packet, cb)
  }

  outgoingEnqueueCombi (subs, packet, cb) {
    if (!this.ready) {
      this.once('ready', this.outgoingEnqueueCombi.bind(this, subs, packet, cb))
      return
    }

    if (!subs || subs.length === 0) {
      return cb(null, packet)
    }

    const newp = new Packet(packet)
    const opts = this._opts

    function createPacket (sub) {
      return {
        clientId: sub.clientId,
        packet: decoratePacket(newp, opts)
      }
    }

    this._cl.outgoing.insertMany(subs.map(createPacket), function (err) {
      cb(err, packet)
    })
  }

  outgoingStream (client) {
    return pump(
      toStream(this._cl.outgoing.find({ clientId: client.id })),
      through.obj(asPacket))
  }

  outgoingUpdate (client, packet, cb) {
    if (!this.ready) {
      this.once('ready', this.outgoingUpdate.bind(this, client, packet, cb))
      return
    }
    if (packet.brokerId) {
      updateWithMessageId(this._cl, client, packet, cb)
    } else {
      updatePacket(this._cl, client, packet, cb)
    }
  }

  outgoingClearMessageId (client, packet, cb) {
    if (!this.ready) {
      this.once('ready', this.outgoingClearMessageId.bind(this, client, packet, cb))
      return
    }

    const outgoing = this._cl.outgoing

    outgoing.findOne({
      clientId: client.id,
      'packet.messageId': packet.messageId
    }, function (err, p) {
      if (err) {
        return cb(err)
      }

      if (!p) {
        return cb(null)
      }

      outgoing.deleteOne({
        clientId: client.id,
        'packet.messageId': packet.messageId
      }, function (err) {
        cb(err, p.packet)
      })
    })
  }

  incomingStorePacket (client, packet, cb) {
    if (!this.ready) {
      this.once('ready', this.incomingStorePacket.bind(this, client, packet, cb))
      return
    }

    const newp = new Packet(packet)
    newp.messageId = packet.messageId

    this._cl.incoming.insertOne({
      clientId: client.id,
      packet: decoratePacket(newp, this._opts)
    }, cb)
  }

  incomingGetPacket (client, packet, cb) {
    if (!this.ready) {
      this.once('ready', this.incomingGetPacket.bind(this, client, packet, cb))
      return
    }

    this._cl.incoming.findOne({
      clientId: client.id,
      'packet.messageId': packet.messageId
    }, function (err, result) {
      if (err) {
        cb(err)
        return
      }

      if (!result) {
        cb(new Error('packet not found'), null, client)
        return
      }

      const packet = result.packet

      if (packet && packet.payload) {
        packet.payload = packet.payload.buffer
      }

      cb(null, packet, client)
    })
  }

  incomingDelPacket (client, packet, cb) {
    if (!this.ready) {
      this.once('ready', this.incomingDelPacket.bind(this, client, packet, cb))
      return
    }

    this._cl.incoming.deleteOne({
      clientId: client.id,
      'packet.messageId': packet.messageId
    }, cb)
  }

  putWill (client, packet, cb) {
    if (!this.ready) {
      this.once('ready', this.putWill.bind(this, client, packet, cb))
      return
    }

    packet.clientId = client.id
    packet.brokerId = this.broker.id
    this._cl.will.insertOne({
      clientId: client.id,
      packet: decoratePacket(packet, this._opts)
    }, function (err) {
      cb(err, client)
    })
  }

  getWill (client, cb) {
    this._cl.will.findOne({
      clientId: client.id
    }, function (err, result) {
      if (err) {
        cb(err)
        return
      }

      if (!result) {
        cb(null, null, client)
        return
      }

      const packet = result.packet

      if (packet && packet.payload) {
        packet.payload = packet.payload.buffer
      }

      cb(null, packet, client)
    })
  }

  delWill (client, cb) {
    const will = this._cl.will
    this.getWill(client, function (err, packet) {
      if (err || !packet) {
        cb(err, null, client)
        return
      }
      will.deleteOne({
        clientId: client.id
      }, function (err) {
        cb(err, packet, client)
      })
    })
  }

  streamWill (brokers) {
    const query = {}

    if (brokers) {
      query['packet.brokerId'] = { $nin: Object.keys(brokers) }
    }
    return pump(toStream(this._cl.will.find(query)), through.obj(asPacket))
  }

  getClientList (topic) {
    const query = {}

    if (topic) {
      query.topic = topic
    }

    return pump(toStream(this._cl.subscriptions.find(query)), through.obj(function asPacket (obj, enc, cb) {
      this.push(obj.clientId)
      cb()
    }))
  }
}

function decoratePacket (packet, opts) {
  if (opts.ttl.packets) {
    packet.added = new Date()
  }
  return packet
}

function executeBulk (that) {
  if (!that.executing && that.packetsQueue.length > 0) {
    that.executing = true
    const bulk = that._cl.retained.initializeOrderedBulkOp()
    const onEnd = []

    while (that.packetsQueue.length) {
      const p = that.packetsQueue.shift()
      onEnd.push(p.cb)
      const criteria = { topic: p.packet.topic }

      if (p.packet.payload.length > 0) {
        bulk.find(criteria).upsert().updateOne({
          $set: decoratePacket(p.packet, that._opts)
        })
      } else {
        bulk.find(criteria).deleteOne()
      }
    }

    bulk.execute(function () {
      while (onEnd.length) onEnd.shift().call()
      that.executing = false
      executeBulk(that)
    })
  }
}

function filterPattern (chunk, enc, cb) {
  if (this.matcher.match(chunk.topic).length > 0) {
    chunk.payload = chunk.payload.buffer
    // this is converting chunk to slow properties
    // https://github.com/sindresorhus/to-fast-properties
    // might make this faster
    delete chunk._id
    this.push(chunk)
  }
  cb()
}

function decorateSubscription (sub, opts) {
  if (opts.ttl.subscriptions) {
    sub.added = new Date()
  }
  return sub
}

function toSub (topic) {
  return {
    topic
  }
}

function asPacket (obj, enc, cb) {
  const packet = obj.packet

  if (packet.payload) {
    const buffer = packet.payload.buffer
    if (buffer && Buffer.isBuffer(buffer)) {
      packet.payload = packet.payload.buffer
    }
  }
  cb(null, packet)
}

function updateWithMessageId (db, client, packet, cb) {
  db.outgoing.updateOne({
    clientId: client.id,
    'packet.brokerCounter': packet.brokerCounter,
    'packet.brokerId': packet.brokerId
  }, {
    $set: {
      'packet.messageId': packet.messageId
    }
  }, function (err) {
    cb(err, client, packet)
  })
}

function updatePacket (db, client, packet, cb) {
  db.outgoing.updateOne({
    clientId: client.id,
    'packet.messageId': packet.messageId
  }, {
    $set: {
      clientId: client.id,
      packet
    }
  }, function (err) {
    cb(err, client, packet)
  })
}

function noop () { }

module.exports = (opts) => new MongoPersistence(opts)

1	'use strict'
2
3	const urlModule = require('native-url')	3✔
4	const escape = require('escape-string-regexp')	3✔
5	const CachedPersistence = require('aedes-cached-persistence')	3✔
6	const Packet = CachedPersistence.Packet	3✔
7	const mongodb = require('mongodb')	3✔
8	const pump = require('pump')	3✔
9	const through = require('through2')	3✔
10	const parallel = require('fastparallel')()	3✔
11	const { Qlobber } = require('qlobber')	3✔
12	const qlobberOpts = {	3✔
13	separator: '/',
14	wildcard_one: '+',
15	wildcard_some: '#',
16	match_empty_levels: true
17	}
18
19	function toStream (op) {
20	return op.stream ? op.stream() : op	300!
21	}
22
23	class MongoPersistence extends CachedPersistence {
24	constructor (opts = {}) {	×
25	super(opts)	168✔
26
27	opts.ttl = opts.ttl \|\| {}	168✔
28
29	if (typeof opts.ttl.packets === 'number') {	168✔
30	opts.ttl.packets = {	15✔
31	retained: opts.ttl.packets,
32	will: opts.ttl.packets,
33	outgoing: opts.ttl.packets,
34	incoming: opts.ttl.packets
35	}
36	}
37
38	this._opts = opts	168✔
39	this._db = null	168✔
40	this._cl = null	168✔
41	this.packetsQueue = [] // used for storing retained packets with ordered bulks	168✔
42	this.executing = false // used as lock while a bulk is executing	168✔
43
44	CachedPersistence.call(this, opts)	168✔
45	}
46
47	_connect (cb) {
48	if (this._opts.db) {	168✔
49	cb(null, this._opts.db)	12✔
50	return	12✔
51	}
52
53	const conn = this._opts.url \|\| 'mongodb://127.0.0.1/aedes'	156!
54
55	const options = { useNewUrlParser: true, useUnifiedTopology: true }	156✔
56
57	if (this._opts.mongoOptions) {	156✔
58	Object.assign(options, this._opts.mongoOptions)	3✔
59	}
60
61	mongodb.MongoClient.connect(conn, options, cb)	156✔
62	}
63
64	_setup () {
65	if (this.ready) {	165!
UNCOV 66	return	×
67	}
68
69	const that = this	165✔
70
71	this._connect(function (err, client) {	165✔
72	if (err) {	165!
NEW 73	that.emit('error', err)	×
NEW 74	return	×
75	}
76
77	that._client = client	165✔
78
79	let db
80	if (that._opts.db) {	165✔
81	db = that._opts.db	12✔
82	} else {
83	const urlParsed = urlModule.parse(that._opts.url)	153✔
84	const databaseName = that._opts.database \|\| (urlParsed.pathname ? urlParsed.pathname.substr(1) : undefined)	153!
85	db = that._db = client.db(databaseName)	153✔
86	}
87
88	const subscriptions = db.collection('subscriptions')	165✔
89	const retained = db.collection('retained')	165✔
90	const will = db.collection('will')	165✔
91	const outgoing = db.collection('outgoing')	165✔
92	const incoming = db.collection('incoming')	165✔
93
94	that._cl = {	165✔
95	subscriptions,
96	retained,
97	will,
98	outgoing,
99	incoming
100	}
101
102	function initCollections () {
103	function finishInit () {
104	toStream(subscriptions.find({	165✔
105	qos: { $gte: 0 }
106	})).on('data', function (chunk) {
107	that._trie.add(chunk.topic, chunk)	3✔
108	}).on('end', function () {
109	that.emit('ready')	165✔
110	}).on('error', function (err) {
NEW 111	that.emit('error', err)	×
112	})
113	}
114
115	function createIndex (opts, cb) {
116	const indexOpts = { name: opts.name }	735✔
117	if (typeof opts.expireAfterSeconds === 'number') {	735✔
118	indexOpts.expireAfterSeconds = opts.expireAfterSeconds	75✔
119	}
120
121	return this._cl[opts.collection].createIndex(opts.key, indexOpts, cb)	735✔
122	}
123
124	const indexes = [	165✔
125	{
126	collection: 'outgoing',
127	key: { clientId: 1, 'packet.brokerId': 1, 'packet.brokerCounter': 1 },
128	name: 'query_clientId_brokerId'
129	},
130	{
131	collection: 'outgoing',
132	key: { clientId: 1, 'packet.messageId': 1 },
133	name: 'query_clientId_messageId'
134	},
135	{
136	collection: 'incoming',
137	key: { clientId: 1, 'packet.brokerId': 1, 'packet.brokerCounter': 1 },
138	name: 'query_clientId_brokerId'
139	},
140	{
141	collection: 'incoming',
142	key: { clientId: 1, 'packet.messageId': 1 },
143	name: 'query_clientId_messageId'
144	}
145	]
146
147	if (that._opts.ttl.subscriptions >= 0) {	165✔
148	indexes.push({	15✔
149	collection: 'subscriptions',
150	key: 'added',
151	name: 'ttl',
152	expireAfterSeconds: that._opts.ttl.subscriptions
153	})
154	}
155
156	if (that._opts.ttl.packets) {	165✔
157	if (that._opts.ttl.packets.retained >= 0) {	15!
158	indexes.push({	15✔
159	collection: 'retained',
160	key: 'added',
161	name: 'ttl',
162	expireAfterSeconds: that._opts.ttl.packets.retained
163	})
164	}
165
166	if (that._opts.ttl.packets.will >= 0) {	15!
167	indexes.push({	15✔
168	collection: 'will',
169	key: 'packet.added',
170	name: 'ttl',
171	expireAfterSeconds: that._opts.ttl.packets.will
172	})
173	}
174
175	if (that._opts.ttl.packets.outgoing >= 0) {	15!
176	indexes.push({	15✔
177	collection: 'outgoing',
178	key: 'packet.added',
179	name: 'ttl',
180	expireAfterSeconds: that._opts.ttl.packets.outgoing
181	})
182	}
183
184	if (that._opts.ttl.packets.incoming >= 0) {	15!
185	indexes.push({	15✔
186	collection: 'incoming',
187	key: 'packet.added',
188	name: 'ttl',
189	expireAfterSeconds: that._opts.ttl.packets.incoming
190	})
191	}
192	}
193
194	parallel(that, createIndex, indexes, finishInit)	165✔
195	}
196
197	// drop existing indexes (if exists)
198	if (that._opts.dropExistingIndexes) {	165✔
199	that._dropIndexes(db, initCollections)	6✔
200	} else {
201	initCollections()	159✔
202	}
203	})
204	}
205
206	_dropIndexes (db, cb) {
207	db.collections(function (err, collections) {	6✔
208	if (err) { throw err }	6!
209
210	let done = 0	6✔
211
212	function finish (err) {
213	if (err) { throw err }	36!
214
215	done++	36✔
216	if (done >= collections.length && typeof cb === 'function') {	36✔
217	cb()	6✔
218	}
219	}
220
221	if (collections.length === 0) {	6!
NEW 222	finish()	×
223	} else {
224	for (let i = 0; i < collections.length; i++) {	6✔
225	collections[i].indexExists('ttl', function (err, exists) {	36✔
226	if (err) { throw err }	36!
227
228	if (exists) {	36✔
229	collections[i].dropIndex('ttl', finish)	15✔
230	} else {
231	finish()	21✔
232	}
233	})
234	}
235	}
236	})
237	}
238
239	storeRetained (packet, cb) {
240	if (!this.ready) {	3,036!
NEW 241	this.once('ready', this.storeRetained.bind(this, packet, cb))	×
NEW 242	return	×
243	}
244
245	this.packetsQueue.push({ packet, cb })	3,036✔
246	executeBulk(this)	3,036✔
247	}
248
249	createRetainedStream (pattern) {
250	return this.createRetainedStreamCombi([pattern])	21✔
251	}
252
253	createRetainedStreamCombi (patterns) {
254	let regex = []	24✔
255
256	const instance = through.obj(filterPattern)	24✔
257	instance.matcher = new Qlobber(qlobberOpts)	24✔
258
259	for (let i = 0; i < patterns.length; i++) {	24✔
260	instance.matcher.add(patterns[i], true)	27✔
261	regex.push(escape(patterns[i]).replace(/(\/#\|\\\+).$/, ''))	27✔
262	}
263
264	regex = regex.join('\|')	24✔
265
266	return pump(	24✔
267	toStream(this._cl.retained.find({
268	topic: new RegExp(regex)
269	})),
270	instance
271	)
272	}
273
274	addSubscriptions (client, subs, cb) {
275	if (!this.ready) {	93!
NEW 276	this.once('ready', this.addSubscriptions.bind(this, client, subs, cb))	×
NEW 277	return	×
278	}
279
280	const that = this	93✔
281	let published = 0	93✔
282	let errored = false	93✔
283	const bulk = this._cl.subscriptions.initializeOrderedBulkOp()	93✔
284	subs	93✔
285	.forEach(function (sub) {
286	const subscription = {	168✔
287	clientId: client.id,
288	topic: sub.topic,
289	qos: sub.qos
290	}
291	bulk.find({	168✔
292	clientId: client.id,
293	topic: sub.topic
294	}).upsert().updateOne({
295	$set: decorateSubscription(subscription, that._opts)
296	})
297	})
298
299	bulk.execute(finish)	93✔
300	this._addedSubscriptions(client, subs, finish)	93✔
301
302	function finish (err) {
303	errored = err	186✔
304	published++	186✔
305	if (published === 2) {	186✔
306	cb(errored, client)	93✔
307	}
308	}
309	}
310
311	removeSubscriptions (client, subs, cb) {
312	if (!this.ready) {	48!
NEW 313	this.once('ready', this.removeSubscriptions.bind(this, client, subs, cb))	×
NEW 314	return	×
315	}
316
317	let published = 0	48✔
318	let errored = false	48✔
319	const bulk = this._cl.subscriptions.initializeOrderedBulkOp()	48✔
320	subs	48✔
321	.forEach(function (topic) {
322	bulk.find({	51✔
323	clientId: client.id,
324	topic
325	}).deleteOne()
326	})
327
328	bulk.execute(finish)	48✔
329	this._removedSubscriptions(client, subs.map(toSub), finish)	48✔
330
331	function finish (err) {
332	if (err && !errored) {	96!
NEW 333	errored = true	×
NEW 334	cb(err, client)	×
NEW 335	return	×
336	}
337	published++	96✔
338	if (published === 2 && !errored) {	96✔
339	cb(null, client)	48✔
340	}
341	}
342	}
343
344	subscriptionsByClient (client, cb) {
345	if (!this.ready) {	48!
NEW 346	this.once('ready', this.subscriptionsByClient.bind(this, client, cb))	×
UNCOV 347	return	×
348	}
349
350	this._cl.subscriptions.find({ clientId: client.id }).toArray(function (err, subs) {	48✔
351	if (err) {	48!
NEW 352	cb(err)	×
NEW 353	return	×
354	}
355
356	const toReturn = subs.map(function (sub) {	48✔
357	return {	57✔
358	topic: sub.topic,
359	qos: sub.qos
360	}
361	})
362
363	cb(null, toReturn.length > 0 ? toReturn : null, client)	48✔
364	})
365	}
366
367	countOffline (cb) {
368	let clientsCount = 0	66✔
369	const that = this	66✔
370	toStream(this._cl.subscriptions.aggregate([{	66✔
371	$group: {
372	_id: '$clientId'
373	}
374	}])).on('data', function () {
375	clientsCount++	63✔
376	}).on('end', function () {
377	cb(null, that._trie.subscriptionsCount, clientsCount)	66✔
378	}).on('error', cb)
379	}
380
381	destroy (cb) {
382	if (!this.ready) {	165!
NEW 383	this.once('ready', this.destroy.bind(this, cb))	×
UNCOV 384	return	×
385	}
386
387	if (this._destroyed) {	165!
NEW 388	throw new Error('destroyed called twice!')	×
389	}
390
391	this._destroyed = true	165✔
392
393	cb = cb \|\| noop	165!
394
395	if (this._opts.db) {	165✔
396	cb()	12✔
397	} else {
398	this._client.close(function () {	153✔
399	// swallow err in case of close
400	cb()	153✔
401	})
402	}
403	}
404
405	outgoingEnqueue (sub, packet, cb) {
406	this.outgoingEnqueueCombi([sub], packet, cb)	12✔
407	}
408
409	outgoingEnqueueCombi (subs, packet, cb) {
410	if (!this.ready) {	39!
NEW 411	this.once('ready', this.outgoingEnqueueCombi.bind(this, subs, packet, cb))	×
NEW 412	return	×
413	}
414
415	if (!subs \|\| subs.length === 0) {	39!
NEW 416	return cb(null, packet)	×
417	}
418
419	const newp = new Packet(packet)	39✔
420	const opts = this._opts	39✔
421
422	function createPacket (sub) {
423	return {	42✔
424	clientId: sub.clientId,
425	packet: decoratePacket(newp, opts)
426	}
427	}
428
429	this._cl.outgoing.insertMany(subs.map(createPacket), function (err) {	39✔
430	cb(err, packet)	39✔
431	})
432	}
433
434	outgoingStream (client) {
435	return pump(	33✔
436	toStream(this._cl.outgoing.find({ clientId: client.id })),
437	through.obj(asPacket))
438	}
439
440	outgoingUpdate (client, packet, cb) {
441	if (!this.ready) {	33!
NEW 442	this.once('ready', this.outgoingUpdate.bind(this, client, packet, cb))	×
NEW 443	return	×
444	}
445	if (packet.brokerId) {	33✔
446	updateWithMessageId(this._cl, client, packet, cb)	30✔
447	} else {
448	updatePacket(this._cl, client, packet, cb)	3✔
449	}
450	}
451
452	outgoingClearMessageId (client, packet, cb) {
453	if (!this.ready) {	6!
NEW 454	this.once('ready', this.outgoingClearMessageId.bind(this, client, packet, cb))	×
NEW 455	return	×
456	}
457
458	const outgoing = this._cl.outgoing	6✔
459
460	outgoing.findOne({	6✔
461	clientId: client.id,
462	'packet.messageId': packet.messageId
463	}, function (err, p) {
464	if (err) {	6!
NEW 465	return cb(err)	×
466	}
467
468	if (!p) {	6✔
469	return cb(null)	3✔
470	}
471
472	outgoing.deleteOne({	3✔
473	clientId: client.id,
474	'packet.messageId': packet.messageId
475	}, function (err) {
476	cb(err, p.packet)	3✔
477	})
478	})
479	}
480
481	incomingStorePacket (client, packet, cb) {
482	if (!this.ready) {	6!
NEW 483	this.once('ready', this.incomingStorePacket.bind(this, client, packet, cb))	×
NEW 484	return	×
485	}
486
487	const newp = new Packet(packet)	6✔
488	newp.messageId = packet.messageId	6✔
489
490	this._cl.incoming.insertOne({	6✔
491	clientId: client.id,
492	packet: decoratePacket(newp, this._opts)
493	}, cb)
494	}
495
496	incomingGetPacket (client, packet, cb) {
497	if (!this.ready) {	6!
NEW 498	this.once('ready', this.incomingGetPacket.bind(this, client, packet, cb))	×
NEW 499	return	×
500	}
501
502	this._cl.incoming.findOne({	6✔
503	clientId: client.id,
504	'packet.messageId': packet.messageId
505	}, function (err, result) {
506	if (err) {	6!
NEW 507	cb(err)	×
NEW 508	return	×
509	}
510
511	if (!result) {	6✔
512	cb(new Error('packet not found'), null, client)	3✔
513	return	3✔
514	}
515
516	const packet = result.packet	3✔
517
518	if (packet && packet.payload) {	3!
519	packet.payload = packet.payload.buffer	3✔
520	}
521
522	cb(null, packet, client)	3✔
523	})
524	}
525
526	incomingDelPacket (client, packet, cb) {
527	if (!this.ready) {	3!
NEW 528	this.once('ready', this.incomingDelPacket.bind(this, client, packet, cb))	×
UNCOV 529	return	×
530	}
531
532	this._cl.incoming.deleteOne({	3✔
533	clientId: client.id,
534	'packet.messageId': packet.messageId
535	}, cb)
536	}
537
538	putWill (client, packet, cb) {
539	if (!this.ready) {	18!
NEW 540	this.once('ready', this.putWill.bind(this, client, packet, cb))	×
UNCOV 541	return	×
542	}
543
544	packet.clientId = client.id	18✔
545	packet.brokerId = this.broker.id	18✔
546	this._cl.will.insertOne({	18✔
547	clientId: client.id,
548	packet: decoratePacket(packet, this._opts)
549	}, function (err) {
550	cb(err, client)	18✔
551	})
552	}
553
554	getWill (client, cb) {
555	this._cl.will.findOne({	18✔
556	clientId: client.id
557	}, function (err, result) {
558	if (err) {	18!
NEW 559	cb(err)	×
NEW 560	return	×
561	}
562
563	if (!result) {	18✔
564	cb(null, null, client)	3✔
565	return	3✔
566	}
567
568	const packet = result.packet	15✔
569
570	if (packet && packet.payload) {	15!
571	packet.payload = packet.payload.buffer	15✔
572	}
573
574	cb(null, packet, client)	15✔
575	})
576	}
577
578	delWill (client, cb) {
579	const will = this._cl.will	12✔
580	this.getWill(client, function (err, packet) {	12✔
581	if (err \|\| !packet) {	12!
NEW 582	cb(err, null, client)	×
NEW 583	return	×
584	}
585	will.deleteOne({	12✔
586	clientId: client.id
587	}, function (err) {
588	cb(err, packet, client)	12✔
589	})
590	})
591	}
592
593	streamWill (brokers) {
594	const query = {}	6✔
595
596	if (brokers) {	6✔
597	query['packet.brokerId'] = { $nin: Object.keys(brokers) }	3✔
598	}
599	return pump(toStream(this._cl.will.find(query)), through.obj(asPacket))	6✔
600	}
601
602	getClientList (topic) {
603	const query = {}	6✔
604
605	if (topic) {	6!
606	query.topic = topic	6✔
607	}
608
609	return pump(toStream(this._cl.subscriptions.find(query)), through.obj(function asPacket (obj, enc, cb) {	6✔
610	this.push(obj.clientId)	9✔
611	cb()	9✔
612	}))
613	}
614	}
615
616	function decoratePacket (packet, opts) {
617	if (opts.ttl.packets) {	3,099✔
618	packet.added = new Date()	15✔
619	}
620	return packet	3,099✔
621	}
622
623	function executeBulk (that) {
624	if (!that.executing && that.packetsQueue.length > 0) {	3,078✔
625	that.executing = true	42✔
626	const bulk = that._cl.retained.initializeOrderedBulkOp()	42✔
627	const onEnd = []	42✔
628
629	while (that.packetsQueue.length) {	42✔
630	const p = that.packetsQueue.shift()	3,036✔
631	onEnd.push(p.cb)	3,036✔
632	const criteria = { topic: p.packet.topic }	3,036✔
633
634	if (p.packet.payload.length > 0) {	3,036✔
635	bulk.find(criteria).upsert().updateOne({	3,033✔
636	$set: decoratePacket(p.packet, that._opts)
637	})
638	} else {
639	bulk.find(criteria).deleteOne()	3✔
640	}
641	}
642
643	bulk.execute(function () {	42✔
644	while (onEnd.length) onEnd.shift().call()	3,036✔
645	that.executing = false	42✔
646	executeBulk(that)	42✔
647	})
648	}
649	}
650
651	function filterPattern (chunk, enc, cb) {
652	if (this.matcher.match(chunk.topic).length > 0) {	21!
653	chunk.payload = chunk.payload.buffer	21✔
654	// this is converting chunk to slow properties
655	// https://github.com/sindresorhus/to-fast-properties
656	// might make this faster
657	delete chunk._id	21✔
658	this.push(chunk)	21✔
659	}
660	cb()	21✔
661	}
662
663	function decorateSubscription (sub, opts) {
664	if (opts.ttl.subscriptions) {	168!
NEW 665	sub.added = new Date()	×
666	}
667	return sub	168✔
668	}
669
670	function toSub (topic) {
671	return {	51✔
672	topic
673	}
674	}
675
676	function asPacket (obj, enc, cb) {
677	const packet = obj.packet	45✔
678
679	if (packet.payload) {	45✔
680	const buffer = packet.payload.buffer	42✔
681	if (buffer && Buffer.isBuffer(buffer)) {	42✔
682	packet.payload = packet.payload.buffer	39✔
683	}
684	}
685	cb(null, packet)	45✔
686	}
687
688	function updateWithMessageId (db, client, packet, cb) {
689	db.outgoing.updateOne({	30✔
690	clientId: client.id,
691	'packet.brokerCounter': packet.brokerCounter,
692	'packet.brokerId': packet.brokerId
693	}, {
694	$set: {
695	'packet.messageId': packet.messageId
696	}
697	}, function (err) {
698	cb(err, client, packet)	30✔
699	})
700	}
701
702	function updatePacket (db, client, packet, cb) {
703	db.outgoing.updateOne({	3✔
704	clientId: client.id,
705	'packet.messageId': packet.messageId
706	}, {
707	$set: {
708	clientId: client.id,
709	packet
710	}
711	}, function (err) {
712	cb(err, client, packet)	3✔
713	})
714	}
715
716	function noop () { }
717
718	module.exports = (opts) => new MongoPersistence(opts)	168✔

moscajs / aedes-persistence-mongodb / 2314212754

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous