Shedding The Monolithic Application With AWS Simple Queue Service (SQS) And Node.js

For too long, I have lived in the world of the monolithic web application - the single application that does everything for everyone. This year, I really want to evolve my understanding of web application architecture and try to think about applications as a collection of loosely coupled services. I think an important first step, in this journey, is to start experimenting with messages queues.

Message queues provide an asynchronous means of communication between components within a system. One (or more) components generates messages and pushes them onto a queue; then, one (or more) components asynchronously pops messages off of that queue and processes them. The beauty of the shared queue is that the relevant parts of the system become independently scalable and much more resilient to failure.

But, I'm new to message queues; so, I'll refrain from speaking out of turn. Rather, I'd like to share my experiment. Since this is my first play with message queues, the Amazon Web Service (AWS), Simple Queue Service (SQS), seemed like a really easy way to begin. I already have an AWS account; and, with the AWS Software Developer's Kit (SDK), jumping right into code is straightforward.

NOTE: Unlike some message queue services, SQS does not guarantee "one time" delivery; rather, it guarantees "at least one time" delivery. Meaning, due to the distributed natures of SQS, a message may be received more than one time, even after it has been deleted.

In addition to playing with message queues, I also want to start getting better at Node.js. While I love JavaScript with the rich, fiery passion of a thousand suns, I've never actually built any production components with Node.js. And, since I'm in the midst of rethinking application architecture, I might as well do that with Node.js.

Now, out of the box, I could have started using the standard error / callback pattern that has been popular in the Node.js community (and core codebase). However, coming from client-side JavaScript - and AngularJS - I have trouble imagining a world without Promises. As such, I'm also using this as an opportunity to see how people are using libraries like Q to "de-nodeify" the old-school approaches to workflow.

To start with, I want to keep it simple. I've created a Node.js script that sends a one-off message to SQS. And, I've created a Node.js script that (long) polls the same message queue, looking for messages to process.

Here is the Node.js script that sends the message:

// Require the demo configuration. This contains settings for this demo, including // the AWS credentials and target queue settings. var config = require( "./config.json" ); // Require libraries. var aws = require( "aws-sdk" ); var Q = require( "q" ); var chalk = require( "chalk" ); // Create an instance of our SQS Client. var sqs = new aws.SQS({ region: config.aws.region, accessKeyId: config.aws.accessID, secretAccessKey: config.aws.secretKey, // For every request in this demo, I'm going to be using the same QueueUrl; so, // rather than explicitly defining it on every request, I can set it here as the // default QueueUrl to be automatically appended to every request. params: { QueueUrl: config.aws.queueUrl } }); // Proxy the appropriate SQS methods to ensure that they "unwrap" the common node.js // error / callback pattern and return Promises. Promises are good and make it easier to // handle sequential asynchronous data. var sendMessage = Q.nbind( sqs.sendMessage, sqs ); // ---------------------------------------------------------- // // ---------------------------------------------------------- // // Now that we have a Q-ified method, we can send the message. sendMessage({ MessageBody: "This is my first ever SQS request... evar!" }) .then( function handleSendResolve( data ) { console.log( chalk.green( "Message sent:", data.MessageId ) ); } ) // Catch any error (or rejection) that took place during processing. .catch( function handleReject( error ) { console.log( chalk.red( "Unexpected Error:", error.message ) ); } );

As you can see, I'm taking the sqs.sendMessage() method and wrapping it in a Q-based proxy. This proxy will translate the error / callback pattern into one that properly resolves or rejects a promise. This way, I can use .then() and .catch() to handle various states of the promise chain.

Once I could get messages onto the SQS message queue, I created a script that could receive them. But, unlike the previous script, that only performed one-off operations, I wanted this receiving script to continuously poll for new messages.

