2318713467

Build Type

push

github

Committed by GitHub

Commit Message

feat: allow subscription to store more props (#76)

Run Details

108 of 149 branches covered (72.48%)

Branch coverage included in aggregate %.

233 of 272 new or added lines in 1 file covered. (85.66%)

5 existing lines in 1 file now uncovered.

244 of 288 relevant lines covered (84.72%)

60.85 hits per line

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

80.55

/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 = Object.assign({}, sub)
        subscription.clientId = client.id
        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) {
        // remove clientId and _id from sub
        const { clientId, _id, ...resub } = sub
        return resub
      })

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

moscajs / aedes-persistence-mongodb / 2318713467

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