Stephen A. Fuqua (SAF) is a Bahá'í, software developer, and conservation and interfaith advocate in the DFW area of Texas.

Node.js, Web API, and RabbitMQ. Part 2

August 3, 2014

So this is my first project in Node.js. I'm not going to claim any expertise or try to explain the framework. I first learned it in an HTML 5 meetup meeting with Eric Sowell, who also claims not to be a Node.js expert, but he certainly did a great job of clearing the path. His blog post Enough Node.js brings those pieces together in written form and I recommend heading over there my post is diving in too deep right away.

Although Eric's a .Net guy, he seems to like the Mac and SublimeText in his avocational capacity. But for me, I wanted to stick to Visual Studio. So the first thing I did was install the Node.js Tools for Visual Studio. I'm happy to say it is a very functional and useful add-in. Back to Eric's lead, I decided to use express as my node web server. I decided to go with node-amqp for interacting with RabbitMQ. There were other options, but the documentation and age of this project convinced me to give it a shot. CloudAMQP has additional documentation and samples that were quite helpful in pulling this together. As you'll see, the code was quite simple in the end. But, being my first foray into JSON and closures, a lot more learning went into this than is readily apparent.

Enough talk. Code. Setup the dependencies and the shell of the web server. Part 1 describes the expected functionality and the RESTful format. This lets us use http://localhost:10025/Message/helloWorld.

/**
 * Module dependencies.
 */
var express = require('express');
var http = require('http');
var amqp = require('amqp');
 
/**
 * Web Server setup.
 */
var app = express();
 
app.set('port', process.env.PORT || 10025);
 
/**
 * A RESTful GET request handler
 */
app.get('/Message/:message', function(request, response) {
 
    var message = request.params.message;
    console.log('Received message: ' + message);
    
    // rabbit stuff goes here

    response.writeHead(202);
    response.end();
});
 
/**
 * Start the web server
 */
http.createServer(app).listen(app.get('port'), function () {
    console.log('Express server listening on port ' + app.get('port'));
});

It was great to learn, thanks to some post or other, about good use of JavaScript namespaces and using || as a null-coalescing operator. console.log(...) was another win that only recently came to my attention.

Now, what about interacting with RabbitMQ? Let's have a brief digression into service bus theory. Again, this is far from expert commentary. I've learned a publish/subscribe methodology, where point A publishes a Message to an Exchange. Point B subscribes to a Queue and binds that Queue to one or more Exchanges. The Exchange receives a Message and sends it to all Queues subscribed to that Exchange (fanout model). If multiple clients are subscribed to the same Queue, then only one of them gets the message. But if multiple clients are subscribed to different Queues, which are bound to the same Exchange, then all of them get the Message.

Thus, in the code below, we must

1. Open a connection to the bus;
2. Create or connect to an exchange;
3. Subscribe to a queue;
4. Handle messages received through that queue.

Note that the nested functions are closures / callbacks - when one action completes, it calls the next action. This is part of the asynchronous nature of Node.js. Using a closure (which is just an anonymous method in .Net terms) as a callback guarantees that the method runs after the call completes. However, I ran into a problem that I haven't figured out yet (nor did I try hard – since I don't need to be an expert). Originally I had a callback when opening the exchange; the connection to the queue was in that callback. If the exchange and queue already exist then this code works. But when the exchange is being created for the first time, the callback is never executed. When I pulled the queue creation into the next line after the exchange, instead of nested in a callback, everything worked – as you can see in the RabbitMQ message displayed below the code.

var mqUrl = process.env.RABBITMQ_URL || "amqp://localhost";
var exchangeName = process.env.EXCHANGE || "RabbitMQ_web_api_testing";
var queueName = process.env.QUEUE || "apitest_webapi_simple";
 
console.log('Opening connection to RabbitMQ');
var connection = amqp.createConnection({ url: mqUrl }, {
    reconnect: true, // Enable reconnection
    reconnectBackoffStrategy: 'linear',
    reconnectBackoffTime: 1000, // Try reconnect once a second
});
    
 
connection.on('ready', function() {
    console.log('Creating/opening exchange: ', exchangeName);
    var exchange = connection.exchange(exchangeName);
 
    console.log('Creating/opening queue: ', queueName);
    connection.queue(queueName, function(queue) {
 
        console.log('Binding queue to exchange');
        queue.bind(exchange, queue.name);
 
        console.log('Publishing message');
 
        var publishMessage = {
            body: message
        };
 
        exchange.publish(queue.name, publishMessage);
 
        console.log('Done');
    });
});

And here's the message that is sitting in the apitest_webapi_simple queue in RabbitMQ:

Exchange	RabbitMQ_web_api_testing
Routing Key	apitest_webapi_simple
Redelivered	∘
Properties	content_type: application/json
Payload 38 bytes Encoding: string	{"body":"one+small+step+for+a+man..."}

My automated test fails, but at least I've established communication with the bus. The next post will look at the equivalent code for Web API.