@knetik/micro-queue/dist/index.js

const { join, resolve } = require('path');
const _ = require('lodash');
const requirePath = require('require-path');
const Arena = require('bull-arena');
const Queue = require('bull');

const Redis = require('ioredis');

let client;
let subscriber;

const JobBase = require('./job_base');

/**
 * Generate a new job
 *
 * ```
 * $ yarn run micro-queue generate {job_name}
 * ```
 *
 * Creates a job inside the /app/jobs/{job_name}.job.js
 *
 * @name MicroQueueAdaptor
*/
class MicroQueueAdaptor {

  /**
   * A Micro Core applicaiton calls init() on all adaptors during application
   * init
   *
   * - create the redis client
   * - load the job classes
   * - create the job queue instsances
   * - assign the processor to each queue
   * - Add this to Core as the Queue Adaptor
  */
  static init(Core) {
    const opts = createClient();

    return loadJobs().then(modules => {
      modules.forEach(module => {
        const queue = new Queue(module.name, _.merge(opts, { prefix: `{${module.name}}` }));

        queue.process(module.name, module.concurrency, join(this.QUEUE_ROOT, 'worker.js'));

        module.queue = queue;

        this.jobs[module.name] = module;
      });

      return Core.AdaptorInitializer.add('Queue', this);
    });
  }

  /**
   * After Micro Core init, load the Arena Queue UI into Core.Server. Only is
   * Core.Server exists
  */
  static afterInitialize(Core) {
    Core.Logger.silly('Queue after initialize: mount queue ui');

    return mountArena(Core, this.jobs);
  }

  /**
   * On App connect, adds the Job instances to the App scoping perform_later to
   * the App instance;
  */
  static connect(App) {
    const jobs = {};

    Object.keys(this.jobs).forEach(key => {
      const Job = this.jobs[key];
      jobs[key] = new Job(App);
    });

    return new this(jobs);
  }

  static loadJobs() {
    return loadJobs().then(modules => {
      const jobs = {};

      modules.forEach(module => {
        jobs[module.name] = module;
      });

      return jobs;
    });
  }

  constructor(jobs) {
    this.jobs = jobs;
  }

  /**
   * Get a Job instance from the App instances compiled Job collection
   *
   * ```
   * const Job = App.Queue.get('SomeJobNameJob');
   * ```
  */
  get(name) {
    return this.jobs[name];
  }
}

function createClient() {
  const { REDIS_HOST } = process.env;
  client = client || new Redis(REDIS_HOST);
  subscriber = subscriber || new Redis(REDIS_HOST);

  return {
    createClient(type) {
      switch (type) {
        case 'client':
          return client;
        case 'subscriber':
          return subscriber;
        default:
          return new Redis(REDIS_HOST);
      }
    }
  };
}

function mountArena(Core, jobs) {
  if (!Core.Server) {
    return Promise.resolve(undefined);
  }

  const names = Object.keys(jobs);

  const arena = Arena({
    queues: names.map(name => ({
      name,
      hostId: 'Queues',
      prefix: `{${name}}`,
      redis: { host: process.env.REDIS_HOST }
    }))
  }, {
    disableListen: true
  });

  Core.Server.use('/arena', arena);

  return Promise.resolve('mounted Arena');
}

function loadJobs() {
  const { APP_ROOT } = process.env;
  const jobsPath = join(APP_ROOT, 'app', 'jobs');

  return requirePath({
    path: jobsPath,
    include: ['*.job.js']
  }).then(modules => _.map(modules, (module, key) => module));
}

module.exports = MicroQueueAdaptor;
module.exports.QUEUE_ROOT = resolve(__dirname);
module.exports.jobs = {};
module.exports.JobBase = JobBase;