commit-3: Standalone version.

OpenAQSQSListener/src/test/java/com/mcsds/cc/openaq/OpenAQSQSListener/AppTest.java

-package com.mcsds.cc.openaq.OpenAQSQSListener;
-
-import junit.framework.Test;
-import junit.framework.TestCase;
-import junit.framework.TestSuite;
-
-/**
- * Unit test for simple App.
- */
-public class AppTest 
-    extends TestCase
-{
-    /**
-     * Create the test case
-     *
-     * @param testName name of the test case
-     */
-    public AppTest( String testName )
-    {
-        super( testName );
-    }
-
-    /**
-     * @return the suite of tests being tested
-     */
-    public static Test suite()
-    {
-        return new TestSuite( AppTest.class );
-    }
-
-    /**
-     * Rigourous Test :-)
-     */
-    public void testApp()
-    {
-        assertTrue( true );
-    }
-}

OpenAQSQSListener/src/test/java/com/mcsds/cc/openaq/test/SimpleQueueServiceSample.java

+package com.mcsds.cc.openaq.test;
+
+import java.util.List;
+import java.util.Map.Entry;
+
+import com.amazonaws.AmazonClientException;
+import com.amazonaws.AmazonServiceException;
+import com.amazonaws.auth.profile.ProfileCredentialsProvider;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.sqs.AmazonSQS;
+import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
+import com.amazonaws.services.sqs.model.DeleteMessageRequest;
+import com.amazonaws.services.sqs.model.DeleteMessageResult;
+import com.amazonaws.services.sqs.model.Message;
+import com.amazonaws.services.sqs.model.MessageAttributeValue;
+import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
+
+/**
+ * <p>
+ * Sample application to test both the openaq queues.
+ * <ul>
+ * <li>arn:aws:sns:us-east-1:470049585876:NewFetchObject</li>
+ * <li>arn:aws:sns:us-east-1:470049585876:OPENAQ_NEW_MEASUREMENT</li>
+ * </ul>
+ * 
+ * This sample will test end to end. Read from the queue, delete from the queue
+ * after read, Convert json message to entity object etc.
+ * 
+ * </p>
+ * 
+ * @author Kalathil Ajay Menon
+ * 
+ */
+public class SimpleQueueServiceSample {
+
+	public ProfileCredentialsProvider getCredentialsProvider() {
+		ProfileCredentialsProvider credentialsProvider = new ProfileCredentialsProvider();
+		try {
+			credentialsProvider.getCredentials();
+		} catch (Exception e) {
+			throw new AmazonClientException("Cannot load the credentials from the credential profiles file. "
+					+ "Please make sure that your credentials file is at the correct "
+					+ "location (C:\\Users\\kajay\\.aws\\credentials), and is in valid format.", e);
+		}
+		return credentialsProvider;
+	}
+
+	public AmazonSQS createAmazonSqs() {
+		AmazonSQS sqs = AmazonSQSClientBuilder.standard().withCredentials(getCredentialsProvider())
+				.withRegion(Regions.US_EAST_1).build();
+		return sqs;
+	}
+
+	public List<Message> getMessages(String queueName, AmazonSQS sqs) {
+		List<Message> messages = null;
+		try {
+			System.out.println("Receiving messages from :" + queueName + "\n");
+
+			String queueUrl = sqs.getQueueUrl(queueName).getQueueUrl();
+			ReceiveMessageRequest receiveMessageRequest1 = new ReceiveMessageRequest(queueUrl);
+			messages = sqs.receiveMessage(receiveMessageRequest1).getMessages();
+
+		} catch (AmazonServiceException ase) {
+			System.out.println("Caught an AmazonServiceException, which means your request made it "
+					+ "to Amazon SQS, but was rejected with an error response for some reason.");
+			System.out.println("Error Message:    " + ase.getMessage());
+			System.out.println("HTTP Status Code: " + ase.getStatusCode());
+			System.out.println("AWS Error Code:   " + ase.getErrorCode());
+			System.out.println("Error Type:       " + ase.getErrorType());
+			System.out.println("Request ID:       " + ase.getRequestId());
+		} catch (AmazonClientException ace) {
+			System.out.println("Caught an AmazonClientException, which means the client encountered "
+					+ "a serious internal problem while trying to communicate with SQS, such as not "
+					+ "being able to access the network.");
+			System.out.println("Error Message: " + ace.getMessage());
+		}
+		return messages;
+	}
+
+	public void printMessages(List<Message> messages) {
+		if (messages != null && !messages.isEmpty()) {
+			for (Message message : messages) {
+				System.out.println("  Message");
+				System.out.println("    MessageId:     " + message.getMessageId());
+				System.out.println("    ReceiptHandle: " + message.getReceiptHandle());
+				System.out.println("    MD5OfBody:     " + message.getMD5OfBody());
+				System.out.println("    Body:          " + message.getBody());
+
+				for (Entry<String, String> entry : message.getAttributes().entrySet()) {
+					System.out.println("  Attribute");
+					System.out.println("    Name:  " + entry.getKey());
+					System.out.println("    Value: " + entry.getValue());
+				}
+
+				for (Entry<String, MessageAttributeValue> entry : message.getMessageAttributes().entrySet()) {
+					System.out.println(" Message Attribute");
+					System.out.println("    Name:  " + entry.getKey());
+					System.out.println("    Value: " + entry.getValue());
+				}
+			}
+		}
+		System.out.println();
+	}
+
+	public boolean clearMessage(Message message, String queueName, AmazonSQS sqs) {
+		String messageReceiptHandle = message.getReceiptHandle();
+		DeleteMessageResult dmr = sqs.deleteMessage(new DeleteMessageRequest(queueName, messageReceiptHandle));
+		if (dmr != null) {
+			return Boolean.TRUE;
+		} else {
+			return Boolean.FALSE;
+		}
+	}
+
+	public static void main(String[] args) {
+
+		System.out.println("===========================================");
+		System.out.println("Getting Started with Amazon SQS");
+		System.out.println("===========================================\n");
+		boolean clearMsgsAfterRead = false;
+
+		SimpleQueueServiceSample sqsSample = new SimpleQueueServiceSample();
+		AmazonSQS sqs = sqsSample.createAmazonSqs();
+		List<Message> messages = sqsSample.getMessages("jvmcoders-openaq-new-measurements", sqs);
+		sqsSample.printMessages(messages);
+		if (clearMsgsAfterRead && messages != null && !messages.isEmpty()) {
+			for (Message message : messages) {
+				boolean isDeleted = sqsSample.clearMessage(message, "jvmcoders-openaq-new-measurements", sqs);
+				if (isDeleted) {
+					System.out.println("Deleted message : " + message.getMessageId());
+				}
+			}
+		}
+
+		messages = sqsSample.getMessages("jvmcoders-openaq-newfetchobject", sqs);
+		sqsSample.printMessages(messages);
+		if (clearMsgsAfterRead && messages != null && !messages.isEmpty()) {
+			for (Message message : messages) {
+				boolean isDeleted = sqsSample.clearMessage(message, "jvmcoders-openaq-newfetchobject", sqs);
+				if (isDeleted) {
+					System.out.println("Deleted message : " + message.getMessageId());
+				}
+			}
+		}
+
+		System.out.println("===========================================");
+		System.out.println("Completed Amazon SQS read");
+		System.out.println("===========================================\n");
+	}
+}