Apache Camel Framework and S3 Integration

In this blog, I am going to demonstrate how to integrate AWS S3 in your Java application using Apache Camel Framework.

Introduction about S3:

AWS S3 provides developers and IT teams with secure, durable, highly-scalable object storage. Amazon S3 is easy to use, with a simple web services interface to store and retrieve any amount of data from anywhere on the web. With Amazon S3, you pay only for the storage you actually use. There is no minimum fee and no setup cost.

In this section I will demonstrate how to upload and download to the AWS S3 bucker with the working code.

Refer my previous “Apache Camel Framework and SQS Integration” blog for the code execution environment setup. 

Upload file to S3 bucket:

The camel component for the S3 activity is aws-s3 and in this example I given bucket name as camelblogs3. All you need is Camel producer template object and File object. Camel S3 component will take care of rest. You don’t need to learn the AWS Java API’s to send files to S3 buckets.

AmazonS3 client = new AmazonS3Client( new DefaultAWSCredentialsProviderChain() ); 

client.setRegion( Region.getRegion(Regions.US_WEST_2));

SimpleRegistry registry = new SimpleRegistry();

registry.put("client", client);

CamelContext camelContext = new DefaultCamelContext(registry);

ProducerTemplate template = camelContext.createProducerTemplate();

File file = new File( "<INSERT_FILE_LOCATION>");

template.sendBodyAndHeader("aws-s3://camelblogs3?amazonS3Client=#client", file , S3Constants.KEY, file.getName());

Download file from S3 bucket:

Don’t remember to pass “deleteAfterRead=false” parameter in the Camel endpoint. 

AmazonS3 client = new AmazonS3Client( new DefaultAWSCredentialsProviderChain() ); 

client.setRegion( Region.getRegion(Regions.US_WEST_2));

SimpleRegistry registry = new SimpleRegistry();

registry.put("client", client);

CamelContext camelContext = new DefaultCamelContext(registry);

ConsumerTemplate consumer = camelContext.createConsumerTemplate();

S3ObjectInputStream s3inputStream = (S3ObjectInputStream)consumer.receiveBody("aws-s3://camelblogs3?amazonS3Client=#client&deleteAfterRead=false&fileName=hello.txt");

String outPutFile = "<INSER_FILE_LOCATION>";

FileWriter writer = new FileWriter(outPutFile);

String thisLine = "";

InputStream fileStream = new BufferedInputStream( s3inputStream);

BufferedReader buffered = new BufferedReader(new InputStreamReader(fileStream, "UTF-8"));

while ( (thisLine = buffered.readLine() ) != null ){

writer.write(thisLine+'\n');

}

writer.close();

Apache Camel Framework and SQS Integration

In this blog, I am going to demonstrate how to integrate AWS SQS in your Java application using Apache Camel Framework.

Little Introduction about Camel Framework:

Apache Camel is an open source Java framework that focuses on making integration easier. It's a complete production-ready framework for people who want to implement their solution to follow the Enterprise integration pattern. Camel will help you to:

1) Consume data from any source/format

2) Process this data

3) Output data to any source/format

In this section I will demonstrate how to send and receive message to the AWS SQS queue with the sample working code.

Follow below instructions to setup code execution environment.

1.  Download the latest Eclipse and install AWS SDK

      a) Open Help → Install New Software….

      b) Enter http://aws.amazon.com/eclipse in the text box labeled “Work with” at the top of the               dialog.

      c) Select “AWS Toolkit for Eclipse” from the list below.

      d) Click “Next.” Eclipse guides you through the remaining installation steps.

2.  Create a Maven project and update the pom.xml for camel library and log4j, 

3.  Create a SQS queue & setup IAM user and assign “AmazonSQSFullAccess” to the IAM user. Then generate the Access and Secret Key.

4. Update the Access Key or Secret Key in the Eclipse AWS Toolkit. For more details

http://docs.aws.amazon.com/AWSToolkitEclipse/latest/GettingStartedGuide/Welcome.html

Sending message to the SQS queue:

First you need a SQSClient object and then register it in the Camel Registry. ProducerTemplate will send message to Camel component. In this example aws-sqs is the camel component and camelblogs is the SQS queue name. Sample working code is below

AmazonSQS client = new AmazonSQSClient( new DefaultAWSCredentialsProviderChain() );  client.setRegion( Region.getRegion(Regions.US_WEST_2)); // Setting the Region, you can set your assigned regions

SimpleRegistry registry = new SimpleRegistry();

registry.put("client", client);       

CamelContext camelContext = new DefaultCamelContext(registry);

ProducerTemplate template = camelContext.createProducerTemplate();

camelContext.start();

String json = "{'message':'Hello World'}";

template.sendBody("aws-sqs://camelblogs?amazonSQSClient=#client", json );

Receive the Message from SQS:

While retrieving the message first you need to define the Camel router with aws-sqs end point and define the processor to process the message. Once you define the Camel router add them into the Camel Context. You can still define the routes using xml configuration. The below example just demonstrate without xml configuration.Sample working code below

AmazonSQS client = new AmazonSQSClient( new DefaultAWSCredentialsProviderChain() ); 

client.setRegion( Region.getRegion(Regions.US_WEST_2)); // Setting the Region, you can set your own selected regions

SimpleRegistry registry = new SimpleRegistry();

registry.put("client", client);

final Processor processor = new Processor(){

public void process(Exchange exchange) throws Exception {

                // TODO Auto-generated method stub

                System.out.println("Received message: " + exchange.getIn().getBody(String.class));

}

};

CamelContext camelContext = new DefaultCamelContext(registry);

camelContext.addRoutes(new RouteBuilder(){

@Override

public void configure() throws Exception {

                System.out.println("CamelContext routers about to add.");

                // TODO Auto-generated method stub

                from ("aws-sqs://camelblogs?amazonSQSClient=#client").process( processor);

}

});

camelContext.start();

System.out.println("CamelContext about to stop.");

Thread.sleep(30000);

camelContext.stop();

Debugging tips:

1) Enable the logger by creating the log4j.properties file in the resource folder.

2) Make sure that SQSClient object picks the Access and SecretKey using log statements

3) Make sure that you have set the AWS region for your sqs queue in the code.