Naturally Processed Meals


    0.0.10 • Public • Published


    rabbitmq_minionpool is a specialized minionpool that will let you process tasks coming in via RabbitMQ (it uses de node-amqp library).

    How it works

    You need to provide some key pieces of information:

    • An exchange name (the "worker's exchange" from now on)
    • A queue name (the "worker's queue" from now on)
    • A routing key (will default to the queue name if missing)
    • A retry timeout for failed operations.

    When you create a rabbitmq_minionpool, 2 exchanges are created in the rabbitmq server:

    • The exchange name specified (let's say "workers").
    • A dead letter exchange for failed operations, automatically named as the original exchange name and with a suffix ".retry" (e.g: "workers.retry").

    Both exchanges are created as 'topic', durable', 'not passive'. Also, the channel is set in 'confirm' mode (in case you want to publish your own messages).

    Also, some queues are created:

    • The worker's queue name specified in the given worker's exchange, and binded to the given routing key. The pool will subscribe to this queue to get messages. This queue is created with the arguments:

    • x-dead-letter-exchange = exchangeName.retry

    • x-dead-letter-routing-key = queueName.retry

    • Another queue in the dead letter exchange, so failed operations can get there. This queue is created with the arguments:

    • x-dead-letter-exchange = exchangeName

    • x-dead-letter-routing-key = queueName

    • x-message-ttl = retryTimeout

    Both queue will subscribe to 'routingKey' but also to their respectively queueName. So they will get messages directed to 'routingKey' but will also get dead-lettered messages (and these messages will reach the correct consumer, the one that rejected them.)

    When messages are routed to the specified worker's queue, the minionpool will dispatch them to the minions. Each minion will get access to the message and the queue object where it came from. If the minion rejects the message, the message will be routed to the queue in the dead letter exchange with the given TTL. When the TTL expires, the message will go back automatically to the original queue, where the operation can be retried.


    var options = {
      name: 'test',
      debug: true,
      concurrency: 5,
      logger: console.log,
      mqOptions: {
        host: '',
        login: 'guest',
        password: 'guest',
        authMechanism: 'AMQPLAIN',
        vhost: '/',
        reconnect: true,
        reconnectBackoffStrategy: 'linear',
        reconnectExponentialLimit: 120000,
        reconnectBackoffTime: 1000,
        exchangeName: 'workers',  // Will also create workers.retry
        queueName: 'myWorkers',   // Will also create myWorkers.retry
        routingKey: 'myWorkers',  // Optional. Equals to queueName if missing
        retryTimeout: 20000
      minionTaskHandler: function(msg, state, callback) {
        var payload = msg.payload;
        var headers = msg.headers;
        var deliveryInfo = msg.deliveryInfo;
        var message = msg.message;
        var queue = msg.queue;
        console.log('got task: %s', util.inspect(payload));
        // See the node-amqp doc for more info.
        message.reject(); // or message.acknowledge();
        callback(undefined, state);
      poolEnd: function() {
    var pool = new minionsMod.RabbitMqMinionPool(options);
    process.on('SIGINT', function() {


    • Design your apps and architecture in such a way that operations are idempotent to max the benefits of this.

    Using multiple cores

    In the case of having rabbitmq and mysql workers, it's very useful to take advantage of multicore cpu's. For this, you can use taskset and launch multiple minionpool instances on different cores.


    npm i rabbitmq_minionpool

    DownloadsWeekly Downloads






    Last publish


    • marcelog