This is a very opinionated abstraction over amqplib to help simplify certain common tasks and (hopefully) reduce the effort required to use RabbitMQ in your Node services.
- Gracefully handle re-connections
- Automatically re-define all topology on re-connection
- Automatically re-send any unconfirmed messages on re-connection
- Support the majority of RabbitMQ's extensions
- Handle batching of acknowledgements and rejections
- Topology & configuration via the JSON configuration method (thanks to @JohnDMathis!)
- Fault-tolerance/resilience over throughput
- Default to publish confirmation
- Default to ack mode on consumers
- Heterogenous services that include statically typed languages
- JSON as the only serialization provider
This library implements promises for many of the calls via when.js.
This syntax uses an options object rather than arguments, here's an example showing all of the available properties:
rabbit.publish( 'exchange.name', {
routingKey: 'hi',
type: 'company.project.messages.textMessage',
correlationId: 'one',
body: { text: 'hello!' },
messageId: '100',
expiresAfter: 1000 // TTL in ms, in this example 1 second
timestamp: // posix timestamp (long)
headers: {
'random': 'application specific value'
}
},
connectionName: '' // another optional way to provide connection name if needed
);
Messages bodies are simple objects. A type specifier is required for the message which will be used to set AMQP's properties.type. If no routing key is provided, the type specifier will be used. A routing key of '' will prevent the type specifier from being used.
// the first 3 arguments are required
// routing key is optional and defaults to the value of typeName
// connectionName is only needed if you have multiple connections to different servers or vhosts
rabbit.publish( 'log.entries', 'company.project.messages.logEntry', {
date: Date.now(),
level: logLevel,
message: message
}, 'log.' + logLevel, someValueToCorrelateBy );
This works just like a publish except that the promise returned provides the response (or responses) from the other side.
// when multiple responses are provided, all but the last will be provided via the .progress callback.
// the last/only reply will always be provided to the .then callback
rabbit.request( 'request.exchange', {
// see publish example to see options for the outgoing message
} )
.progress( function( reply ) {
// if multiple replies are provided, all but the last will be sent via the progress callback
} )
.then( function( final ) {
// the last message in a series OR the only reply will be sent to this callback
} );
Handle calls should happen before starting subscriptions.
Message handlers are registered to handle a message based on the typeName. Calling handle will return a reference to the handler that can later be removed. The message that is passed to the handler is the raw Rabbit payload. The body property contains the message body published. The message has ack
, nack
(requeue the message) and reject
(don't requeue the message) methods control what Rabbit does with the message.
In this example, any possible error is caught in an explicit try/catch:
var handler = rabbit.handle( 'company.project.messages.logEntry', function( message ) {
try {
// do something meaningful?
console.log( message.body );
message.ack();
} catch( err ) {
message.nack();
}
} );
handler.remove();
This example shows how to have wascally wrap all handlers with a try catch that:
- nacks the message on error
- console.log that an error has occurred in a handle
// after this call, any new callbacks attached via handle will be wrapped in a try/catch
// that nacks the message on an error
rabbit.nackOnError();
var handler = rabbit.handle( 'company.project.messages.logEntry', function( message ) {
console.log( message.body );
message.ack();
} );
handler.remove();
// after this call, new callbacks attached via handle will *not* be wrapped in a try/catch
rabbit.ignoreHandlerErrors();
Provide a strategy for handling errors to multiple handles or attach an error handler after the fact.
var handler = rabbit.handle( 'company.project.messages.logEntry', function( message ) {
console.log( message.body );
message.ack();
} );
handler.catch( function( err, msg ) {
// do something with the error & message
msg.nack();
} );
Failure to handle errors will result in silent failures and lost messages.
In previous versions, if a subscription was started in ack mode (the default) without a handler to process the message, the message would get lost in limbo until the connection (or channel) was closed and then the messages would be returned to the queue. This is very confusing and undesirable behavior. To help protect against this, the new default behavior is that any message received that doesn't have any elligible handlers will get nack
'd and sent back to the queue immediately.
This is still problematic because it can create churn on the client and server as the message will be redelivered indefinitely.
To change this behavior, use one of the following calls:
Note: only one of these strategies can be activated at a time
rabbit.onUnhandled( function( message ) {
// handle the message here
} );
Sends all unhandled messages back to the queue.
rabbit.nackUnhandled();
Rejects unhandled messages so that will will not be requeued. DO NOT use this unless there are dead letter exchanges for all queues.
rabbit.rejectUnhandled();
Recommendation: set handlers for anticipated types up before starting subscriptions.
Starts a consumer on the queue specified. connectionName
is optional and only required if subscribing to a queue on a connection other than the default one.
Wascally defaults to (and assumes) queues are in ack mode. It batches ack and nack operations in order to improve total throughput. Ack/Nack calls do not take effect immediately.
Enqueues the message for acknowledgement.
Enqueues the message for rejection. This will re-enqueue the message.
Rejects the message without re-queueing it. Please use with caution and consider having a dead-letter-exchange assigned to the queue before using this feature.
Acknowledges the messages and sends the message back to the requestor. The message
is only the body of the reply. Providing true to more
will cause the message to get sent to the .progress callback of the request promise so that you can send multiple replies. The replyType
argument sets the type of the reply message. (important when messaging with statically typed languages)
Reply Queues
By default, wascally creates a unique reply queue for each connection which is automatically subscribed to and deleted on connection close. This can be modified or turned off altogether.
Changing the behavior is done by passing one of three values to the replyQueue
property on the connection hash:
!!! IMPORTANT !!! wascally cannot prevent queue naming collisions across services instances or connections when using the first two options.
Only changes the name of the reply queue that wascally creates - autoDelete
and subscribe
will be set to true
.
rabbit.addConnection( {
name: 'default',
replyQueue: 'myOwnQueue',
user: 'guest',
pass: 'guest',
server: '127.0.0.1',
port: 5672,
timeout: 2000,
vhost: '%2f'
} );
To take full control of the queue name and behavior, provide a queue definition in place of the name.
wascally provides no defaults - it will only use the definition provided
rabbit.addConnection( {
name: 'default',
replyQueue: {
name: 'myOwnQueue',
subscribe: 'true',
durable: true
},
user: 'guest',
pass: 'guest',
server: '127.0.0.1',
port: 5672,
timeout: 2000,
vhost: '%2f'
} );
Only pick this option if request/response isn't in use or when providing a custom overall strategy
rabbit.addConnection( {
name: 'default',
replyQueue: false,
user: 'guest',
pass: 'guest',
server: '127.0.0.1',
port: 5672,
timeout: 2000,
vhost: '%2f'
} );
The call returns a promise that can be used to determine when the exchange has been created on the server.
Valid exchangeTypes:
- 'direct'
- 'fanout'
- 'topic'
Options is a hash that can contain the following:
- autoDelete true|false delete when consumer count goes to 0
- durable true|false survive broker restarts
- persistent true|false a.k.a. persistent delivery, messages saved to disk
- alternate 'alt.exchange' define an alternate exchange
The call returns a promise that can be used to determine when the queue has been created on the server.
Options is a hash that can contain the following:
- autoDelete true|false delete when consumer count goes to 0
- durable true|false survive broker restarts
- exclusive true|false limits queue to the current connection only (danger)
- subscribe true|false auto-start the subscription
- limit 2^16 max number of unacked messages allowed for consumer
- noAck true|false the server will remove messages from the queue as soon as they are delivered
- queueLimit 2^32 max number of ready messages a queue can hold
- messageTtl 2^32 time in ms before a message expires on the queue
- expires 2^32 time in ms before a queue with 0 consumers expires
- deadLetter 'dlx.exchange' the exchange to dead-letter messages to
Binds the target exchange to the source exchange. Messages flow from source to target.
Binds the target queue to the source exchange. Messages flow from source to target.
Note: setting subscribe to true will result in subscriptions starting immediately upon queue creation.
This example shows most of the available options described above.
var settings = {
connection: {
user: 'guest',
pass: 'guest',
server: '127.0.0.1',
port: 5672,
timeout: 2000,
vhost: '%2fmyhost'
},
exchanges:[
{ name: 'config-ex.1', type: 'fanout' },
{ name: 'config-ex.2', type: 'topic', alternate: 'alternate-ex.2', persistent: true },
{ name: 'dead-letter-ex.2', type: 'fanout' }
],
queues:[
{ name:'config-q.1', limit: 100, queueLimit: 1000 },
{ name:'config-q.2', subscribe: true, deadLetter: 'dead-letter-ex.2' }
],
bindings:[
{ exchange: 'config-ex.1', target: 'config-q.1', keys: [ 'bob','fred' ] },
{ exchange: 'config-ex.2', target: 'config-q.2', keys: 'test1' }
]
};
To establish a connection with all settings in place and ready to go call configure:
var rabbit = require( 'wascally' );
rabbit.configure( settings ).done( function() {
// ready to go!
} );
Providing the following configuration options setting the related environment varibles will cause wascally to attempt connecting via AMQPS. For more details about which settings perform what role, refer to the amqplib's page on SSL.
connection: { // sample connection hash
caPath: '', // comma delimited paths to CA files. RABBIT_CA
certPath: '', // path to cert file. RABBIT_CERT
keyPath: '', // path to key file. RABBIT_KEY
passphrase: '', // passphrase associated with cert/pfx. RABBIT_PASSPHRASE
pfxPath: '' // path to pfx file. RABBIT_PFX
}
PRs with insufficient coverage, broken tests or deviation from the style will not be accepted.
PRs should include modified or additional test coverage in both integration and behavioral specs. Integration tests assume RabbitMQ is running on localhost with guest/guest credentials and the consistent hash exchange plugin enabled.
Running gulp will run both sets after every file change and display a coverage summary. To view a detailed report, run gulp coverage once to bring up the browser.
This project has both an .editorconfig
and .esformatter
file to help keep adherance to style simple. Please also take advantage of the .jshintrc
file and avoid linter warnings.
- additional test coverage
- support RabbitMQ backpressure mechanisms
- (configurable) limits & behavior when publishing during connectivity issues
- ability to capture/log unpublished messages on shutdown
- add support for Rabbit's HTTP API
- enable better cluster utilization by spreading connections out over all nodes in cluster