feat(producer): class Producer init

Author: leizongmin

lib/producer.js

@@ -6,3 +6,112 @@
  * @authro Zongmin Lei <leizongmin@gmail.com>
  */
 
+const EventEmitter = require('events').EventEmitter;
+const leiPromise = require('lei-promise');
+const utils = require('./utils');
+const debug = utils.debug('producer');
+
+class Producer {
+
+  /**
+   * Constructor
+   *
+   * @param {Object} options
+   *   - {String} queue
+   *   - {Number} maxAge
+   *   - {Object} redis
+   *     - {String} host
+   *     - {Number} port
+   *     - {Number} db
+   *     - {String} prefix
+   */
+  constructor(options) {
+
+    if (!options.queue) throw new Error('missing queue name');
+
+    this._redis = utils.createRedisClient(options.redis);
+    this.name = utils.generateClientId('producer');
+    this._maxAge = options.maxAge || 0;
+    this._redisPrefix = (options.redis && options.redis.prefix) || '';
+    this.queue = options.queue;
+    this._queueKey = this._redisPrefix + options.queue;
+
+    this._msgId = 0;
+    this._msgCallback = new EventEmitter();
+    const msgCallback = (id, err, result) => this._msgCallback.emit(id, err, result);
+
+    this._redis.subscribe(`${this._redisPrefix}CB:${this.name}`, (chennel, message) => {
+
+      // msg: success => id,s,data    error => id,e,data
+      const info = utils.splitString(message, 3);
+      if (info.length !== 3) return;
+
+      const id = info[0];
+      const type = info[1];
+
+      if (type === 's') {
+        let data;
+        try {
+          data = JSON.parse(info[2]);
+        } catch (err) {
+          return msgCallback(id, new Error(`JSON.parse(data) fail: ${err.message}`));
+        }
+        msgCallback(id, null, {result: data});
+      } else if (type === 'e') {
+        const err = info[2];
+        msgCallback(id, new Error(err));
+      }
+
+    });
+
+    this.push = leiPromise.promisify(this.push.bind(this));
+    this.exit = leiPromise.promisify(this.exit.bind(this));
+
+  }
+
+  /**
+   * push to queue
+   *
+   * @param {Object} info
+   *   - {Object} data
+   *   - {Number} maxAge
+   * @param {Function}
+   */
+  push(info, callback) {
+
+    info.maxAge = ('maxAge' in info ? info.maxAge : this._maxAge);
+
+    const expire = utils.secondTimestamp() + info.maxAge;
+    let data;
+    try {
+      data = JSON.stringify(info.data);
+    } catch (err) {
+      return callback(new Error(`JSON.stringify(data) fail: ${err.message}`));
+    }
+
+    const id = utils.integerToBase64(++this._msgId);
+    // msg: producer,id,expire,data
+    const content = `${this.name},${id},${expire},${data}`;
+
+    this._redis.lpush(this._queueKey, content, err => {
+      if (err) return callback(err);
+
+      this._msgCallback.once(id, callback);
+    });
+
+  }
+
+  /**
+   * exit
+   *
+   * @param {Function} callback
+   */
+  exit(callback) {
+    this._redis.end();
+    this._msgCallback = null;
+    callback && callback(null);
+  }
+
+}
+
+module.exports = Producer;

lib/utils.js

@@ -0,0 +1,95 @@
+'use strict';
+
+/**
+ * super-queue utils
+ *
+ * @authro Zongmin Lei <leizongmin@gmail.com>
+ */
+
+const createDebug = require('debug');
+const Redis = require('ioredis');
+const utils = exports = module.exports = require('lei-utils').extend();
+
+// if set environment variable NODE_TEST includes "super-queue" then IS_TEST=true
+const IS_TEST = (String(process.env.NODE_TEST).indexOf('super-queue') !== -1);
+
+
+/**
+ * Create Debug Function
+ *
+ * @param {String} name
+ * @return {Function}
+ */
+utils.debug = function (name) {
+  return 'super-queue:' + name;
+};
+
+/**
+ * Generate Redis Client
+ *
+ * @param {Object} options
+ * @return {Object}
+ */
+utils.createRedisClient = function (options) {
+  if (IS_TEST) {
+    return require('fakeredis').createClient();
+  } else {
+    return new Redis(options);
+  }
+};
+
+/**
+ * Generate Client ID
+ *
+ * @param {String} type should be one of "producer", "consumer", "monitor"
+ * @return {String}
+ */
+utils.generateClientId = function (type) {
+  return type.slice(0, 1).toUpperCase() + (utils.md5(Date.now() + '' + Math.random()).slice(0, 9));
+};
+
+/**
+ * Get Second Timestamp
+ *
+ * @return {Number}
+ */
+utils.secondTimestamp = function () {
+  return (Date.now() / 1000) << 0;
+};
+
+/**
+ * Integer Value to Base64 String
+ *
+ * @param {Number} value
+ * @param {Number} byteLength
+ * @return {String}
+ */
+utils.integerToBase64 = function (value, byteLength) {
+  const buf = new Buffer(byteLength);
+  buf.writeIntLE(value, 0, byteLength);
+  return buf.toString('base64');
+};
+
+/**
+ * Split String by Separator
+ *
+ * @param {String} text
+ * @param {String} separator
+ * @param {Number} length
+ * @return {Array}
+ */
+utils.splitString = function (text, separator, length) {
+  const list = [];
+  let lastIndex = 0;
+  for (let i = 0; i < length - 1; i++) {
+    let j = text.indexOf(separator, lastIndex);
+    if (j === -1) {
+      break;
+    } else {
+      list.push(text.slice(lastIndex, j));
+      lastIndex = j + 1;
+    }
+  }
+  list.push(text.slice(lastIndex, text.length));
+  return list;
+};