mqemitter-redis.js

'use strict'

const Redis = require('ioredis')
const MQEmitter = require('mqemitter')
const hyperid = require('hyperid')()
const inherits = require('inherits')
const LRU = require('lru-cache')
const msgpack = require('msgpack-lite')
const EE = require('events').EventEmitter
const Pipeline = require('ioredis-auto-pipeline')

function MQEmitterRedis (opts) {
  if (!(this instanceof MQEmitterRedis)) {
    return new MQEmitterRedis(opts)
  }

  opts = opts || {}

  this._opts = opts

  if (opts.isCluster) {
    this.subConn = new Redis.Cluster(opts.cluster)
    this.pubConn = new Redis.Cluster(opts.cluster)
  } else {
    this.subConn = new Redis(opts.connectionString || opts)
    this.pubConn = new Redis(opts.connectionString || opts)
  }

  this._pipeline = Pipeline(this.pubConn)

  this._topics = {}

  this._cache = new LRU({
    max: 10000,
    ttl: 60 * 1000 // one minute
  })

  this.state = new EE()

  this.state.on('error', (err) => {
    console.error('Error from mqemitter-redis:', err)
  })

  const that = this

  function onError (err) {
    if (err && !that.closing) {
      that.state.emit('error', err)
    }
  }

  this._onError = onError

  function handler (sub, topic, payload) {
    const packet = msgpack.decode(payload)
    if (!that._cache.get(packet.id)) {
      that._emit(packet.msg)
    }
    that._cache.set(packet.id, true)
  }

  this.subConn.on('messageBuffer', function (topic, message) {
    handler(topic, topic, message)
  })

  this.subConn.on('pmessageBuffer', function (sub, topic, message) {
    handler(sub, topic, message)
  })

  this.subConn.on('connect', function () {
    that.state.emit('subConnect')
  })

  this.subConn.on('error', function (err) {
    that._onError(err)
  })

  this.pubConn.on('connect', function () {
    that.state.emit('pubConnect')
  })

  this.pubConn.on('error', function (err) {
    that._onError(err)
  })

  MQEmitter.call(this, opts)

  this._opts.regexWildcardOne = new RegExp(this._opts.wildcardOne.replace(/([/,!\\^${}[\]().*+?|<>\-&])/g, '\\$&'), 'g')
  this._opts.regexWildcardSome = new RegExp((this._opts.matchEmptyLevels ? this._opts.separator.replace(/([/,!\\^${}[\]().*+?|<>\-&])/g, '\\$&') + '?' : '') + this._opts.wildcardSome.replace(/([/,!\\^${}[\]().*+?|<>\-&])/g, '\\$&'), 'g')
}

inherits(MQEmitterRedis, MQEmitter)

;['emit', 'on', 'removeListener', 'close'].forEach(function (name) {
  MQEmitterRedis.prototype['_' + name] = MQEmitterRedis.prototype[name]
})

MQEmitterRedis.prototype.close = function (cb) {
  cb = cb || noop

  if (this.closed || this.closing) {
    return cb()
  }

  this.closing = true

  let count = 2
  const that = this

  function onEnd () {
    if (--count === 0) {
      that._close(cb)
    }
  }

  this.subConn.on('end', onEnd)
  this.subConn.quit()

  this.pubConn.on('end', onEnd)
  this.pubConn.quit()

  return this
}

MQEmitterRedis.prototype._subTopic = function (topic) {
  return topic
    .replace(this._opts.regexWildcardOne, '*')
    .replace(this._opts.regexWildcardSome, '*')
}

MQEmitterRedis.prototype.on = function on (topic, cb, done) {
  const subTopic = this._subTopic(topic)
  const onFinish = function () {
    if (done) {
      setImmediate(done)
    }
  }

  this._on(topic, cb)

  if (this._topics[subTopic]) {
    this._topics[subTopic]++
    onFinish.call(this)
    return this
  }

  this._topics[subTopic] = 1

  if (this._containsWildcard(topic)) {
    this.subConn.psubscribe(subTopic, onFinish.bind(this))
  } else {
    this.subConn.subscribe(subTopic, onFinish.bind(this))
  }

  return this
}

MQEmitterRedis.prototype.emit = function (msg, done) {
  done = done || this._onError

  if (this.closed) {
    const err = new Error('mqemitter-redis is closed')
    return done(err)
  }

  const packet = {
    id: hyperid(),
    msg
  }

  this._pipeline.publish(msg.topic, msgpack.encode(packet)).then(() => done()).catch(done)
}

MQEmitterRedis.prototype.removeListener = function (topic, cb, done) {
  const subTopic = this._subTopic(topic)
  const onFinish = function () {
    if (done) {
      setImmediate(done)
    }
  }

  this._removeListener(topic, cb)

  if (--this._topics[subTopic] > 0) {
    onFinish()
    return this
  }

  delete this._topics[subTopic]

  if (this._containsWildcard(topic)) {
    this.subConn.punsubscribe(subTopic, onFinish)
  } else if (this._matcher.match(topic)) {
    this.subConn.unsubscribe(subTopic, onFinish)
  }

  return this
}

MQEmitterRedis.prototype._containsWildcard = function (topic) {
  return (topic.indexOf(this._opts.wildcardOne) >= 0) ||
         (topic.indexOf(this._opts.wildcardSome) >= 0)
}

function noop () {}

module.exports = MQEmitterRedis

function MQEmitterRedisPrefix (pubSubPrefix, options) {
  MQEmitterRedis.call(this, options)
  this._pubSubPrefix = pubSubPrefix
  this._sym_proxiedCallback = Symbol('proxiedCallback')
}
inherits(MQEmitterRedisPrefix, MQEmitterRedis)
MQEmitterRedisPrefix.prototype.on = function (topic, cb, done) {
  const t = this._pubSubPrefix + topic
  cb[this._sym_proxiedCallback] = function (packet, cbcb) {
    const t = packet.topic.slice(this._pubSubPrefix.length)
    const p = { ...packet, topic: t }
    return cb.call(this, p, cbcb)
  }.bind(this)
  return MQEmitterRedis.prototype.on.call(this, t, cb[this._sym_proxiedCallback], done)
}
MQEmitterRedisPrefix.prototype.removeListener = function (topic, func, done) {
  const t = this._pubSubPrefix + topic
  const f = func[this._sym_proxiedCallback]
  return MQEmitterRedis.prototype.removeListener.call(this, t, f, done)
}
MQEmitterRedisPrefix.prototype.emit = function (packet, done) {
  const t = this._pubSubPrefix + packet.topic
  const p = { ...packet, topic: t }
  return MQEmitterRedis.prototype.emit.call(this, p, done)
}

module.exports.MQEmitterRedisPrefix = MQEmitterRedisPrefix