// Require the demo configuration. This contains settings for this demo, including // the AWS credentials and target queue settings. var config = require( "./config.json" ); // Require libraries. var aws = require( "aws-sdk" ); var Q = require( "q" ); var chalk = require( "chalk" ); // Create an instance of our SQS Client. var sqs = new aws.SQS({ region: config.aws.region, accessKeyId: config.aws.accessID, secretAccessKey: config.aws.secretKey, // For every request in this demo, I'm going to be using the same QueueUrl; so, // rather than explicitly defining it on every request, I can set it here as the // default QueueUrl to be automatically appended to every request. params: { QueueUrl: config.aws.queueUrl } }); // Proxy the appropriate SQS methods to ensure that they "unwrap" the common node.js // error / callback pattern and return Promises. Promises are good and make it easier to // handle sequential asynchronous data. var receiveMessage = Q.nbind( sqs.receiveMessage, sqs ); var deleteMessage = Q.nbind( sqs.deleteMessage, sqs ); // ---------------------------------------------------------- // // ---------------------------------------------------------- // // When pulling messages from Amazon SQS, we can open up a long-poll which will hold open // until a message is available, for up to 20-seconds. If no message is returned in that // time period, the request will end "successfully", but without any Messages. At that // time, we'll want to re-open the long-poll request to listen for more messages. To // kick off this cycle, we can create a self-executing function that starts to invoke // itself, recursively. (function pollQueueForMessages() { console.log( chalk.yellow( "Starting long-poll operation." ) ); // Pull a message - we're going to keep the long-polling timeout short so as to // keep the demo a little bit more interesting. receiveMessage({ WaitTimeSeconds: 3, // Enable long-polling (3-seconds). VisibilityTimeout: 10 }) .then( function handleMessageResolve( data ) { // If there are no message, throw an error so that we can bypass the // subsequent resolution handler that is expecting to have a message // delete confirmation. if ( ! data.Messages ) { throw( workflowError( "EmptyQueue", new Error( "There are no messages to process." ) ) ); } // --- // TODO: Actually process the message in some way :P // --- console.log( chalk.green( "Deleting:", data.Messages[ 0 ].MessageId ) ); // Now that we've processed the message, we need to tell SQS to delete the // message. Right now, the message is still in the queue, but it is marked // as "invisible". If we don't tell SQS to delete the message, SQS will // "re-queue" the message when the "VisibilityTimeout" expires such that it // can be handled by another receiver. return( deleteMessage({ ReceiptHandle: data.Messages[ 0 ].ReceiptHandle }) ); } ) .then( function handleDeleteResolve( data ) { console.log( chalk.green( "Message Deleted!" ) ); } ) // Catch any error (or rejection) that took place during processing. .catch( function handleError( error ) { // The error could have occurred for both known (ex, business logic) and // unknown reasons (ex, HTTP error, AWS error). As such, we can treat these // errors differently based on their type (since I'm setting a custom type // for my business logic errors). switch ( error.type ) { case "EmptyQueue": console.log( chalk.cyan( "Expected Error:", error.message ) ); break; default: console.log( chalk.red( "Unexpected Error:", error.message ) ); break; } } ) // When the promise chain completes, either in success of in error, let's kick the // long-poll operation back up and look for moar messages. .finally( pollQueueForMessages ); })(); // When processing the SQS message, we will use errors to help control the flow of the // resolution and rejection. We can then use the error "type" to determine how to // process the error object. function workflowError( type, error ) { error.type = type; return( error ); }

As you can see, I'm once again wrapping the relevant SQS methods in a Q-based proxy so that both sqs.receiveMessage() and sqs.deleteMessage() work with promises. For me personally, this just makes the workflow easier to follow.

To get this script to run continuously, I defined the long-polling inside of a self-executing function block that could recursively call itself. This way, as soon as one request completes, either in success or in error, the long poll will be kicked back up, in search of the next message to process.

If I run these scripts side-by-side, I get the following terminal output:

When I saw this work for the first time, I almost jump out of my seat with excitement! Obviously, there's a lot more to consider when working with message queues; but, seeing these messages flow from sender to receiver via asynchronous queue feels like a huge milestone in my evolution as a developer.

Tweet This Deep thoughts by @BenNadel - Shedding The Monolithic Application With AWS Simple Queue Service (SQS) And Node.js Woot woot — you rock the party that rocks the body!







