fh-amqp-js-test

fh-amqp-js

FeedHenry AMQP client wrapper for communication with a Rabbit cluster.

Overview

The FeedHenry Platform includes a RabbitMQ cluster, which is set up to have highly available Mirrored Queues. The messaging pattern that best suits our development requirements is the 'Topic' pattern, described in the RabbitMQ documentation.

This client library helps with the following:

• Automatic detection of Rabbit node failure and reconnecting to another node in the Rabbit Cluster.

• 'publishTopic' and 'subscribeToTopic' type functions that set up correct Exchange and Q configuration behind the scenes.

You probably won't, but if you think you need to use a different Q type or Message Q Pattern, you will need to either extend this library, or use the AMQP driver directly.

This module can also be used from the command line, to quick publish a message to the FeedHenry Rabbit Cluster, or also as a handy way to quickly subscribe to FeedHenry messages.

Module Usage

Sample sub.js:

var fhamqpjs = require('fh-amqp-js');
var cfg = {
  clusterNodes: ["amqp://guest:[e-mail]:5672/fh-events", "amqp://guest:[e-mail]:5673/fh-events"],
  maxReconnectAttempts: 10  // specifies how may connection failures to attempt before giving up.
};
 
var amqpManager = new fhamqpjs.AMQPManager(cfg);
amqpManager.connectToCluster();
 
// Note that the 'connection' event will fire every time a connection is made to a different Rabbit Node in the cluster (i.e. if the first node we're connected to fails, this connection event will fire when we get connected to the next node in the cluster. Somewhat suboptimal but can't do much about it currently.
amqpManager.on("connection", function(){
  amqpManager.subscribeToTopic("fh-event-exchange", "my-event-q", "my-topic-filter.#", subscribeFunc, function(err){
    if(err) console.error("Fatal error setting up subscriber: ", err); 
...
  });
});
 
// error handler: something fatal (like can't connect to any nodes in a cluster) needs to happen for this to fire.
amqpManager.on("error", function(err){
  console.error("Fatal error: ", err);
});
 
// the function that gets called each time a message is recieved
function subscribeFunc (json, headers, deliveryInfo) {
  console.log("GOT MESSAGE: ", json);
};

Sample pub.js:

var fhamqpjs = require('fh-amqp-js');
var cfg = {
  clusterNodes: ["amqp://guest:[e-mail]:5672/fh-events", "amqp://guest:[e-mail]:5673/fh-events"],
  maxReconnectAttempts: 10
};
 
var amqpManager = new fhamqpjs.AMQPManager(cfg);
amqpManager.connectToCluster();
 
// note we clear the timer on each (re)connection
var t; var count=0;
amqpManager.on("connection", function(){
  if (t) clearInterval(t);
  t = setInterval(function(){
    amqpManager.publishTopic("fh-event-exchange", "topic", {count: count++}, function(err){
      if (err) console.error("Fatal publishing error: ", err);
    });
  }, 1000);
});
 
amqpManager.on("error", function(err){
  console.error("Fatal error: ", err);
});

Note that the vhosts, Exchange names and Topics used internally in FeedHenry are documented here: TODO.

CLI Usage

Usage: fh-amqp-js pub <exchange> <topic> <message> --clusterNodes=[<amqp-url>,*]
fh-amqp-js sub <exchange> <topic> --clusterNodes=[<amqp-url>,*]

The Command Line Interface can be used to quickly publish messages, e.g.

$ fh-amqp-js pub "fh-topic2" "fh.event.count" '{"count": 1}' --clusterNodes='["amqp://guest:[e-mail]:5672"]'

There is also a 'sub' command, for quickly subscribing to messages:

$ fh-amqp-js sub "fh-topic2" "fh.event.count" --clusterNodes='["amqp://guest:[e-mail]:5672"]'

Configuration:

The CLI uses the RC node module for incredibly flexible config finding (see its documentation). Config options currently are:

{
  clusterNodes: ["amqp://guest:[e-mail]:5672/fh-events", "amqp://guest:[e-mail]:5673/fh-events"],
  maxReconnectAttempts: 10  // specifies how may connection failures to attempt before giving up.
}

Development

To run the tests:

make test

or:

make test-coverage-cli

or:

make test-coverage-html

Build artifacts are located on Denzil here: http://denzil.henora.net:8080/view/common/job/fh-amqp-js/