rabbit-hapi

HAPI plugin to wrap basic RabbitMQ operations based on amqp library for NodeJS.

This plugin enables these RabbitMQ operations : - pub/sub - send to queue or exchange - consume messages from queue or exchange - send a request in RPC mode

For further informations about RabbitMQ possibilities, see RabbitMQ official documentation.

Installation

npm install --save rabbit-hapi

Usage

server.register({
    plugin  : require('rabbit-hapi'),
    options : {}
});

Options

### Global settings

Global settings use for broker connection.

Params	Description	Default
hostname	RabbitMQ broker hostname	localhost
port	Port for AMQP	5672
credentials	Credential to connect to RabbitMQ broker	-
heartbeat	Period of the connection heartbeat (seconds)	30
maxRetry	Max retry allowed on connection or channel error	5

RabbitMQ settings

Default settings applied on queues/exchanges.

Params	Description	Default
type	Exchange type	direct
options	Queue options	-
options.durable	Durable options for queue	true
options.noAck	Disable acknowledgement on messages if true	false
options.RPCExpire	Expiration time (milliseconds) for answer queue in RPC mode	60000
options.allUpTo	allUpTo on ack/noAck	false
options.requeue	Requeue on noAck	true
options.debug	Debug configuration
options.debug.isActivated	Activate debug	false
options.debug.expires	Queue expires after x ms	86400000 // 24 hours
options.debug.durable	Queue is durable	true
options.debug.persistent	Message requeue on error are persistent	true

Messages settings

Default messages settings.

Params	Description	Default
content	Message content	-
options	Message options	-
options.contentType	Content MIME type	application/json
options.persistent	If true, message will survive a broker restart	true

Debug mode

When debug is activated, if consumer fails during message treatment, message will be also queue in a specific queue. Queue can be named in options.debug.queue. If not, name will be the original queue's name plus debug.

Server Methods

publish

Publish a message through a fanout exchange.

Parameters

Field	Type	Description
message	Object/String	Message to publish (if string, automatically assign to message.content)
message.content	*	Message content
message.options	Object	Message options (same as those provided by amqp lib)
exchange	String	Exchange name
options	Object	Exchange options
routingKey	String	(optional) Routing key to use
queue	String	(optional) Queue name

Usage

rabbitHapi.publish({
    exchange    : 'publishExchange',
    options     : {
        durable     : false
    },
    message     : 'Hello World !'
}).then(() => {
    // do some stuff here...
}).catch(() => {
    // something went wrong...
});

subscribe

Subscribe to a fanout exchange.

Parameters

Field	Type	Description
exchange	String	Exchange name
options	Object	Exchange options
queue	String	(optional) Queue name
waitingFunc	Function	Function to call when starting consuming
receiveFunc	Function	Function to call on message reception

Usage

rabbitHapi.subscribe({
    exchange    : 'publishExchange',
    options     : {
        durable     : false,
        noAck       : false,
        exclusive   : true
    },
    receiveFunc : (message) => {
        console.log('Receive', message.content.toString());
    },
    waitingFunc : () => {
        console.log('Waiting for message');
    }
});

send

Send a message to an exchange or a queue. If both routing key and queue are specified, bind the exchange to the queue using the routing key. In order to use queue generated from amqp (amq.gen...), you need to specified the parameter generatedQueue.

Parameters

Field	Type	Description
message	Object/String	Message to publish (if string, automatically assign to message.content)
message.content	*	Message content
message.options	Object	Message options (same as those provided by amqp lib)
exchange	String	Exchange name
type	String	Exchange type
options	Object	Exchange options
routingKey	String	(optional) Routing key to use
queue	String	(optional) Queue name
generatedQueue	Boolean	(optional) True to use queue generated by the broker

Usage

rabbitHapi.send({
    queue       : 'hello',
    options     : {
        durable : false,
        noAck   : false
    },
    message     : 'Hello World !'
}).then(() => {
    // do some stuff....
}).catch(() => {
    // something went wrong...
});

consume

Consume messages from an exchange or a queue. Automatic reconnection to a new channel on connection error/lost.

Parameters

Field	Type	Description
exchange	String	Exchange name
type	String	Exchange type
options	Object	Exchange options
prefetch	Number	Specify prefetch on the channel
queue	String	(optional) Queue name
waitingFunc	Function	Function to call when starting consuming
receiveFunc	Function	Function to call on message reception

Usage

rabbitHapi.consume({
    queue       : 'hello',
    options     : {
        durable : false,
        noAck   : false
    },
    receiveFunc : (message) => {
        console.log('Direct message receive ', message.content.toString());
    },
    waitingFunc : () => {
        console.log('Waiting for message');
    }
});

bindExchange

Bind a key (or an array of keys) to exchange/queue. Create the exchange and/or the queue if it doesn't exist.

Parameters

Field	Type	Description
exchange	String	Exchange name
type	String	Exchange type
options	Object	Exchange options
queue	String	(optional) Queue name
routingKeys	String/String[]	Routing keys to bind to

Usage

rabbitHapi.bindExchange({
    exchange    : 'logExchange',
    type        : 'direct',
    queue       : 'logQueue',
    routingKeys : [ 'error', 'info', 'debug' ],
    options     : {
        durable     : true,
        noAck       : false
    }
});

sendRPC

Send a message acting like a client RPC.

Parameters

Field	Type	Description
message	Object/String	Message to publish (if string, automatically assign to message.content)
message.content	*	Message content
message.options	Object	Message options (same as those provided by amqp lib)
queue	String	Queue to send request on
options	Object	Exchange options
RPCTimeout	Number	Timeout if no answer is received from the RPC server (default to 30 sec)
receiveFunc	Function	Function to call on server response

Usage

rabbitHapi.sendRPC({
    queue       : 'rpc_queue',
    options     : {
        durable : false,
        noAck   : false
    },
    message     : 'Hello RPC server !',
    receiveFunc : (answer) => {
        console.log('Server answer with', answer.content.toString());
        // do some stuff...
    }
});

answerToRPC

Answer to an RPC request.

Note : answer queue is specified and not exclusive in order to enable retry on connection loss.

Parameters

Field	Type	Description
prefetch	Number	Specify prefetch on the channel
queue	String	Queue to send the response on
options	Object	Exchange options
waitingFunc	Function	Function to call when starting consuming
receiveFunc	Function	Function to call on message reception (return value of the function will be send to client RPC)

Usage

rabbitHapi.answerToRPC({
    queue       : 'rpc_queue',
    options     : {
        durable     : false,
        noAck       : false
    },
    receiveFunc : (message) => {
        console.log('Message received', message.content.toString(), '... sending response...');
        return 'Hello client !';
    }
});

close

Close manually RabbitMQ connection.