Alpha AMQP Consumer

Library for reliable message consumption via AMQP protocol.

Features:

• Automatically reconnects and restores channels for consumers
• Super easy to use
• Ability to stop/resume consumption
• Supports queue <-> exchange binding
• Supports queue assertion
• Allows to retry message consumption after certain amount of time
• Maintains counter of ongoing consumptions which helps implement graceful app shutdown
• Easy debugging with debug module

Install

npm install alpha-amqp-consumer

Usage

const connect = require('alpha-amqp-consumer');

connect('amqp://localhost')
    .then(manager => {
        manager.consume(
            (message) => {
                // once function execution is done the message will be ACK-ed
            },
            {queue: 'example-queue'},
        );
    }); 

Usage with promises

// (...)

manager.consume(
    {queue: 'example-queue'},
    () => {
        // automatically ACK-ed once promise gets resolved (in that case after 1 second)
        return new Promise((resolve) => setTimeout(resolve, 1000));
    }
);
// (...)

API

See special API declaration file and examples directory.

Message ACK-ing and REJECT-ing

Every consumer has "resultHandler" which a function that decides what to do with the messages based on the result from consumer function. Message is rejected automatically it consumer function throws an error or returned promise gets rejected, otherwise message is ACKed. You can customize the behavior by providing resultHandler

manager.consume((message) => {
    // do something
}, {
    queue: 'some-queue',
    resultHandler(context, error) {
        if (error) {
            // maybe thrown error is fine?
            if (isAcceptableError(error)) {
                context.ack();
            } else {
                context.reject();
            }
        } else {
            context.ack();
        }
    }
})

Retry with delay

There is no way to say AMQP to retry message consumption after certain period of time. In order to achieve that we need a bit more typology setup.

We need:

• name of "pre" retry exchange
• name of "post-retry" exchange
• name of queue for temporary messages storage

For more information how retry works another document.

manager.setupDelayedRetryTopology({
    exchange: {
        pre: 'pre-retry',
        post: 'post-retry'
    },
    queue: 'messages-to-retry'
})
    .then(() => {
        manager.consume(() => {
            // do something with message
        }, {
            queue: 'some-queue',
            resultHandler(context, error) {
                if (error) {
                    context.retry(5000);
                } else {
                    context.ack();
                }
            }
        })
    });

Debugging

Run your app with DEBUG env variable. See debug package docs for more details.

DEBUG=alpha-amqp-consumer:* node app.js

Changelog

0.4.3

• Added extra exported types

0.4.2

• "channel" and "retryTopology" for ConsumerManager are now public
• "ConnectionManagerOptions" interface exported