Rabbit-tools

This library is aimed to ease amqplib usage.

Installation:

npm i rabbitmq-tools -S

Following features are implemented:

• Client reconnection
• Failed channel reopening
• Routing and message validation via jsonschema
• Promise based querying interface that is curried by default

Usage

The library allow you to easily handle amqp routing and request/publish routines. The simpliest usage is:

const client = ReactiveMQ.create({
  url,
  appId,
  routerConfig: {
    routes: [
      {
        exchange: 'amq.topic',
        routingKey: 'hardware.v1.created',
        resolver: () => {}
      }
    ]
  }
})

client.publish('amq.topic', 'routing.key', { foo: 'bar' })
client.request('amq.topic', 'routing.key', { foo: 'bar' })
  .then(response => {})

Reference

RxConnection

connect(url, options)

returns Rx.BehaviorSubject instance emitting amqp.connection or null in case of no connection/connection close

Params

url (string | object): amqp url string or object, same as described in amqplib

options.logger ({ log, warn } | null | false): logger to output connection state changes. Default is console. Set falsy value to disable logging

import { connect } from 'rabgbitmq-tools'

const rxConnection = connect(url, options)

RxChannel

openChannel(rxConnection, options)

returns Rx.BehaviorSubject instance emitting amqp.channel or null in case of no connection/connection close

Params

rxConnection (Rx.BehaviorSubject): Rx.BehaviorSubject instance emitting amqp.connection or null in case of no connection/connection close

options.logger ({ log, warn } | null | false): logger to output connection state changes. Default is console. Set falsy value to disable logging

import { openChannel } from 'rabgbitmq-tools'

const rxChannel = openChannel(rxConnection, options)

class ReactiveMQ

constructor(options)

Constructor params

options.url (string | object): amqp url string or object, same as described in amqplib

options.appId (string): application identifier used in publishing/request logic. read more...

options.connectionId (string): application identifier used as a prefix in logging. Consists of logging scope and vhost if available

options.logger ({ log, warn } | null | false): logger to output connection state changes. Default is console. Set falsy value to disable logging

options.routerConfig ({ routes: object }): see Router doc below

const client = ReactiveMQ.create({
  url: 'rabbitMQUrl',
  appId: 'appId'
})

Instance methods

publish(exchange, routingKey, message)

returns Promise

exchange (string): amqp exchange routingKey (string): amqp routingKey to publish to message (any): data to buplish

method is curried by default

request(exchange, replyTo, routingKey, message)

returns Promise

exchange (string): amqp exchange replyTo (string): queue to receive a reply routingKey (string): amqp routingKey to publish to message (any): data to buplish

method is curried by default

connectRouter(routerConfig)

returns void

options.routerConfig ({ routes: object }): see Router doc below

This is useful if you need to provide routes after client initialization. Starts up router and does all the logic related to listening to events.

class Router

constructor(options)

Constructor params

options.appId (string): application identifier used in publishing/request logic. read more...

options.connectionId (string): application identifier used as a prefix in logging. Consists of logging scope and vhost if available

options.logger ({ log, warn } | null | false): logger to output connection state changes. Default is console. Set falsy value to disable logging

options.routes ([object]): actual router config representing queue and consumer options, validation and handling

options.routes.exchange (string): exchange to bind queue to

options.routes.routingKey (string): routingKey to bind queue to

options.routes.resolver (function): event handler that is passed a message payload and expected to return any output that may be received by initial emitter

options.routes.requestSchema (json): json schema object to validate a message. read more...

options.routes.queueOptions (object): options passed to channel.bindQueue. read more...

options.routes.consumerOptions (object): options passed to channel.consume. read more...

Roadmap

• Implement more flexible amqp configurations
• Cover with tests
• Improve documentation