'Version 1.0.0 of the Amazon SQS Java Messaging Library'

Author: balentr

src/main/java/com/amazonaws/sqsjms/AmazonSQSClientJMSWrapper.java

@@ -35,6 +35,7 @@
 import com.amazonaws.services.sqs.model.DeleteMessageBatchRequest;
 import com.amazonaws.services.sqs.model.DeleteMessageBatchResult;
 import com.amazonaws.services.sqs.model.DeleteMessageRequest;
+import com.amazonaws.services.sqs.model.CreateQueueRequest;
 import com.amazonaws.services.sqs.model.GetQueueUrlRequest;
 import com.amazonaws.services.sqs.model.GetQueueUrlResult;
 import com.amazonaws.services.sqs.model.QueueDoesNotExistException;
@@ -47,14 +48,17 @@
  * This is a JMS Wrapper of AmazonSQSClient. This class changes all
  * AmazonServiceException and AmazonClientException into
  * JMSException/JMSSecurityException.
- * 
- * @author jinsrhee
  */
 public class AmazonSQSClientJMSWrapper {
     private static final Log LOG = LogFactory.getLog(AmazonSQSClientJMSWrapper.class);
 
     private static final Set<String> SECURITY_EXCEPTION_ERROR_CODES;
     static {
+        /**
+         * List of exceptions that can classified as security. These exceptions
+         * are not thrown during connection-set-up rather after the service
+         * calls of the AmazonSQSClient
+         */
         SECURITY_EXCEPTION_ERROR_CODES = new HashSet<String>();
         SECURITY_EXCEPTION_ERROR_CODES.add("MissingClientTokenId");
         SECURITY_EXCEPTION_ERROR_CODES.add("InvalidClientTokenId");
@@ -63,7 +67,13 @@ public class AmazonSQSClientJMSWrapper {
     }
 
     private final AmazonSQS amazonSQSClient;
-
+    
+    /**
+     * @param amazonSQSClient
+     *            The AWS SDK Client for SQS.
+     * @throws JMSException
+     *            if the client is null
+     */
     public AmazonSQSClientJMSWrapper(AmazonSQS amazonSQSClient) throws JMSException {
         if (amazonSQSClient == null) {
             throw new JMSException("Amazon SQS client cannot be null");
@@ -75,11 +85,18 @@ public AmazonSQSClientJMSWrapper(AmazonSQS amazonSQSClient) throws JMSException
      * If one uses any other AWS SDK operations other than explicitly listed
      * here, the exceptions thrown by those operations will not be wrapped as
      * JMSExceptions.
+     * @return amazonSQSClient
      */
     public AmazonSQS getAmazonSQSClient() {
         return amazonSQSClient;
     }
 
+    /**
+     * @param endpoint
+     *            The endpoint (ex: "sqs.us-east-1.amazonaws.com") of the region
+     *            specific AWS endpoint this client will communicate with.
+     * @throws JMSException
+     */
     public void setEndpoint(String endpoint) throws JMSException {
         try {
             amazonSQSClient.setEndpoint(endpoint);
@@ -89,6 +106,13 @@ public void setEndpoint(String endpoint) throws JMSException {
         }
     }
 
+    /**
+     * @param region
+     *            The region this client will communicate with. See
+     *            {@link Region#getRegion(com.amazonaws.regions.Regions)} for
+     *            accessing a given region.
+     * @throws JMSException
+     */
     public void setRegion(Region region) throws JMSException {
         try {
             amazonSQSClient.setRegion(region);
@@ -98,22 +122,41 @@ public void setRegion(Region region) throws JMSException {
         }
     }
 
+    /**
+     * @param deleteMessageRequest
+     *            Container for the necessary parameters to execute the
+     *            deleteMessage service method on AmazonSQS.
+     * @throws JMSException
+     */
     public void deleteMessage(DeleteMessageRequest deleteMessageRequest) throws JMSException {
         try {
             amazonSQSClient.deleteMessage(deleteMessageRequest);
         } catch (AmazonClientException e) {
             throw handleException(e, "deleteMessage");
         }
     }
-
+    
+    /**
+     * @param deleteMessageBatchRequest
+     *            Container for the necessary parameters to execute the
+     *            deleteMessageBatch service method on AmazonSQS. This is the
+     *            batch version of deleteMessage. Max batch size is 10.
+     * @throws JMSException
+     */
     public DeleteMessageBatchResult deleteMessageBatch(DeleteMessageBatchRequest deleteMessageBatchRequest) throws JMSException {
         try {
             return amazonSQSClient.deleteMessageBatch(deleteMessageBatchRequest);
         } catch (AmazonClientException e) {
             throw handleException(e, "deleteMessageBatch");
         }    
     }
-
+    
+    /**
+     * @param sendMessageRequest
+     *            Container for the necessary parameters to execute the
+     *            sendMessage service method on AmazonSQS.
+     * @throws JMSException
+     */
     public SendMessageResult sendMessage(SendMessageRequest sendMessageRequest) throws JMSException {
         try {
             return amazonSQSClient.sendMessage(sendMessageRequest);
@@ -141,11 +184,25 @@ public boolean queueExists(String queueName) throws JMSException {
             throw handleException(e, "getQueueUrl");  
         } 
     }
-
+    
+    /**
+     * @param queueName
+     * @return The response from the GetQueueUrl service method, as returned by
+     *         AmazonSQS, which will include queue`s URL
+     * @throws JMSException
+     */
     public GetQueueUrlResult getQueueUrl(String queueName) throws JMSException {
         return getQueueUrl(new GetQueueUrlRequest(queueName));
     }
-
+     
+    /**
+     * @param getQueueUrlRequest
+     *            Container for the necessary parameters to execute the
+     *            getQueueUrl service method on AmazonSQS.
+     * @return The response from the GetQueueUrl service method, as returned by
+     *         AmazonSQS, which will include queue`s URL
+     * @throws JMSException
+     */
     public GetQueueUrlResult getQueueUrl(GetQueueUrlRequest getQueueUrlRequest) throws JMSException {
         try {
             return amazonSQSClient.getQueueUrl(getQueueUrlRequest);
@@ -154,30 +211,80 @@ public GetQueueUrlResult getQueueUrl(GetQueueUrlRequest getQueueUrlRequest) thro
         }    
     }
 
+    /**
+     * This function creates the queue with the default queue attributes. 
+     * 
+     * @param queueName
+     * @return The response from the createQueue service method, as returned by
+     *         AmazonSQS. This call creates a new queue, or returns the URL of
+     *         an existing one.
+     * @throws JMSException
+     */
     public CreateQueueResult createQueue(String queueName) throws JMSException {
         try {
             return amazonSQSClient.createQueue(queueName);
         } catch (AmazonClientException e) {
             throw handleException(e, "createQueue");   
         }    
     }
-
-    public void changeMessageVisibility(ChangeMessageVisibilityRequest changeMessageVisibilityRequest) throws JMSException {
+    
+    /**
+     * @param createQueueRequest
+     *            Container for the necessary parameters to execute the
+     *            createQueue service method on AmazonSQS.
+     * @return The response from the createQueue service method, as returned by
+     *         AmazonSQS. This call creates a new queue, or returns the URL of
+     *         an existing one.
+     * @throws JMSException
+     */
+    public CreateQueueResult createQueue(CreateQueueRequest createQueueRequest) throws JMSException {
         try {
-            amazonSQSClient.changeMessageVisibility(changeMessageVisibilityRequest);
+            return amazonSQSClient.createQueue(createQueueRequest);
         } catch (AmazonClientException e) {
-            throw handleException(e, "changeMessageVisibility");
+            throw handleException(e, "createQueue");   
         }    
     }
 
+    /**
+     * @param receiveMessageRequest
+     *            Container for the necessary parameters to execute the
+     *            receiveMessage service method on AmazonSQS.
+     * @return The response from the ReceiveMessage service method, as returned
+     *         by AmazonSQS.
+     * @throws JMSException
+     */    
     public ReceiveMessageResult receiveMessage(ReceiveMessageRequest receiveMessageRequest) throws JMSException {
         try {
             return amazonSQSClient.receiveMessage(receiveMessageRequest);
         } catch (AmazonClientException e) {
             throw handleException(e, "receiveMessage");
         }
     }
-
+    
+    /**
+     * @param changeMessageVisibilityRequest
+     *            Container for the necessary parameters to execute the
+     *            changeMessageVisibility service method on AmazonSQS.
+     * @return The response from the changeMessageVisibility service method, as
+     *         returned by AmazonSQS.
+     * @throws JMSException
+     */
+    public void changeMessageVisibility(ChangeMessageVisibilityRequest changeMessageVisibilityRequest) throws JMSException {
+        try {
+            amazonSQSClient.changeMessageVisibility(changeMessageVisibilityRequest);
+        } catch (AmazonClientException e) {
+            throw handleException(e, "changeMessageVisibility");
+        }    
+    }
+    
+    /**
+     * @param changeMessageVisibilityBatchRequest
+     *            Container for the necessary parameters to execute the
+     *            changeMessageVisibilityBatch service method on AmazonSQS.
+     * @return The response from the changeMessageVisibilityBatch service
+     *         method, as returned by AmazonSQS.
+     * @throws JMSException
+     */
     public ChangeMessageVisibilityBatchResult changeMessageVisibilityBatch(ChangeMessageVisibilityBatchRequest changeMessageVisibilityBatchRequest)
             throws JMSException {
         try {
@@ -188,19 +295,20 @@ public ChangeMessageVisibilityBatchResult changeMessageVisibilityBatch(ChangeMes
     }
 
     /**
-     * Create generic error message for AmazonServiceException.
-     * Message include ActionCall, RequestId, HTTPStatusCode, AmazonErrorCode.
+     * Create generic error message for AmazonServiceException. Message include
+     * ActionCall, RequestId, HTTPStatusCode, AmazonErrorCode.
      */
     private String logAndGetAmazonServiceException(AmazonServiceException ase, String action) {
-        String errorMessage = "AmazonServiceException: " + action + ". RequestId: " + ase.getRequestId() + "\nHTTPStatusCode: "
-                + ase.getStatusCode() + " AmazonErrorCode: " + ase.getErrorCode();
+        String errorMessage = "AmazonServiceException: " + action + ". RequestId: " + ase.getRequestId() +
+                              "\nHTTPStatusCode: " + ase.getStatusCode() + " AmazonErrorCode: " +
+                              ase.getErrorCode();
         LOG.error(errorMessage, ase);
         return errorMessage;
     }
 
     /**
-     * Create generic error message for AmazonClientException.
-     * Message include ActionCall.
+     * Create generic error message for AmazonClientException. Message include
+     * ActionCall.
      */
     private String logAndGetAmazonClientException(AmazonClientException ace, String action) {
         String errorMessage = "AmazonClientException: " + action + ".";

src/main/java/com/amazonaws/sqsjms/BulkSQSOperation.java

@@ -24,14 +24,19 @@
 
 import com.amazonaws.sqsjms.acknowledge.SQSMessageIdentifier;
 
+/**
+ * This is used by different acknowledgers that requires partitioning of the
+ * list, and execute actions on the partitions
+ */
 public abstract class BulkSQSOperation {
+    
     /**
-     * Bulk action on List of messageIdentifier. Up to the indexOfMessage.
+     * Bulk action on list of message identifiers up to the indexOfMessage.
      */
     public void bulkAction(List<SQSMessageIdentifier> messageIdentifierList, int indexOfMessage)
             throws JMSException {
         Map<String, List<String>> receiptHandleWithSameQueueUrl = new HashMap<String, List<String>>();
-
+        
         // Add all messages up to and including requested message into Map.
         // Map contains key as queueUrl and value as list receiptHandles from
         // that queueUrl.

src/main/java/com/amazonaws/sqsjms/PrefetchManager.java

@@ -15,12 +15,14 @@
 package com.amazonaws.sqsjms;
 
 /**
- * This interface is helper to notify when the prefetch should be stopped.
+ * This interface is helper to notify when the prefetch should be resumed.
  */
 public interface PrefetchManager {
 
     /**
-     * Notify the prefetchThread that the message is dispatched from messageQueue when user calls for receive.
+     * Notify the prefetchThread that the message is dispatched from
+     * messageQueue when user calls for receive or message listener onMessage is
+     * called.
      */
     public void messageDispatched();
-}
+}

src/main/java/com/amazonaws/sqsjms/SQSBytesMessage.java

@@ -20,6 +20,7 @@
 import java.io.DataOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
+import java.util.Arrays;
 
 import javax.jms.BytesMessage;
 import javax.jms.JMSException;
@@ -53,8 +54,8 @@ public class SQSBytesMessage extends SQSMessage implements BytesMessage {
     SQSBytesMessage(Acknowledger acknowledger, String queueUrl, Message sqsMessage) throws JMSException {
         super(acknowledger, queueUrl, sqsMessage);
         try {
-            bytes = Base64.decode(sqsMessage.getBody());
-            dataOut.write(bytes);
+        	/** Bytes is set by the reset() */
+            dataOut.write(Base64.decode(sqsMessage.getBody()));
             reset();
         } catch (IOException e) {
             LOG.error("IOException: Message cannot be written", e);
@@ -77,7 +78,7 @@ public long getBodyLength() throws JMSException {
         checkCanRead();
         return bytes.length;
     }
-
+    
     @Override
     public boolean readBoolean() throws JMSException {
         checkCanRead();
@@ -218,7 +219,8 @@ public int readBytes(byte[] value) throws JMSException {
     @Override
     public int readBytes(byte[] value, int length) throws JMSException {
         if (length < 0) {
-            throw new IndexOutOfBoundsException("Length bytes to read can't be smaller than 0 but was " + length);
+            throw new IndexOutOfBoundsException("Length bytes to read can't be smaller than 0 but was " +
+                                                length);
         }
         checkCanRead();
         try {
@@ -385,24 +387,21 @@ public void writeObject(Object value) throws JMSException {
         } else if (value instanceof byte[]) {
             writeBytes((byte[]) value);
         } else {
-            throw new MessageFormatException("Cannot write non-primitive type:" + value.getClass());
+            throw new MessageFormatException("Cannot write non-primitive type: " + value.getClass());
         }
     }
 
     /**
-     * Puts the message body in read-only mode and repositions the stream of bytes to the beginning. 
+     * Puts the message body in read-only mode and repositions the stream of
+     * bytes to the beginning.
      */
     @Override
     public void reset() throws JMSException {
-        try {
-            if (dataOut != null) {
-                dataOut.flush();
-                bytes = bytesOut.toByteArray();
-                dataOut = null;
-                bytesOut = null;
-            }
-        } catch (IOException e) {
-            throw convertExceptionToJMSException(e);
+
+        if (dataOut != null) {
+            bytes = bytesOut.toByteArray();
+            dataOut = null;
+            bytesOut = null;
         }
         dataIn = new DataInputStream(new ByteArrayInputStream(bytes));
     }
@@ -423,7 +422,20 @@ public void clearBody() throws JMSException {
         dataOut = new DataOutputStream(bytesOut);
         setBodyWritePermissions(true);
     }
-
+    
+	/**
+	 * Reads the body of message, which can be either the body returned from the
+	 * the receives message as bytes or the bytes put in bytesOut if it is a
+	 * sent message.
+	 */
+	byte[] getBodyAsBytes() throws JMSException {
+		if (bytes != null) {
+			return Arrays.copyOf(bytes, bytes.length);
+		} else {
+			return bytesOut.toByteArray();
+		}
+	}
+    
     private void checkCanRead() throws JMSException {        
         if (bytes == null) {
             throw new MessageNotReadableException("Message is not readable");