mq.js

const mqlight = require('mqlight');
const joi = require('joi');
const url = require('url');

const debug = require('debug')('mqService');

module.exports = (cfg) => {

  const { protocol, hostname, port } = cfg.endpoint.mq;

  const baseURL = url.format({
    slashes: true,
    protocol,
    hostname,
    port
  });

  const opts = Object.assign({}, { service: baseURL });

  const client = mqlight.createClient(opts);

  const wrapper = {
    send: (topic, data, opts) => {
      joi.validate(topic, joi.string(), (err) => {
        if (!err) {
          debug('send', topic, data, opts);
          client.send(topic, data, opts);
        }
      });
    },
    subscribe: (topic, share, opts) => {
      joi.validate(topic, joi.string(), (err) => {
        if (!err) {
          debug('subscribe', topic, share, opts);
          client.subscribe(topic, share, opts);
        }
      });
    },
    unsubscribe: (topic, share, opts) => {
      joi.validate(topic, joi.string(), (err) => {
        if (!err) {
          debug('unsubscribe', topic, share, opts);
          client.unsubscribe(topic, share, opts);
        }
      });
    },
    stop: client.stop(),
    id: client.id,
    service: client.service,
    state: client.state
  };

  return new Promise((resolve, reject) => {
    client.on('started', () => resolve(wrapper));
    client.on('message', msg => debug('mq: ', msg));
    client.on('error', err => reject(err));
  });

};