Separation Of Concerns When Consuming Amazon SQS Queues In Lucee CFML 5.3.8.201
Last week, I started to explore the consumption of Amazon SQS (Simple Queue Service) in a Lucee CFML application. That first post was a low-level look at the mechanics of using the AWS (Amazon Web Services) Java SDK to add and remove messages to and from a given queue, respectively. Today, I wanted to step back and start thinking about the separation of concerns in the larger ColdFusion application context.
View this code in my Amazon SQS Lucee CFML project on GitHub.
In last week's post, I read messages from the Amazon SQS queue by using a <meta> refresh tag to perform the "long polling". And, when a message was received from the queue, there was no additional processing to be done - I just output the contents of the message to the screen. To make things more interesting in today's post, I want to add some complexity:
I want to create an actual "background thread" for the long polling of the queue.
I want to perform some additional processing based on the received message.
The "processing" in this case will be taking HEX (hexadecimal) color values, such as FF3366, from the SQS messages and converting them into color swatch images that can be rendered in the ColdFusion application. And, for the background thread, I'll be using an old-school ColdFusion scheduled task.
Now that we have more moving parts, we need to figure out which components are responsible for which aspects of the message queue workflow. And, when I think about dividing responsibilities, I try to think about "how little" each component can know - what is the least that each component can be responsible for. This approach creates constructs that can then be "swapped out" or refactored more easily.
At least, that's the hope.
The first ColdFusion component that I created in this exploration generated the actual color swatch. I was able to create this ColdFusion component before I even created my Amazon SQS queue because generating color swatches has nothing to do with queues.
component
output = false
hint = "I generate color swatch images from 6-digit hexadecimal color values."
{
/**
* I initialize the color swatch service.
*/
public void function init() {
variables.swatchWidth = 200;
variables.swatchHeight = 150;
}
// ---
// PUBLIC METHODS.
// ---
/**
* I generate and return a color swatch image for the given hexadecimal value.
*/
public struct function generateSwatch( required string hexColor ) {
var normalizedHex = ( "##" & hexColor.right( 6 ).ucase() );
var annotationFont = {
font: "monospace",
size: 16
};
var labelWidth = 92;
var labelHeight = 36;
var swatch = imageNew( "", swatchWidth, swatchHeight, "rgb", normalizedHex )
.setAntialiasing( true )
.setDrawingColor( "ffffff" )
.drawRect( 0, ( swatchHeight - labelHeight ), labelWidth, labelHeight, true )
.setDrawingColor( "000000" )
.drawText( normalizedHex, 10, ( swatchHeight - 11 ), annotationFont )
;
return( swatch );
}
/**
* I generate a color swatch image for the given hexadecimal value and save it to the
* given filename.
*/
public void function generateSwatchFile(
required string hexColor,
required string destination
) {
var quality = 1;
var overwrite = true;
var noMetaData = true;
// NOTE: Instead of passing-in variables to the .write() method, I would normally
// just used named-arguments. However, it seems that attempting to call the
// .write() method with named arguments throws an error. Perhaps the documented
// argument names are wrong.
generateSwatch( hexColor )
.write( destination, quality, overwrite, noMetaData )
;
}
}
As you can see, all this ColdFusion component does is take a HEX color and a file destination and generate a PNG image. It doesn't know anything about queues; and, we could easily consume this component directly from any part of our ColdFusion application. But, for the sake of discussion, let's assume that generating the image is CPU and time intensive; and, that it would be untenable for a user to submit a HEX color and wait for the image generation in the same HTTP request-response life-cycle. As such, we're going to decouple the submission of the HEX color from the image processing through the use of our Amazon SQS queue.
Which means, when the user submits the HEX color, the only action that we're going to take is to push a new message onto the queue. Here's what the user interface looks like:
<cfscript>
param name="form.hexColor" type="string" default="";
// If we have a new hex color to process, put it on the message queue - a background
// thread will monitor the queue for new messages and generate the color swatch image
// asynchronously.
if ( form.hexColor.len() ) {
application.colorSwatchQueueService.addMessage( form.hexColor );
}
// Query for the existing color swatches. Since we don't have a database, we'll just
// read the images right off of the file-system.
swatches = directoryList(
path = expandPath( "/swatches" ),
listInfo = "name",
type = "file",
filter = "*.png"
);
swatches.sort( "textnocase", "desc" );
</cfscript>
<cfoutput>
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<title>
Separation Of Concerns When Consuming Amazon SQS Queues In Lucee CFML 5.3.8.201
</title>
<link rel="stylesheet" type="text/css" href="./index.css" />
<!---
If we just submitted a hexColor, let's automatically refresh the page in a
few seconds. For this particular demo, a few seconds is all it takes for the
ColdFusion application to poll the queue, get the new messages, and generate
the associated color swatch image. Refreshing the page will, therefore, read
the new image off the file-system.
--->
<cfif form.hexColor.len()>
<meta http-equiv="refresh" content="2" />
</cfif>
</head>
<body>
<h1>
Separation Of Concerns When Consuming Amazon SQS Queues In Lucee CFML 5.3.8.201
</h1>
<form method="post" action="./index.cfm">
<input
type="text"
name="hexColor"
placeholder="Enter hex color..."
size="20"
maxlength="6"
autofocus
/>
<button type="submit">
Generate swatch
</button>
</form>
<hr />
<h2>
Existing Swatches
</h2>
<ul>
<cfloop value="filename" array="#swatches#">
<li>
<img src="./swatches/#filename#" />
</li>
</cfloop>
</ul>
</body>
</html>
</cfoutput>
Note that when the user submits the form with the hex color, all we're doing is adding the hex color to the queue via:
colorSwatchQueueService.addMessage( form.hexColor )
The ColorSwatchQueueService is the glue component that binds the Amazon SQS queue to the greater ColdFusion application. This is the only component that knows about the SQS queue, that understands the structure of the SQS messages, and that knows which other ColdFusion components to invoke in response to SQS messages. It is the "traffic cop" for messages.
And, it is the component implementation that we could theoretically swap out if we needed to move from Amazon SQS to another queue provider, such as RabbitMQ. Or, if we needed to completely change the underlying mechanics, perhaps switching from a queue workflow over to something like Redis' blocking list operations.
Since this component is the "glue", when we instantiate it we need to provide both the SqsClient and the ColorSwatchService in the constructor:
component
output = false
hint = "I provide methods for interacting with the color-swatch-queue - I GLUE the concept of the queue to larger application context."
{
/**
* I initialize the color swatch queue service with the given SQS client. Note that
* the SQS client is assumed to be created specifically for the color-swatch-queue
* in Amazon SQS.
*/
public void function init(
required any sqsClient,
required any colorSwatchService
) {
variables.sqsClient = arguments.sqsClient;
variables.colorSwatchService = arguments.colorSwatchService;
}
// ---
// PUBLIC METHODS.
// ---
/**
* I create and persist a new message for processing the given HEX color.
*/
public void function addMessage( required string hexColor ) {
sqsClient.sendMessage(
serializeJson({
hexColor: hexColor
})
);
}
/**
* I delete the given message from the queue.
*/
public void function deleteMessage( required struct message ) {
sqsClient.deleteMessage( message.receiptHandle );
}
/**
* I process the given message, translating the message into an interaction with the
* rest of the application logic. In this case, we're generating a color swatch image
* for the hexColor contained within the SQS message.
*/
public void function processMessage( required struct message ) {
var body = deserializeJson( message.body );
var hexColor = body.hexColor;
var filename = getFilenameForHex( hexColor );
var destination = expandPath( "/swatches/#filename#" );
systemOutput( "Generating color swatch for ###hexColor.ucase()#", true );
colorSwatchService.generateSwatchFile( hexColor, destination );
}
/**
* I look for new messages on the queue.
*
* CAUTION: While this request will block-and-wait for new messages to arrive (if
* waitTime argument is non-zero), it will do so only once. We are not putting the
* onus of continual polling inside this component. Instead, we are placing that
* responsibility in another area of the app (in this demo, a scheduled task).
*/
public void function processNewMessages(
numeric maxNumberOfMessages = 3,
numeric waitTime = 20,
numeric visibilityTimeout = 10
) {
var messages = sqsClient.receiveMessages( argumentCollection = arguments );
for ( var message in messages ) {
// Since we are gathering more than one message at a time in this demo (in
// order to reduce the dollars-and-cents cost of making API calls to Amazon
// SQS), we want to wrap each message processing in a try-catch so that one
// "poison pill" doesn't prevent the other messages from being processed.
try {
processMessage( message );
} catch ( any error ) {
systemOutput( "A color-swatch-queue message failed to process.", true, true );
systemOutput( message, true, true );
systemOutput( error, true, true );
}
deleteMessage( message );
}
}
// ---
// PRIVATE METHODS.
// ---
/**
* I generate a natural-sort-friendly filename for the given hexColor.
*/
private string function getFilenameForHex( required string hexColor ) {
// For this demo, we know that the files are going to be read directly off of the
// local file-system. As such, if we prefix each color swatch with a date/time
// stamp, we know that we can list the newest swatches first using an alpha-
// numeric sort on the file names.
return( now().dateTimeFormat( "yyyy-mm-dd HH-nn-ss" ) & "-" & hexColor.right( 6 ) & ".png" );
}
}
In this implementation, the processMessage() method knows how to translate a single SQS message into a color swatch PNG that is persisted to the file-system. And, the processMessages() method knows how to perform a single long-poll operation on the the Amazon SQS queue. But, this component knows nothing about the "background thread" - it has no idea how long-polling is performed over the full life-cycle of the application.
Is the application using Lucee CFML's "Task thread" concept?
Is the application using a cron job?
Is the application using the ColdBox's Async library?
Is there a room full of people just hitting "Refresh" on a page that invokes the processMessages() method?
The ColorSwatchQueueService doesn't have to know or care how the long-polling is being performed. And, for the sake of this demo, I'm going to implement the long-term polling using a traditional ColdFusion scheduled task. My scheduled task will execute every 10-seconds; and, will perform a while() loop that calls the processMessages() method over-and-over again:
<cfscript>
// This template is being invoked as a ColdFusion Scheduled Task that fires every
// 10-seconds (the smallest scheduled task increment in Lucee CFML). This means that
// any given request will almost always be overlapping with the previous, still-
// executing request. As such, let's use a no-error lock to synchronize our
// monitoring of the color swatch queue.
lock
name = "color-swatch-queue-manager"
type = "exclusive"
timeout = 1
throwOnTimeout = false
{
systemOutput( "Start polling color-swatch-queue for new messages.", true );
// When we ask the queue service to process new messages, it only processes one
// batch of new messages at a time. This way, we can separate the concern of
// processing new messages from the concern of polling the queue over a long
// period of time. Due to the constraints of the web server, requests timeout if
// they run for too long. As such, we have to explicitly increase the timeout of
// this page so that it doesn't get terminated forcefully by the server.
// --
// NOTE: For the demo, I'm using a relatively low-number so that as I edit this
// template and I don't have to restart the server to kill this thread. But, in a
// production setting, I would use a rather large number.
maxRuntimeInSeconds = 100;
maxTickCount = ( getTickCount() + ( maxRuntimeInSeconds * 1000 ) );
// Increase the execution timeout for this web-server request.
setting
requestTimeout = maxRuntimeInSeconds
;
// Since each call to the queue service will only process a single batch of new
// messages, we have to continually loop in order to keep polling the queue.
while ( getTickCount() <= maxTickCount ) {
systemOutput( "LOOP: Checking color-swatch-queue for new messages.", true );
application.colorSwatchQueueService.processNewMessages();
}
systemOutput( "LOOP: Exiting long-polling while-loop.", true );
}
// If a previous instance of the scheduled task was still polling the message queue,
// the lock will have failed to be obtained.
if ( ! cflock.succeeded ) {
systemOutput( "Lock failure, color-swatch-queue monitoring already in place.", true, true );
}
</cfscript>
For the sake of my development process, the while() loop in this ColdFusion scheduled task only runs for up-to 100-seconds. This way, as I edit the code, I only have to wait (at most) 100-seconds for the new code to be picked-up. In a production environment, I'd likely have the while() loop run for much longer.
And, since the while() loop runs for a long period of time, it means that subsequent executions of the ColdFusion scheduled task will likely overlap with previous invocations that are still executing. In order to prevent overlap, I'm using a no-error CFLock tag to synchronize execution. This way, if one scheduled task runs while the previous one is still running, the new one will wait 1-second (the shortest timeout a CFLock tag can take) and then quietly exit the request.
Now that we know what each component in this ColdFusion application is doing, here's a high-level overview of how they all fit together:
And, if we run this Lucee CFML application and start entering HEX color values, here's the experience we get:
And, if we look at the server logging that we're getting, we see:
[INFO ] Start polling color-swatch-queue for new messages.
[INFO ] LOOP: Checking color-swatch-queue for new messages.
[ERROR] Lock failure, color-swatch-queue monitoring already in place.
[INFO ] Generating color swatch for #FF3366
[INFO ] LOOP: Checking color-swatch-queue for new messages.
[ERROR] Lock failure, color-swatch-queue monitoring already in place.
[INFO ] Generating color swatch for #00AACC
[INFO ] LOOP: Checking color-swatch-queue for new messages.
[INFO ] Generating color swatch for #3399CC
[INFO ] LOOP: Checking color-swatch-queue for new messages.
[ERROR] Lock failure, color-swatch-queue monitoring already in place.
[INFO ] Generating color swatch for #00FFCC
[INFO ] LOOP: Checking color-swatch-queue for new messages.
[INFO ] Generating color swatch for #FF0088
[INFO ] LOOP: Checking color-swatch-queue for new messages.
[ERROR] Lock failure, color-swatch-queue monitoring already in place.
[INFO ] Generating color swatch for #9F0CAA
[INFO ] LOOP: Checking color-swatch-queue for new messages.
[ERROR] Lock failure, color-swatch-queue monitoring already in place.
[ERROR] Lock failure, color-swatch-queue monitoring already in place.
The [ERROR] lines that we're seeing are the overlapping scheduled task executions. This is where the CFLock tag waits for 1-second and then quietly exits since there is another "background thread" currently long-polling the Amazon SQS queue.
And, again, I'm using a ColdFusion scheduled task to implement the long-polling here. But, we could swap that implementation out with any number of other options and the overall architecture would remain exactly the same since we drew tight boundaries around the responsibilities of each component in this architecture.
For completeness, here's the Application.cfc ColdFusion framework component that ties it all together:
component
output = false
hint = "I define the application settings and event handlers."
{
// Configure the application settings.
this.name = "ColorSwatchesQueueDemo";
this.applicationTimeout = createTimeSpan( 0, 1, 0, 0 );
this.sessionManagement = false;
this.setClientCookies = false;
this.mappings = {
"/swatches": "./swatches",
"/vendor": "../../vendor"
};
// ---
// LIFE-CYCLE METHODS.
// ---
/**
* I get called once when the application is being initialized.
*/
public void function onApplicationStart() {
var config = deserializeJson( fileRead( "./config.json" ) );
// This service generates color swatches and KNOWS NOTHING about Amazon SQS.
application.colorSwatchService = new ColorSwatchService();
// This service interacts specifically with the "color-swatch-queue" but KNOWS
// NOTHING about color watches, how they are used within this application, or how
// the queue will be monitored on an ongoing basis.
application.sqsClient = new SqsClient(
classLoader = new AwsClassLoader(),
accessID = config.aws.accessID,
secretKey = config.aws.secretKey,
region = config.aws.region,
queueName = config.aws.queue, // This component instance if QUEUE SPECIFIC.
defaultWaitTime = 20,
defaultVisibilityTimeout = 60
);
// This service is the TRANSLATION GLUE between the Amazon SQS client and the
// application's business logic. However, it KNOWS NOTHING about how the queue
// will be monitored on an ongoing basis.
application.colorSwatchQueueService = new ColorSwatchQueueService(
sqsClient = application.sqsClient,
colorSwatchService = application.colorSwatchService
);
// For this demo, this scheduled task will be the only thing in the application
// that manages the monitoring of the queue over the long-term. However, it
// doesn't actually know anything about the queue, the color swatches, or how
// they are used within the application - the scheduled task ONLY KNOWS about the
// ColorSwatchQueueService component (and its ".processNewMessages()" method).
schedule
action = "update"
task = "ColorSwatchQueueManager"
operation = "HTTPRequest"
url = "http://#cgi.server_name#:#cgi.server_port#/demos/color-swatches/color-swatch-queue-manager.cfm"
startDate = "2021-09-10"
startTime = "00:00 AM"
interval = 10 // Every 10 seconds (smallest increment allowed in Lucee CFML).
;
}
/**
* I get called once when the request is being initialized.
*/
public void function onRequestStart() {
// If the INIT flag is defined, restart the application in order to refresh the
// in-memory cache of components.
if ( url.keyExists( "init" ) ) {
applicationStop();
location( url = cgi.script_name, addToken = false );
}
}
}
I'm not going to bother showing the SqsClient.cfc code since it's exactly the same was is was in the last post and can be seen in the GitHub project.
A few days ago, when I started thinking about consuming Amazon SQS queues in ColdFusion, it felt a bit overwhelming - a lot of unknowns. But, as I've begun to baby-step my way to a solution, breaking the problem down into small parts and then trying to find the "simplest approach that works", it suddenly starts to feel quite manageable. Honestly, I'm not even all that bothered by the use of a ColdFusion scheduled task to power the long-term polling. In fact, that feels like "good enough" solution to me.
Epilogue on Long-Running Requests and Application Performance Monitoring (APM)
In any application, having long-running tasks can have a negative impact on Application Performance Monitor (APM) - that is, the ability to see the overall runtime health of the application. The vast majority of requests flowing through a ColdFusion application are very fast. But, when you add a request that purposefully runs for thousands of seconds, it can dilute, skew, and obscure the meaning of measurements like Apdex scores and P95 response times.
That said, most APM instrumentation provides some means to ignore long-running requests that are known to be long-running. The NewRelic Java Agent provides methods like ignoreApdex() and ignoreTransaction() and the FusionReactor Java Agent almost certainly provides similar techniques.
In other words, you don't want to architecture your application around your APM instrumentation; instead, you want to keep your architecture simple and configure your APM instrumentation accordingly.
Want to use code from this post? Check out the license.
Reader Comments
Post A Comment — ❤️ I'd Love To Hear From You! ❤️
Post a Comment →