Camel Component For Amazon SQS

As its name implies Amazon Simple Queue Service (Amazon SQS) is a simplified version of a Messaging system. The great thing about it is its reliability and high scalability features. Amazon SQS makes it easy to build an automated workflow, working in close conjunction with the Amazon Elastic Compute Cloud (Amazon EC2) and the other AWS infrastructure web services. You can learn more about Amazon SQS here.

The Problem
My colleagues (Tom Purcell and Rod Biresch) and myself have been working on a demo application for Chariot Solutions' Emerging Technologies for the Enterprise conference and the need for a transport component to simplify the access of Amazon SQS arose. We have decided to use FUSE ESB(ServiceMix) 4 as the services integration platform for the demo application, and knowing that Apache Camel collaborates very well with ServiceMix, we decided to use it for the routing and transformation needs.

Currently, neither ServiceMix, nor Camel provide a component to access Amazon SQS. Besides this limitation the remaining capabilities provided by these products still made it a great choice for the demo application.

To learn more about the demo application, please visit Tom Purcell's blog. I also recommend attending to the Tom Purcell's presentation titled "Bus In the Cloulds" where he'll be speaking about Could Computing use cases and how Open Source Integration technologies can help resolve issues encountered while integrating with business partners.

Another important part in production applications is management and monitoring. To learn more about how to monitor this application, please visit Rod Biresch's blog.

A Solution
Apache Camel makes it relatively easy to compose your own components, so the next logical step was to try to create one. Having seen Dave Kavanagh's Typica API in action and how easy it makes the communication with SQS, we decided to try and come up with such Camel component using this API as the back-bone.

The Camel Amazon SQS Component
The basic idea is to create a component that could be used in the following fashion:

<camel:route>
<camel:from uri="file:data-in" />
<camel:convertBodyTo type="java.lang.String" />
<camel:to uri="sqs:sqsProducer?accessId=...&secretKey=...&queueName=dataQueue"/>
</camel:route>


<camel:route>
<camel:from
uri="sqs:sqsConsumer?accessId=...&secretKey=...&queueName=dataQueue" />
<camel:to uri="file:data-out" />
</camel:route>

The above example shows a Camel route that sends data found in files placed in the data directory to a SQS queue named "dataQueue". In the to URI "sqs:sqsProducer?accessId=....", the protocol "sqs" will tell Camel what component to configure and attributes will be mapped the properties of either the Endpoint, or the Producer class. The documentation on writing Camel Components can be found here. Let's dive into the design of the Camel SQS component. The main steps for writing a Camel Component are:

• Write a POJO which implements the Component interface. The simplest approach is just to derive from DefaultComponent. In our case, the class SQSComponent was created. This class extends DefaultComponent.

• To support auto-discovery of your component add a file to META-INF/services/org/apache/camel/component/sqs

where 'sqs' is the URI scheme for your component and any related endpoints created on the fly. The sqs file will contain the class attribute with the name of Class that implements the Component:

class=com.chariotsolutions.soalab.camel.component.sqs.SQSComponent

 The Camel SQS Component is composed of the following artifacts:

The SQSComponent Class
This class serves as factory of Endpoints. When the component is being configured by Camel, the method of the createEndpoint is called. This method creates and configures the Endpoint (SQSEndpoint).

The SQSEndpoint Class
This class serves as a factory for Producers, Consumers, and Exchanges.
After the SQSComponent.createEndpoint is execute, an instance of the class SQSEndoint is created.
This class contains the properties used by the SQSConsumer class. These properties are:

private String accessId;
private String secretKey;
private String queueName;

This class extends ScheduledPollEndpoint and is used to create instances of the SQSProducer and SQSConsumer. The use of the ScheduledPollEndpoint is with the intention to use polling facilities it provides to query the SQS Queue at a specified time intervals.

The SQSConsumer Class
This class extends the ScheduledPollConsumer class to indicate that it will be excueted in poll intervals.
The poll intervals are configured in the Component URI using the consumer.initialDelay and consumer.delay attributes like so:

 <from uri="sqs:sqsConsumer?...&consumer.initialDelay=1000&consumer.delay=25000" />

The poll() will be executed based on the configuration specified above.This method uses the Typica API to interact with the Amazon SQS Web services. If a message was found in the SQS Queue, it will e forward to the handleMessage. This method will set the message in a wrapper class called SQSObject, place it in the SQSExchange and send it flow of the route using the synchronous Processor by default. The attribute asyncProcess=true would cause it to use the AsyncProcess.

The SQSProducer Class
This class extends the DefaultProducer class. It processes the message coming from the Camel flow and sends it to the SQS queue, then it sets the assigned message id to the SQSObject, which is later added to the SQSExchange in case other components need it down the road.

The SQSExchange Class
This The base message exchange interface providing access to the request, response and fault instances.

The SQSObject Class
This is convenient POJO that wraps the payload and properties coming from the SQS Queue.

Taking the Camel SQS Component For A Ride
To see the Camel SQS Component in action, I created a small sample application.
The sample application can be found in here. Here are the steps to build and run the application:

1. If you haven't already done so, create an Amazon SQS account using the "Sign Up" button in this page.
2. Extract the sample code to a directory of your choice. For example, /tmp/sqs-sample. You should see 2 directories in there, camel-sqs (camel sqs component) and camel-sqs-route (sample camel route).
3. Navigate to the camel-sqs directory and build the code using: mvn clean install
4. Navigate to the camel-sqs-route directory.
5. Edit the file src/main/resources/aws.properties. Specify aws.accessId and aws.secretKey for your Amazon SQS account.
6. Edit the file src/main/resources/META-INF/spring/camel-context.xml. Find the file endpoints and modify them to point to the directories of your choice. The File-In endpoint is where you would supply the data files to be sent to the SQS queue. The File-Out endpoint is the directory where the files containing the contents of the SQS queue will be placed on.
7. Build the code using: mvn clean install
8. Run the code using: mvn camel:run
9. You should be able to place your sample text files directory pointed by the File-In endpoint. After a few seconds the output files will be placed in directory pointed by File-Out.

Conclusion

Amazon SQS combined with ServiceMix and Camel are a great platform for building your  IaaS (Infrastructure as a Service) projects. Hopefully future versions of Apache Camel will include such component. The Camel CAMEL-1432 JIRA issue has been created for this reason.Happy Cloud Computing!

Recommended Sites
Tom Purcell's Blog
Rod Biresch's Blog
Amazon SQS
FUSE ESB (ServiceMix) 4
Apache Camel
Chariot Solutions