File Explorer

1'use strict'2 3/**4 * Module dependencies5 */6const xtend = require('xtend')7 8const Readable = require('readable-stream').Readable9const streamsOpts = { objectMode: true }10const defaultStoreOptions = {11  clean: true12}13 14/**15 * In-memory implementation of the message store16 * This can actually be saved into files.17 *18 * @param {Object} [options] - store options19 */20function Store (options) {21  if (!(this instanceof Store)) {22    return new Store(options)23  }24 25  this.options = options || {}26 27  // Defaults28  this.options = xtend(defaultStoreOptions, options)29 30  this._inflights = new Map()31}32 33/**34 * Adds a packet to the store, a packet is35 * anything that has a messageId property.36 *37 */38Store.prototype.put = function (packet, cb) {39  this._inflights.set(packet.messageId, packet)40 41  if (cb) {42    cb()43  }44 45  return this46}47 48/**49 * Creates a stream with all the packets in the store50 *51 */52Store.prototype.createStream = function () {53  const stream = new Readable(streamsOpts)54  const values = []55  let destroyed = false56  let i = 057 58  this._inflights.forEach(function (value, key) {59    values.push(value)60  })61 62  stream._read = function () {63    if (!destroyed && i < values.length) {64      this.push(values[i++])65    } else {66      this.push(null)67    }68  }69 70  stream.destroy = function () {71    if (destroyed) {72      return73    }74 75    const self = this76 77    destroyed = true78 79    setTimeout(function () {80      self.emit('close')81    }, 0)82  }83 84  return stream85}86 87/**88 * deletes a packet from the store.89 */90Store.prototype.del = function (packet, cb) {91  packet = this._inflights.get(packet.messageId)92  if (packet) {93    this._inflights.delete(packet.messageId)94    cb(null, packet)95  } else if (cb) {96    cb(new Error('missing packet'))97  }98 99  return this100}101 102/**103 * get a packet from the store.104 */105Store.prototype.get = function (packet, cb) {106  packet = this._inflights.get(packet.messageId)107  if (packet) {108    cb(null, packet)109  } else if (cb) {110    cb(new Error('missing packet'))111  }112 113  return this114}115 116/**117 * Close the store118 */119Store.prototype.close = function (cb) {120  if (this.options.clean) {121    this._inflights = null122  }123  if (cb) {124    cb()125  }126}127 128module.exports = Store129