public class AmazonSQSMessagingClientWrapper {
    private static final Log LOG = LogFactory.getLog(AmazonSQSMessagingClientWrapper.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 <code>AmazonSQSClient</code>.
         */
        SECURITY_EXCEPTION_ERROR_CODES = new HashSet<String>();
        SECURITY_EXCEPTION_ERROR_CODES.add("MissingClientTokenId");
        SECURITY_EXCEPTION_ERROR_CODES.add("InvalidClientTokenId");
        SECURITY_EXCEPTION_ERROR_CODES.add("MissingAuthenticationToken");
        SECURITY_EXCEPTION_ERROR_CODES.add("AccessDenied");
    }

    private final AmazonSQS amazonSQSClient;
    private final AWSCredentialsProvider credentialsProvider;
    
    /**
     * @param amazonSQSClient
     *            The AWS SDK Client for SQS.
     * @throws JMSException
     *            if the client is null
     */
    public AmazonSQSMessagingClientWrapper(AmazonSQS amazonSQSClient) throws JMSException {
        this(amazonSQSClient, null);
    }
    
    /**
     * @param amazonSQSClient
     *            The AWS SDK Client for SQS.
     * @throws JMSException
     *            if the client is null
     */
    public AmazonSQSMessagingClientWrapper(AmazonSQS amazonSQSClient, AWSCredentialsProvider credentialsProvider) throws JMSException {
        if (amazonSQSClient == null) {
            throw new JMSException("Amazon SQS client cannot be null");
        }
        this.amazonSQSClient = amazonSQSClient;
        this.credentialsProvider = credentialsProvider;
    }
    
    /**
     * If one uses any other AWS SDK operations other than explicitly listed
     * here, the exceptions thrown by those operations will not be wrapped as
     * <code>JMSException</code>.
     * @return amazonSQSClient
     */
    public AmazonSQS getAmazonSQSClient() {
        return amazonSQSClient;
    }

    /**
     * Sets SQS endpoint and wraps IllegalArgumentException. 
     * Deprecated. Instead of manipulating settings of existing AmazonSQS client, provide correct configuration when creating it through SQSConnectionFactory constructors.
     * 
     * @param endpoint
     *            The endpoint (ex: "sqs.us-east-1.amazonaws.com") of the region
     *            specific AWS endpoint this client will communicate with.
     * @throws JMSException
     */
    @Deprecated
    public void setEndpoint(String endpoint) throws JMSException {
        try {
            amazonSQSClient.setEndpoint(endpoint);
        } catch (IllegalArgumentException ase) {
            JMSException jmsException = new JMSException(ase.getMessage());
            throw (JMSException) jmsException.initCause(ase);
        }
    }
    
    /**
     * Sets SQS region and wraps <code>IllegalArgumentException</code>. 
     * Deprecated. Instead of manipulating settings of existing AmazonSQS client, provide correct configuration when creating it through SQSConnectionFactory constructors.
     * 
     * @param region
     *            The region this client will communicate with. See
     *            {@link Region#getRegion(com.amazonaws.regions.Regions)} for
     *            accessing a given region.
     * @throws JMSException
     */
    @Deprecated
    public void setRegion(Region region) throws JMSException {
        try {
            amazonSQSClient.setRegion(region);
        } catch (IllegalArgumentException ase) {
            JMSException jmsException = new JMSException(ase.getMessage());
            throw (JMSException) jmsException.initCause(ase);
        }
    }
    
    /**
     * Calls <code>deleteMessage</code> and wraps <code>AmazonClientException</code>. This is used to
     * acknowledge single messages, so that they can be deleted from SQS queue.
     * 
     * @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 {
            prepareRequest(deleteMessageRequest);
            amazonSQSClient.deleteMessage(deleteMessageRequest);
        } catch (AmazonClientException e) {
            throw handleException(e, "deleteMessage");
        }
    }
    
    /**
     * Calls <code>deleteMessageBatch</code> and wraps
     * <code>AmazonClientException</code>. This is used to acknowledge multiple
     * messages on client_acknowledge mode, so that they can be deleted from SQS
     * queue.
     * 
     * @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.
     * @return The response from the deleteMessageBatch service method, as
     *         returned by AmazonSQS
     * @throws JMSException
     */
    public DeleteMessageBatchResult deleteMessageBatch(DeleteMessageBatchRequest deleteMessageBatchRequest) throws JMSException {
        try {
            prepareRequest(deleteMessageBatchRequest);
            return amazonSQSClient.deleteMessageBatch(deleteMessageBatchRequest);
        } catch (AmazonClientException e) {
            throw handleException(e, "deleteMessageBatch");
        }    
    }
    
    /**
     * Calls <code>sendMessage</code> and wraps
     * <code>AmazonClientException</code>.
     * 
     * @param sendMessageRequest
     *            Container for the necessary parameters to execute the
     *            sendMessage service method on AmazonSQS.
     * @return The response from the sendMessage service method, as returned by
     *         AmazonSQS
     * @throws JMSException
     */
    public SendMessageResult sendMessage(SendMessageRequest sendMessageRequest) throws JMSException {
        try {
            prepareRequest(sendMessageRequest);
            return amazonSQSClient.sendMessage(sendMessageRequest);
        } catch (AmazonClientException e) {
            throw handleException(e, "sendMessage");
        }    
    }
    
    /**
     * Check if the requested queue exists. This function calls
     * <code>GetQueueUrl</code> for the given queue name, returning true on
     * success, false if it gets <code>QueueDoesNotExistException</code>.
     * 
     * @param queueName
     *            the queue to check
     * @return true if the queue exists, false if it doesn't.
     * @throws JMSException
     */
    public boolean queueExists(String queueName) throws JMSException {
        try {
            amazonSQSClient.getQueueUrl(prepareRequest(new GetQueueUrlRequest(queueName)));
            return true;
        } catch (QueueDoesNotExistException e) {
            return false;
        } catch (AmazonClientException e) {
            throw handleException(e, "getQueueUrl");  
        } 
    }
    
    /**
     * Check if the requested queue exists. This function calls
     * <code>GetQueueUrl</code> for the given queue name with the given owner
     * accountId, returning true on success, false if it gets
     * <code>QueueDoesNotExistException</code>.
     * 
     * @param queueName
     *            the queue to check
     * @param queueOwnerAccountId
     *            The AWS accountId of the account that created the queue
     * @return true if the queue exists, false if it doesn't.
     * @throws JMSException
     */
    public boolean queueExists(String queueName, String queueOwnerAccountId) throws JMSException {
        try {
            GetQueueUrlRequest getQueueUrlRequest = new GetQueueUrlRequest(queueName);
            getQueueUrlRequest.setQueueOwnerAWSAccountId(queueOwnerAccountId);
            prepareRequest(getQueueUrlRequest);
            amazonSQSClient.getQueueUrl(getQueueUrlRequest);
            return true;
        } catch (QueueDoesNotExistException e) {
            return false;
        } catch (AmazonClientException e) {
            throw handleException(e, "getQueueUrl");
        }
    }
    
    /**
     * Gets the queueUrl of a queue given a queue name.
     * 
     * @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));
    }
    
    /**
     * Gets the queueUrl of a queue given a queue name owned by the provided accountId.
     * 
     * @param queueName
     * @param queueOwnerAccountId The AWS accountId of the account that created the queue
     * @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, String queueOwnerAccountId) throws JMSException {
        return getQueueUrl(new GetQueueUrlRequest(queueName).withQueueOwnerAWSAccountId(queueOwnerAccountId));
    }
     
    /**
     * Calls <code>getQueueUrl</code> and wraps <code>AmazonClientException</code>
     * 
     * @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 {
            prepareRequest(getQueueUrlRequest);
            return amazonSQSClient.getQueueUrl(getQueueUrlRequest);
        } catch (AmazonClientException e) {
            throw handleException(e, "getQueueUrl");  
        }    
    }
    
    /**
     * Calls <code>createQueue</code> to create the queue with the default queue attributes,
     * and wraps <code>AmazonClientException</code>
     * 
     * @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 {
        return createQueue(new CreateQueueRequest(queueName));
    }
    
    /**
     * Calls <code>createQueue</code> to create the queue with the provided queue attributes
     * if any, and wraps <code>AmazonClientException</code>
     * 
     * @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 {
            prepareRequest(createQueueRequest);
            return amazonSQSClient.createQueue(createQueueRequest);
        } catch (AmazonClientException e) {
            throw handleException(e, "createQueue");   
        }    
    }
    
    /**
     * Calls <code>receiveMessage</code> and wraps <code>AmazonClientException</code>. Used by
     * {@link SQSMessageConsumerPrefetch} to receive up to minimum of
     * (<code>numberOfMessagesToPrefetch</code>,10) messages from SQS queue into consumer
     * prefetch buffers.
     * 
     * @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 {
            prepareRequest(receiveMessageRequest);
            return amazonSQSClient.receiveMessage(receiveMessageRequest);
        } catch (AmazonClientException e) {
            throw handleException(e, "receiveMessage");
        }
    }
    
    /**
     * Calls <code>changeMessageVisibility</code> and wraps <code>AmazonClientException</code>. This is 
     * used to for negative acknowledge of a single message, so that messages can be received again without any delay.
     * 
     * @param changeMessageVisibilityRequest
     *            Container for the necessary parameters to execute the
     *            changeMessageVisibility service method on AmazonSQS.
     * @throws JMSException
     */
    public void changeMessageVisibility(ChangeMessageVisibilityRequest changeMessageVisibilityRequest) throws JMSException {
        try {
            prepareRequest(changeMessageVisibilityRequest);
            amazonSQSClient.changeMessageVisibility(changeMessageVisibilityRequest);
        } catch (AmazonClientException e) {
            throw handleException(e, "changeMessageVisibility");
        }    
    }
    
    /**
     * Calls <code>changeMessageVisibilityBatch</code> and wraps <code>AmazonClientException</code>. This is
     * used to for negative acknowledge of messages in batch, so that messages
     * can be received again without any delay.
     * 
     * @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 {
            prepareRequest(changeMessageVisibilityBatchRequest);
            return amazonSQSClient.changeMessageVisibilityBatch(changeMessageVisibilityBatchRequest);
        } catch (AmazonClientException e) {
            throw handleException(e, "changeMessageVisibilityBatch");
        }
    }

    /**
     * Create generic error message for <code>AmazonServiceException</code>. Message include
     * Action, RequestId, HTTPStatusCode, and AmazonErrorCode.
     */
    private String logAndGetAmazonServiceException(AmazonServiceException ase, String action) {
        String errorMessage = "AmazonServiceException: " + action + ". RequestId: " + ase.getRequestId() +
                              "\nHTTPStatusCode: " + ase.getStatusCode() + " AmazonErrorCode: " +
                              ase.getErrorCode();
        LOG.error(errorMessage, ase);
        return errorMessage;
    }

    /**
     * Create generic error message for <code>AmazonClientException</code>. Message include
     * Action.
     */
    private String logAndGetAmazonClientException(AmazonClientException ace, String action) {
        String errorMessage = "AmazonClientException: " + action + ".";
        LOG.error(errorMessage, ace);
        return errorMessage;
    }
    
    private JMSException handleException(AmazonClientException e, String operationName) throws JMSException {
        JMSException jmsException;
        if (e instanceof AmazonServiceException) {
            AmazonServiceException se = ( AmazonServiceException ) e;
            
            if (e instanceof QueueDoesNotExistException) {
                jmsException = new InvalidDestinationException(
                        logAndGetAmazonServiceException(se, operationName), se.getErrorCode());
            } else if (isJMSSecurityException(se)) {
                jmsException = new JMSSecurityException(
                        logAndGetAmazonServiceException(se, operationName), se.getErrorCode());
            } else {
                jmsException = new JMSException(
                        logAndGetAmazonServiceException(se, operationName), se.getErrorCode());
            }

        } else {
            jmsException = new JMSException(logAndGetAmazonClientException(e, operationName));
        }
        jmsException.initCause(e);
        return jmsException;
    }
    
    private boolean isJMSSecurityException(AmazonServiceException e) {
        return SECURITY_EXCEPTION_ERROR_CODES.contains(e.getErrorCode()) ;
    }
    
    private <T extends AmazonWebServiceRequest> T prepareRequest(T request) {
        request.getRequestClientOptions().appendUserAgent(SQSMessagingClientConstants.APPENDED_USER_AGENT_HEADER_VERSION);
        if (credentialsProvider != null) {
            request.setRequestCredentialsProvider(credentialsProvider);
        }
        return request;
    }
    
} public class SQSConnection implements Connection, QueueConnection {
    private static final Log LOG = LogFactory.getLog(SQSConnection.class);
    
    /** For now this doesn't do anything. */
    private ExceptionListener exceptionListener;
    /** For now this doesn't do anything. */
    private String clientID;


    /** Used for interactions with connection state. */
    private final Object stateLock = new Object();
    
    private final AmazonSQSMessagingClientWrapper amazonSQSClient;

    /**
     * Configures the amount of messages that can be prefetched by a consumer. A
     * single consumer cannot prefetch more than 10 messages in a single call to SQS,
     * but it will make multiple calls as necessary.
     */
    private final int numberOfMessagesToPrefetch;
    private volatile boolean closed = false;
    private volatile boolean closing = false;

    /** Used to determine if the connection is stopped or not. */
    private volatile boolean running = false;
    
    /**
     * Used to determine if any other action was taken on the
     * connection, that might prevent setting the clientId
     */
    private volatile boolean actionOnConnectionTaken = false;

    private final Set<Session> sessions = Collections.newSetFromMap(new ConcurrentHashMap<Session, Boolean>());

    SQSConnection(AmazonSQSMessagingClientWrapper amazonSQSClientJMSWrapper, int numberOfMessagesToPrefetch) {
        amazonSQSClient = amazonSQSClientJMSWrapper;
        this.numberOfMessagesToPrefetch = numberOfMessagesToPrefetch;

    }
    
    /**
     * Get the AmazonSQSClient used by this connection. This can be used to do administrative operations
     * that aren't included in the JMS specification, e.g. creating new queues.
     * 
     * @return the AmazonSQSClient used by this connection
     */
    public AmazonSQS getAmazonSQSClient() {
        return amazonSQSClient.getAmazonSQSClient();
    }

    /**
     * Get a wrapped version of the AmazonSQSClient used by this connection. The wrapper transforms 
     * all exceptions from the client into JMSExceptions so that it can more easily be used
     * by existing code that already expects JMSExceptions. This client can be used to do 
     * administrative operations that aren't included in the JMS specification, e.g. creating new queues.
     * 
     * @return  wrapped version of the AmazonSQSClient used by this connection
     */
    public AmazonSQSMessagingClientWrapper getWrappedAmazonSQSClient() {
        return amazonSQSClient;        
    }
    
    int getNumberOfMessagesToPrefetch() {
        return numberOfMessagesToPrefetch;
    }
    
    /**
     * Creates a <code>QueueSession</code>
     * 
     * @param transacted
     *            Only false is supported.
     * @param acknowledgeMode
     *            Legal values are <code>Session.AUTO_ACKNOWLEDGE</code>,
     *            <code>Session.CLIENT_ACKNOWLEDGE</code>,
     *            <code>Session.DUPS_OK_ACKNOWLEDGE</code>, and
     *            <code>SQSSession.UNORDERED_ACKNOWLEDGE</code>
     * @return a new queue session.
     * @throws JMSException
     *             If the QueueConnection object fails to create a session due
     *             to some internal error or lack of support for the specific
     *             transaction and acknowledge mode.
     */
    @Override
    public QueueSession createQueueSession(boolean transacted, int acknowledgeMode) throws JMSException {
        return (QueueSession) createSession(transacted, acknowledgeMode);
    }
    
    /**
     * Creates a <code>Session</code>
     * 
     * @param transacted
     *            Only false is supported.
     * @param acknowledgeMode
     *            Legal values are <code>Session.AUTO_ACKNOWLEDGE</code>,
     *            <code>Session.CLIENT_ACKNOWLEDGE</code>,
     *            <code>Session.DUPS_OK_ACKNOWLEDGE</code>, and
     *            <code>SQSSession.UNORDERED_ACKNOWLEDGE</code>
     * @return a new session.
     * @throws JMSException
     *             If the QueueConnection object fails to create a session due
     *             to some internal error or lack of support for the specific
     *             transaction and acknowledge mode.
     */
    @Override
    public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException {
        checkClosed();
        actionOnConnectionTaken = true;
        if (transacted || acknowledgeMode == Session.SESSION_TRANSACTED)
            throw new JMSException("SQSSession does not support transacted");

        SQSSession sqsSession;
        if (acknowledgeMode == Session.AUTO_ACKNOWLEDGE) {
            sqsSession = new SQSSession(this, AcknowledgeMode.ACK_AUTO.withOriginalAcknowledgeMode(acknowledgeMode));
        } else if (acknowledgeMode == Session.CLIENT_ACKNOWLEDGE || acknowledgeMode == Session.DUPS_OK_ACKNOWLEDGE) {
            sqsSession = new SQSSession(this, AcknowledgeMode.ACK_RANGE.withOriginalAcknowledgeMode(acknowledgeMode));
        } else if (acknowledgeMode == SQSSession.UNORDERED_ACKNOWLEDGE) {
            sqsSession = new SQSSession(this, AcknowledgeMode.ACK_UNORDERED.withOriginalAcknowledgeMode(acknowledgeMode));
        } else {
            LOG.error("Unrecognized acknowledgeMode. Cannot create Session.");
            throw new JMSException("Unrecognized acknowledgeMode. Cannot create Session.");
        }
        synchronized (stateLock) { 
            checkClosing();
            sessions.add(sqsSession);

            /**
             * Any new sessions created on a started connection should be
             * started on creation
             */
            if (running) {
                sqsSession.start();
            }
        }
               
        return sqsSession;
    }

    @Override
    public ExceptionListener getExceptionListener() throws JMSException {
        checkClosing();
        return exceptionListener;
    }

    @Override
    public void setExceptionListener(ExceptionListener listener) throws JMSException {
        checkClosing();
        actionOnConnectionTaken = true;
        this.exceptionListener = listener;
    }
    
    /**
     * Checks if the connection close is in-progress or already completed.
     * 
     * @throws IllegalStateException
     *             If the connection close is in-progress or already completed.
     */
    public void checkClosing() throws IllegalStateException {
        if (closing) {
            throw new IllegalStateException("Connection is closed or closing");
        }
    }

    /**
     * Checks if the connection close is already completed.
     * 
     * @throws IllegalStateException
     *             If the connection close is already completed.
     */
    public void checkClosed() throws IllegalStateException {
        if (closed) {
            throw new IllegalStateException("Connection is closed");
        }
    }
    
    /**
     * Starts a connection's delivery of incoming messages. A call to
     * <code>start</code> on a connection that has already been started is
     * ignored.
     * <P>
     * This will not return until all the sessions start internally.
     * 
     * @throws JMSException
     *             On internal error
     */
    @Override
    public void start() throws JMSException {
        checkClosed();
        actionOnConnectionTaken = true;

        if (running) {
            return;
        }
        synchronized (stateLock) {
            checkClosing();
            if (!running) {
                try {
                    for (Session session : sessions) {
                        SQSSession sqsSession = (SQSSession) session;
                        sqsSession.start();
                    }
                } finally {
                    running = true;
                }
            }
        }
    }
    
    /**
     * Stops a connection's delivery of incoming messages. A call to
     * <code>stop</code> on a connection that has already been stopped is
     * ignored.
     * <P>
     * This will not return until all the sessions stop internally, which blocks
     * until receives and/or message listeners in progress have completed. While
     * these message listeners are completing, they must have the full services
     * of the connection available to them.
     * <P>
     * A call to stop must not return until delivery of messages has paused.
     * This means that a client can rely on the fact that none of its message
     * listeners will be called and that all threads of control waiting for
     * receive calls to return will not return with a message until the
     * connection is restarted. The receive timers for a stopped connection
     * continue to advance, so receives may time out while the connection is
     * stopped.
     * <P>
     * A message listener must not attempt to stop its own connection; otherwise
     * throws a IllegalStateException.
     * 
     * @throws IllegalStateException
     *             If called by a message listener on its own
     *             <code>Connection</code>.
     * @throws JMSException
     *             On internal error or called if close is in progress.
     */
    @Override
    public void stop() throws JMSException {
        checkClosed();
                
        if (!running) {
            return;
        }
        actionOnConnectionTaken = true;
        
        if (SQSSession.SESSION_THREAD_FACTORY.wasThreadCreatedWithThisThreadGroup(Thread.currentThread())) {
            throw new IllegalStateException(
                    "MessageListener must not attempt to stop its own Connection to prevent potential deadlock issues");
        }

        synchronized (stateLock) {
            checkClosing();
            if (running) {
                try {
                    for (Session session : sessions) {
                        SQSSession sqsSession = (SQSSession) session;
                        /**
                         * Session stop call blocks until receives and/or
                         * message listeners in progress have completed.
                         */
                        sqsSession.stop();
                    }
                } finally {
                    running = false;
                }

            }
        }
    }
    
    /**
     * Closes the connection.
     * <P>
     * This will not return until all the sessions close internally, which
     * blocks until receives and/or message listeners in progress have
     * completed.
     * <P>
     * The receives may return with a message or with null, depending on whether
     * there was a message available at the time of the close. If one or more of
     * the connection's sessions' message listeners is processing a message at
     * the time when connection close is invoked, all the facilities of the
     * connection and its sessions must remain available to those listeners
     * until they return control to the JMS provider.
     * <P>
     * A message listener must not attempt to close its own connection;
     * otherwise throws a IllegalStateException.
     * 
     * @throws IllegalStateException
     *             If called by a message listener on its own
     *             <code>Connection</code>.
     * @throws JMSException
     *             On internal error.
     */
    @Override
    public void close() throws JMSException {

        if (closed) {
            return;
        }

        /**
         * A message listener must not attempt to close its own connection as
         * this would lead to deadlock.
         */
        if (SQSSession.SESSION_THREAD_FACTORY.wasThreadCreatedWithThisThreadGroup(Thread.currentThread())) {
            throw new IllegalStateException(
                    "MessageListener must not attempt to close its own Connection to prevent potential deadlock issues");
        }

        boolean shouldClose = false;
        synchronized (stateLock) {
            if (!closing) {
                shouldClose = true;
                closing = true;
            }
        }

        if (shouldClose) {
            synchronized (stateLock) {
                try {
                    for (Session session : sessions) {
                        SQSSession sqsSession = (SQSSession) session;
                        sqsSession.close();
                    }
                    sessions.clear();
                } finally {
                    closed = true;
                    stateLock.notifyAll();

                }
            }
        }/** Blocks until closing of the connection completes */
        else {
            synchronized (stateLock) {
                while (!closed) {
                    try {
                        stateLock.wait();
                    } catch (InterruptedException e) {
                        LOG.error("Interrupted while waiting the session to close.", e);
                    }
                }
            }
        }

    } 

    
    /**
     * This is used in Session. When Session is closed it will remove itself
     * from list of Sessions.
     */
    void removeSession(Session session) throws JMSException {
        /**
         * No need to synchronize on stateLock assuming this can be only called
         * by session.close(), on which point connection will not be worried
         * about missing closing this session.
         */
        sessions.remove(session);
    }
    
    /**
     * Gets the client identifier for this connection.
     * 
     * @return client identifier
     * @throws JMSException
     *             If the connection is being closed
     */
    @Override
    public String getClientID() throws JMSException {
        checkClosing();
        return clientID;
    }
    
    /**
     * Sets the client identifier for this connection.
     * <P>
     * Does not verify uniqueness of client ID, so does not detect if another
     * connection is already using the same client ID
     * 
     * @param clientID
     *            The client identifier
     * @throws JMSException
     *             If the connection is being closed
     * @throws InvalidClientIDException
     *             If empty or null client ID is used
     * @throws IllegalStateException
     *             If the client ID is already set or attempted to set after an
     *             action on the connection already took place
     */
    @Override
    public void setClientID(String clientID) throws JMSException {
        checkClosing();
        if (clientID == null || clientID.isEmpty()) {
            throw new InvalidClientIDException("ClientID is empty");
        }
        if (this.clientID != null) {
            throw new IllegalStateException("ClientID is already set");
        }
        if (actionOnConnectionTaken) {
            throw new IllegalStateException(
                    "Client ID cannot be set after any action on the connection is taken");
        }
        this.clientID = clientID;
    }
    
    /**
     * Get the metadata for this connection
     * 
     * @return the connection metadata
     * @throws JMSException
     *             If the connection is being closed
     */
    @Override
    public ConnectionMetaData getMetaData() throws JMSException {
        checkClosing();
        return SQSMessagingClientConstants.CONNECTION_METADATA;
    }

    /** This method is not supported. */
    @Override
    public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector, ServerSessionPool sessionPool,
            int maxMessages) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName, String messageSelector,
            ServerSessionPool sessionPool, int maxMessages) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public ConnectionConsumer createConnectionConsumer(Queue queue, String messageSelector, ServerSessionPool sessionPool, int maxMessages)
            throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /*
     * Unit Test Utility Functions
     */
    void setClosed(boolean closed) {
        this.closed = closed;
    }

    boolean isClosed() {
        return closed;
    }

    void setClosing(boolean closing) {
        this.closing = closing;
    }

    void setRunning(boolean running) {
        this.running = running;
    }

    boolean isRunning() {
        return running;
    }

    void setActionOnConnectionTaken(boolean actionOnConnectionTaken) {
        this.actionOnConnectionTaken = actionOnConnectionTaken;
    }

    boolean isActionOnConnectionTaken() {
        return actionOnConnectionTaken;
    }

    Set<Session> getSessions() {
        return sessions;
    }

    Object getStateLock() {
        return stateLock;
    }
} public class SQSMessageProducer implements MessageProducer, QueueSender {
    private static final Log LOG = LogFactory.getLog(SQSMessageProducer.class);

    private long MAXIMUM_DELIVERY_DELAY_MILLISECONDS = TimeUnit.MILLISECONDS.convert(15, TimeUnit.MINUTES);
    
    private int deliveryDelaySeconds = 0;

    /** This field is not actually used. */
    private long timeToLive;
    /** This field is not actually used. */
    private int defaultPriority;
    /** This field is not actually used. */
    private int deliveryMode;
    /** This field is not actually used. */
    private boolean disableMessageTimestamp;
    /** This field is not actually used. */
    private boolean disableMessageID;

    /**
     * State of MessageProducer.
     * True if MessageProducer is closed.
     */
    final AtomicBoolean closed = new AtomicBoolean(false);

    private final AmazonSQSMessagingClientWrapper amazonSQSClient;
    private final SQSQueueDestination sqsDestination;
    private final SQSSession parentSQSSession;

    SQSMessageProducer(AmazonSQSMessagingClientWrapper amazonSQSClient, SQSSession parentSQSSession,
                       SQSQueueDestination destination) throws JMSException {
        this.sqsDestination = destination;
        this.amazonSQSClient = amazonSQSClient;
        this.parentSQSSession = parentSQSSession;
    }

    void sendInternal(SQSQueueDestination queue, Message rawMessage) throws JMSException {
        checkClosed();
        String sqsMessageBody = null;
        String messageType = null;
        if (!(rawMessage instanceof SQSMessage)) {
            throw new MessageFormatException(
                    "Unrecognized message type. Messages have to be one of: SQSBytesMessage, SQSObjectMessage, or SQSTextMessage");            
        }
        
        SQSMessage message = (SQSMessage)rawMessage;
        message.setJMSDestination(queue);
        if (message instanceof SQSBytesMessage) {
            sqsMessageBody = Base64.encodeAsString(((SQSBytesMessage) message).getBodyAsBytes());
            messageType = SQSMessage.BYTE_MESSAGE_TYPE;
        } else if (message instanceof SQSObjectMessage) {
            sqsMessageBody = ((SQSObjectMessage) message).getMessageBody();
            messageType = SQSMessage.OBJECT_MESSAGE_TYPE;
        } else if (message instanceof SQSTextMessage) {            
            sqsMessageBody = ((SQSTextMessage) message).getText();
            messageType = SQSMessage.TEXT_MESSAGE_TYPE;
        }
        
        if (sqsMessageBody == null || sqsMessageBody.isEmpty()) {
            throw new JMSException("Message body cannot be null or empty");
        }
        Map<String, MessageAttributeValue> messageAttributes = propertyToMessageAttribute((SQSMessage) message);

        /**
         * These will override existing attributes if they exist. Everything that
         * has prefix JMS_ is reserved for JMS Provider, but if the user sets that
         * attribute, it will be overwritten.
         */
        addStringAttribute(messageAttributes, SQSMessage.JMS_SQS_MESSAGE_TYPE, messageType);
        addReplyToQueueReservedAttributes(messageAttributes, message);
        addCorrelationIDToQueueReservedAttributes(messageAttributes, message);

        SendMessageRequest sendMessageRequest = new SendMessageRequest(queue.getQueueUrl(), sqsMessageBody);
        sendMessageRequest.setMessageAttributes(messageAttributes);

        if (deliveryDelaySeconds != 0) {
            sendMessageRequest.setDelaySeconds(deliveryDelaySeconds);
        }

        //for FIFO queues, we have to specify both MessageGroupId, which we obtain from standard property JMSX_GROUP_ID
        //and MessageDeduplicationId, which we obtain from a custom provider specific property JMS_SQS_DEDUPLICATION_ID
        //notice that this code does not validate if the values are actually set by the JMS user
        //this means that failure to provide the required values will fail server side and throw a JMSException
        if (queue.isFifo()) {
            sendMessageRequest.setMessageGroupId(message.getSQSMessageGroupId());
            sendMessageRequest.setMessageDeduplicationId(message.getSQSMessageDeduplicationId());
        }

        SendMessageResult sendMessageResult = amazonSQSClient.sendMessage(sendMessageRequest);
        String messageId = sendMessageResult.getMessageId();
        LOG.info("Message sent to SQS with SQS-assigned messageId: " + messageId);
        /** TODO: Do not support disableMessageID for now. */
        message.setSQSMessageId(messageId);

        // if the message was sent to FIFO queue, the sequence number will be
        // set in the response
        // pass it to JMS user through provider specific JMS property
        if (sendMessageResult.getSequenceNumber() != null) {
            message.setSequenceNumber(sendMessageResult.getSequenceNumber());
        }
    }

    @Override
    public Queue getQueue() throws JMSException {
        return sqsDestination;
    }
    
    /**
     * Sends a message to a queue.
     * 
     * @param queue
     *            the queue destination to send this message to
     * @param message
     *            the message to send
     * @throws InvalidDestinationException
     *             If a client uses this method with a destination other than
     *             SQS queue destination.
     * @throws MessageFormatException
     *             If an invalid message is specified.
     * @throws UnsupportedOperationException
     *             If a client uses this method with a MessageProducer that
     *             specified a destination at creation time.
     * @throws JMSException
     *             If session is closed or internal error.
     */
    @Override
    public void send(Queue queue, Message message) throws JMSException {
        if (!(queue instanceof SQSQueueDestination)) {
            throw new InvalidDestinationException(
                    "Incompatible implementation of Queue. Please use SQSQueueDestination implementation.");
        }
        checkIfDestinationAlreadySet();
        sendInternal((SQSQueueDestination)queue, message);
    }

    /**
     * Not verified on the client side, but SQS Attribute names must be valid
     * letter or digit on the basic multilingual plane in addition to allowing
     * '_', '-' and '.'. No component of an attribute name may be empty, thus an
     * attribute name may neither start nor end in '.'. And it may not contain
     * "..".
     */
    Map<String, MessageAttributeValue> propertyToMessageAttribute(SQSMessage message)
            throws JMSException {
        Map<String, MessageAttributeValue> messageAttributes = new HashMap<String, MessageAttributeValue>();
        Enumeration<String> propertyNames = message.getPropertyNames();

        while (propertyNames.hasMoreElements()) {
            String propertyName = propertyNames.nextElement();

            // This is generated from SQS message attribute "ApproximateReceiveCount"
            if (propertyName.equals(SQSMessagingClientConstants.JMSX_DELIVERY_COUNT)) {
                continue;
            }

            // This property will be used as DeduplicationId argument of SendMessage call
            // On receive it is mapped back to this JMS property
            if (propertyName.equals(SQSMessagingClientConstants.JMS_SQS_DEDUPLICATION_ID)) {
                continue;
            }

            // the JMSXGroupID and JMSXGroupSeq are always stored as message
            // properties, so they are not lost between send and receive
            // even though SQS Classic does not respect those values when returning messages
            // and SQS FIFO has a different understanding of message groups

            JMSMessagePropertyValue propertyObject = message.getJMSMessagePropertyValue(propertyName);
            MessageAttributeValue messageAttributeValue = new MessageAttributeValue();

            messageAttributeValue.setDataType(propertyObject.getType());
            messageAttributeValue.setStringValue(propertyObject.getStringMessageAttributeValue());

            messageAttributes.put(propertyName, messageAttributeValue);
        }
        return messageAttributes;
    }

    /**
     * Adds the message type attribute during send as part of the send message
     * request.
     */
    private void addMessageTypeReservedAttribute(Map<String, MessageAttributeValue> messageAttributes,
                                                 SQSMessage message, String value) throws JMSException {


        addStringAttribute(messageAttributes, SQSMessage.JMS_SQS_MESSAGE_TYPE, value);
    }

    /**
     * Adds the reply-to queue name and url attributes during send as part of the send message
     * request, if necessary
     */
    private void addReplyToQueueReservedAttributes(Map<String, MessageAttributeValue> messageAttributes,
                                                   SQSMessage message) throws JMSException {

        Destination replyTo = message.getJMSReplyTo();
        if (replyTo instanceof SQSQueueDestination) {
            SQSQueueDestination replyToQueue = (SQSQueueDestination)replyTo;
    
            /**
             * This will override the existing attributes if exists. Everything that
             * has prefix JMS_ is reserved for JMS Provider, but if the user sets that
             * attribute, it will be overwritten.
             */
            addStringAttribute(messageAttributes, SQSMessage.JMS_SQS_REPLY_TO_QUEUE_NAME, replyToQueue.getQueueName());
            addStringAttribute(messageAttributes, SQSMessage.JMS_SQS_REPLY_TO_QUEUE_URL, replyToQueue.getQueueUrl());
        }
    }

    /**
     * Adds the correlation ID attribute during send as part of the send message
     * request, if necessary
     */
    private void addCorrelationIDToQueueReservedAttributes(Map<String, MessageAttributeValue> messageAttributes,
                                                   SQSMessage message) throws JMSException {

        String correlationID = message.getJMSCorrelationID();
        if (correlationID != null) {
            addStringAttribute(messageAttributes, SQSMessage.JMS_SQS_CORRELATION_ID, correlationID);
        }
    }

    /**
     * Convenience method for adding a single string attribute.
     */
    private void addStringAttribute(Map<String, MessageAttributeValue> messageAttributes,
                                    String key, String value) {
        MessageAttributeValue messageAttributeValue = new MessageAttributeValue();
        messageAttributeValue.setDataType(SQSMessagingClientConstants.STRING);
        messageAttributeValue.setStringValue(value);
        messageAttributes.put(key, messageAttributeValue);
    }
    
    /**
     * Sends a message to a queue.
     * <P>
     * Send does not support deliveryMode, priority, and timeToLive. It will
     * ignore anything in deliveryMode, priority, and timeToLive.
     * 
     * @param queue
     *            the queue destination to send this message to
     * @param message
     *            the message to send
     * @param deliveryMode
     * @param priority
     * @param timeToLive           
     * @throws InvalidDestinationException
     *             If a client uses this method with a destination other than
     *             SQS queue destination.
     * @throws MessageFormatException
     *             If an invalid message is specified.
     * @throws UnsupportedOperationException
     *             If a client uses this method with a MessageProducer that
     *             specified a destination at creation time.
     * @throws JMSException
     *             If session is closed or internal error.
     */
    @Override
    public void send(Queue queue, Message message, int deliveryMode, int priority, long timeToLive)
            throws JMSException {
        send(queue, message);
    }
    
    /**
     * Gets the destination associated with this MessageProducer.
     * 
     * @return this producer's queue destination
     */
    @Override
    public Destination getDestination() throws JMSException {
        return sqsDestination;
    }
    
    /**
     * Closes the message producer.
     */
    @Override
    public void close() throws JMSException {

        if (closed.compareAndSet(false, true)) {
            parentSQSSession.removeProducer(this);
        }
    }
    
    /**
     * Sends a message to a destination created during the creation time of this
     * message producer.
     * 
     * @param message
     *            the message to send
     * @throws MessageFormatException
     *             If an invalid message is specified.
     * @throws UnsupportedOperationException
     *             If a client uses this method with a MessageProducer that did
     *             not specify a destination at creation time.
     * @throws JMSException
     *             If session is closed or internal error.
     */
    @Override
    public void send(Message message) throws JMSException {
        if (sqsDestination == null) {
            throw new UnsupportedOperationException(
                    "MessageProducer has to specify a destination at creation time.");
        }
        sendInternal(sqsDestination, message);
    }
  
    /**
     * Sends a message to a destination created during the creation time of this
     * message producer.
     * <P>
     * Send does not support deliveryMode, priority, and timeToLive. It will
     * ignore anything in deliveryMode, priority, and timeToLive.
     * 
     * @param message
     *            the message to send
     * @param deliveryMode
     * @param priority
     * @param timeToLive           
     * @throws MessageFormatException
     *             If an invalid message is specified.
     * @throws UnsupportedOperationException
     *             If a client uses this method with a MessageProducer that did
     *             not specify a destination at creation time.
     * @throws JMSException
     *             If session is closed or internal error.
     */
    @Override
    public void send(Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
        send(message);
    }
    
    /**
     * Sends a message to a queue destination.
     * 
     * @param destination
     *            the queue destination to send this message to
     * @param message
     *            the message to send
     * @throws InvalidDestinationException
     *             If a client uses this method with a destination other than
     *             valid SQS queue destination.
     * @throws MessageFormatException
     *             If an invalid message is specified.
     * @throws UnsupportedOperationException
     *             If a client uses this method with a MessageProducer that
     *             specified a destination at creation time.
     * @throws JMSException
     *             If session is closed or internal error.
     */
    @Override
    public void send(Destination destination, Message message) throws JMSException {
        if (destination == null) {
            throw new InvalidDestinationException("Destination cannot be null");
        }
        if (destination instanceof SQSQueueDestination) {
            send((Queue) destination, message);
        } else {
            throw new InvalidDestinationException("Incompatible implementation of Destination. Please use SQSQueueDestination implementation.");
        }
    }

    /**
     * Sends a message to a queue destination.
     * <P>
     * Send does not support deliveryMode, priority, and timeToLive. It will
     * ignore anything in deliveryMode, priority, and timeToLive.
     * 
     * @param destination
     *            the queue destination to send this message to
     * @param message
     *            the message to send
     * @param deliveryMode
     * @param priority
     * @param timeToLive
     * @throws InvalidDestinationException
     *             If a client uses this method with a destination other than
     *             valid SQS queue destination.
     * @throws MessageFormatException
     *             If an invalid message is specified.
     * @throws UnsupportedOperationException
     *             If a client uses this method with a MessageProducer that
     *             specified a destination at creation time.
     * @throws JMSException
     *             If session is closed or internal error.
     */
    @Override
    public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
        send(destination, message);
    }

    /** This method is not supported. */
    @Override
    public void setDisableMessageID(boolean value) throws JMSException {
        this.disableMessageID = value;
    }

    /** This method is not supported. */
    @Override
    public boolean getDisableMessageID() throws JMSException {
        return disableMessageID;
    }

    /** This method is not supported. */
    @Override
    public void setDisableMessageTimestamp(boolean value) throws JMSException {
        this.disableMessageTimestamp = value;
    }

    /** This method is not supported. */
    @Override
    public boolean getDisableMessageTimestamp() throws JMSException {
        return disableMessageTimestamp;
    }

    /** This method is not supported. */
    @Override
    public void setDeliveryMode(int deliveryMode) throws JMSException {
        this.deliveryMode = deliveryMode;
    }

    /** This method is not supported. */
    @Override
    public int getDeliveryMode() throws JMSException {
        return deliveryMode;
    }

    /** This method is not supported. */
    @Override
    public void setPriority(int defaultPriority) throws JMSException {
        this.defaultPriority = defaultPriority;
    }

    /** This method is not supported. */
    @Override
    public int getPriority() throws JMSException {
        return defaultPriority;
    }

    /** This method is not supported. */
    @Override
    public void setTimeToLive(long timeToLive) throws JMSException {
        this.timeToLive = timeToLive;
    }

    /** This method is not supported. */
    @Override
    public long getTimeToLive() throws JMSException {
        return timeToLive;
    }
    
    /**
     * Sets the minimum length of time in milliseconds that must elapse after a 
     * message is sent before the JMS provider may deliver the message to a consumer.
     * <p>
     * This must be a multiple of 1000, since SQS only supports delivery delays
     * in seconds.
     */
    public void setDeliveryDelay(long deliveryDelay) {
        if (deliveryDelay < 0 || deliveryDelay > MAXIMUM_DELIVERY_DELAY_MILLISECONDS) {
            throw new IllegalArgumentException("Delivery delay must be non-negative and at most 15 minutes: " + deliveryDelay);
        }
        if (deliveryDelay % 1000 != 0) {
            throw new IllegalArgumentException("Delivery delay must be a multiple of 1000: " + deliveryDelay);
        }
        this.deliveryDelaySeconds = (int)(deliveryDelay / 1000);
    }

    /**
     * Gets the minimum length of time in milliseconds that must elapse after a 
     * message is sent before the JMS provider may deliver the message to a consumer.
     */
    public long getDeliveryDelay() {
        return deliveryDelaySeconds * 1000;
    }
    
    void checkClosed() throws IllegalStateException {
        if (closed.get()) {
            throw new IllegalStateException("The producer is closed.");
        }
    }

    void checkIfDestinationAlreadySet() {
        if (sqsDestination != null) {
            throw new UnsupportedOperationException(
                    "MessageProducer already specified a destination at creation time.");
        }
    }

    /*
     * Unit Tests Utility Functions
     */

    AtomicBoolean isClosed() {
        return closed;
    }
} public class SQSQueueDestination implements Destination, Queue {
    
    private final String queueName;

    private final String queueUrl;

    private final boolean isFifo;

    SQSQueueDestination(String queueName, String queueUrl) {
        this.queueName = queueName;
        this.queueUrl = queueUrl;
        this.isFifo = this.queueName.endsWith(".fifo");
    }

    /**
     * Returns the name of this queue.
     * 
     * @return queueName
     */
    @Override
    public String getQueueName() throws JMSException {
        return this.queueName;
    }

    /**
     * Returns the queueUrl of this queue.
     * 
     * @return queueUrl
     */
    public String getQueueUrl() {
        return queueUrl;
    }

    public boolean isFifo() {
        return isFifo;
    }

    @Override
    public String toString() {
        return "SQSDestination [queueName=" + queueName + ", queueUrl=" + queueUrl + "]";
    }
    
    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((queueName == null) ? 0 : queueName.hashCode());
        result = prime * result + ((queueUrl == null) ? 0 : queueUrl.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        SQSQueueDestination other = (SQSQueueDestination) obj;
        if (queueName == null) {
            if (other.queueName != null)
                return false;
        } else if (!queueName.equals(other.queueName))
            return false;
        if (queueUrl == null) {
            if (other.queueUrl != null)
                return false;
        } else if (!queueUrl.equals(other.queueUrl))
            return false;
        return true;
    }
} public class SQSSession implements Session, QueueSession {
    private static final Log LOG = LogFactory.getLog(SQSSession.class);
    
    private static final int SESSION_EXECUTOR_GRACEFUL_SHUTDOWN_TIME = 10;
    
    static final String SESSION_EXECUTOR_NAME = "SessionCallBackScheduler";

    /** Used to create session callback scheduler threads */
    static final SQSMessagingClientThreadFactory SESSION_THREAD_FACTORY = new SQSMessagingClientThreadFactory(
            SESSION_EXECUTOR_NAME, false, true);

    static final String CONSUMER_PREFETCH_EXECUTER_NAME = "ConsumerPrefetch";

    /** Used to create consumer prefetcher threads */
    static final SQSMessagingClientThreadFactory CONSUMER_PREFETCH_THREAD_FACTORY = new SQSMessagingClientThreadFactory(
            CONSUMER_PREFETCH_EXECUTER_NAME, true);

    /**
     * Non standard acknowledge mode. This is a variation of CLIENT_ACKNOWLEDGE
     * where Clients need to remember to call acknowledge on message. Difference
     * is that calling acknowledge on a message only acknowledge the message
     * being called.
     */
    public static final int UNORDERED_ACKNOWLEDGE = 100;

    /**
     * True if Session is closed.
     */
    private volatile boolean closed = false;

    /**
     * False if Session is stopped.
     */
    volatile boolean running = false;
    
    /**
     * True if Session is closed or close is in-progress.
     */
    private volatile boolean closing = false;

    private final AmazonSQSMessagingClientWrapper amazonSQSClient;
    private final SQSConnection parentSQSConnection;

    /**
     * AcknowledgeMode of this Session.
     */
    private final AcknowledgeMode acknowledgeMode;

    /**
     * Acknowledger of this Session.
     */
    private final Acknowledger acknowledger;
    
    /**
     * Negative acknowledger of this Session
     */
    private final NegativeAcknowledger negativeAcknowledger;

    /**
     * Set of MessageProducer under this Session
     */
    private final Set<SQSMessageProducer> messageProducers;

    /**
     * Set of MessageConsumer under this Session
     */
    private final Set<SQSMessageConsumer> messageConsumers;
    
    /**
     * Thread that is responsible to guarantee serial execution of message
     * delivery on message listeners
     */
    private final SQSSessionCallbackScheduler sqsSessionRunnable;

    /**
     * Executor service for running MessageListener.
     */
    private final ExecutorService executor;

    private final Object stateLock = new Object();
    
    /**
     * Used to determine if the caller thread is the session callback thread.
     * Guarded by stateLock
     */
    private Thread activeCallbackSessionThread;

    /**
     * Used to determine the active consumer, whose is dispatching the message
     * on the callback. Guarded by stateLock
     */
    private SQSMessageConsumer activeConsumerInCallback = null;

    SQSSession(SQSConnection parentSQSConnection, AcknowledgeMode acknowledgeMode) throws JMSException{
        this(parentSQSConnection, acknowledgeMode,
                Collections.newSetFromMap(new ConcurrentHashMap<SQSMessageConsumer, Boolean>()),
                Collections.newSetFromMap(new ConcurrentHashMap<SQSMessageProducer, Boolean>()));
    }

    SQSSession(SQSConnection parentSQSConnection, AcknowledgeMode acknowledgeMode,
               Set<SQSMessageConsumer> messageConsumers,
               Set<SQSMessageProducer> messageProducers) throws JMSException{
        this.parentSQSConnection = parentSQSConnection;
        this.amazonSQSClient = parentSQSConnection.getWrappedAmazonSQSClient();
        this.acknowledgeMode = acknowledgeMode;
        this.acknowledger = this.acknowledgeMode.createAcknowledger(amazonSQSClient, this);
        this.negativeAcknowledger = new NegativeAcknowledger(amazonSQSClient);
        this.sqsSessionRunnable = new SQSSessionCallbackScheduler(this, acknowledgeMode, acknowledger, negativeAcknowledger);
        this.executor = Executors.newSingleThreadExecutor(SESSION_THREAD_FACTORY);
        this.messageConsumers = messageConsumers;
        this.messageProducers = messageProducers;

        executor.execute(sqsSessionRunnable);
    }
    
    SQSConnection getParentConnection() {
        return parentSQSConnection;
    }
    
    /**
     * @return True if the current thread is the callback thread
     */
    boolean isActiveCallbackSessionThread() {
        synchronized (stateLock) {
            return activeCallbackSessionThread == Thread.currentThread();
        }
    }
    
    /**
     * Creates a <code>QueueReceiver</code> for the specified queue.
     * @param queue
     *            a queue receiver
     * @return new message consumer
     * @throws JMSException
     *             If session is closed
     */
    @Override
    public QueueReceiver createReceiver(Queue queue) throws JMSException {
        return (QueueReceiver) createConsumer(queue);
    }

    /**
     * Creates a <code>QueueReceiver</code> for the specified queue. Does not
     * support messageSelector. It will drop anything in messageSelector.
     * 
     * @param queue
     *            a queue destination
     * @param messageSelector           
     * @return new message receiver
     * @throws JMSException
     *             If session is closed
     */
    @Override
    public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException {
        return createReceiver(queue);
    }
    
    /**
     * Creates a <code>QueueSender</code> for the specified queue.
     * 
     * @param queue
     *            a queue destination 
     * @return new message sender
     * @throws JMSException
     *             If session is closed
     */
    @Override
    public QueueSender createSender(Queue queue) throws JMSException {
        return (QueueSender) createProducer(queue);
    }
    
    /**
     * Creates a <code>BytesMessage</code>.
     * 
     * @return new <code>BytesMessage</code>
     * @throws JMSException
     *             If session is closed or internal error
     */
    @Override
    public BytesMessage createBytesMessage() throws JMSException {
        checkClosed();
        return new SQSBytesMessage();
    }
   
    /**
     * According to JMS specification, a message can be sent with only headers
     * without any payload, SQS does not support messages with empty payload. so
     * this method is not supported
     */
    @Override
    public Message createMessage() throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }
    
    /**
     * Creates a <code>ObjectMessage</code>.
     * 
     * @return new <code>ObjectMessage</code>
     * @throws JMSException
     *             If session is closed or internal error
     */
    @Override
    public ObjectMessage createObjectMessage() throws JMSException {
        checkClosed();
        return new SQSObjectMessage();
    }
    
    /**
     * Creates an initialized <code>ObjectMessage</code>.
     * 
     * @param object
     *            The initialized <code>ObjectMessage</code>
     * @return new <code>ObjectMessage</code>
     * @throws JMSException
     *             If session is closed or internal error
     */
    @Override
    public ObjectMessage createObjectMessage(Serializable object) throws JMSException {
        checkClosed();
        return new SQSObjectMessage(object);
    }
    
    /**
     * Creates a <code>TextMessage</code>.
     * 
     * @return new <code>TextMessage</code>
     * @throws JMSException
     *             If session is closed or internal error
     */
    @Override
    public TextMessage createTextMessage() throws JMSException {
        checkClosed();
        return new SQSTextMessage();
    }
    
    /**
     * Creates an initialized <code>TextMessage</code>.
     * 
     * @param text
     *            The initialized <code>TextMessage</code>
     * @return new <code>TextMessage</code>
     * @throws JMSException
     *             If session is closed or internal error
     */
    @Override
    public TextMessage createTextMessage(String text) throws JMSException {
        checkClosed();
        return new SQSTextMessage(text);
    }

    /**
     * Returns the acknowledge mode of the session. The acknowledge mode is set
     * at the time that the session is created.
     * 
     * @return acknowledge mode
     */
    @Override
    public int getAcknowledgeMode() throws JMSException {
        return acknowledgeMode.getOriginalAcknowledgeMode();
    }
    
    /**
     * Closes the session.
     * <P>
     * This will not return until all the message consumers and producers close
     * internally, which blocks until receives and/or message listeners in
     * progress have completed. A blocked message consumer receive call returns
     * null when this session is closed.
     * <P>
     * Since consumer prefetch threads use SQS long-poll feature with 20 seconds
     * timeout, closing each consumer prefetch thread can take up to 20 seconds,
     * which in-turn will impact the time on session close.
     * <P>
     * This method is safe for concurrent use.
     * <P>
     * A message listener must not attempt to close its own session; otherwise
     * throws a IllegalStateException.
     * <P>
     * Invoking any other session method on a closed session must throw a
     * <code>IllegalStateException</code>.
     * 
     * @throws IllegalStateException
     *             If called by a message listener on its own
     *             <code>Session</code>.
     * @throws JMSException
     *             On internal error.
     */
    @Override
    public void close() throws JMSException {

        if (closed) {
            return;
        }
        
        /**
         * A MessageListener must not attempt to close its own Session as
         * this would lead to deadlock
         */
        if (isActiveCallbackSessionThread()) {
            throw new IllegalStateException(
                    "MessageListener must not attempt to close its own Session to prevent potential deadlock issues");
        }
        
        doClose();
    }

    void doClose() throws JMSException {
        boolean shouldClose = false;
        synchronized (stateLock) {
            if (!closing) {
                shouldClose = true;
                closing = true;
            }
            stateLock.notifyAll();
        }
        
        if (closed) {
            return;
        }

        if (shouldClose) {
            try {
                parentSQSConnection.removeSession(this);

                for (SQSMessageConsumer messageConsumer : messageConsumers) {
                    messageConsumer.close();
                }
                
                recover();
                
                try {
                    if (executor != null) {
                        LOG.info("Shutting down " + SESSION_EXECUTOR_NAME + " executor");

                        /** Shut down executor. */
                        executor.shutdown();
                        
                        waitForCallbackComplete();
                        
                        sqsSessionRunnable.close();
                        
                        for (MessageProducer messageProducer : messageProducers) {
                            messageProducer.close();
                        }

                        if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {

                            LOG.warn("Can't terminate executor service " + SESSION_EXECUTOR_NAME + " after " +
                                     SESSION_EXECUTOR_GRACEFUL_SHUTDOWN_TIME +
                                     " seconds, some running threads will be shutdown immediately");
                            executor.shutdownNow();
                        }
                    }
                } catch (InterruptedException e) {
                    LOG.error("Interrupted while closing the session.", e);
                }
               
            } finally {
                synchronized (stateLock) {
                    closed = true;
                    running = false;
                    stateLock.notifyAll();
                }
            }
        }/** Blocks until closing of the session completes */
        else {
            synchronized (stateLock) {
                while (!closed) {
                    try {
                        stateLock.wait();
                    } catch (InterruptedException e) {
                        LOG.error("Interrupted while waiting the session to close.", e);
                    }
                }
            }
        }
    }


    /**
     * Negative acknowledges all the messages on the session that is delivered
     * but not acknowledged.
     * 
     * @throws JMSException
     *             If session is closed or on internal error.
     */
    @Override
    public void recover() throws JMSException {
        checkClosed();
        
        //let's get all unacknowledged message identifiers
        List<SQSMessageIdentifier> unAckedMessages = acknowledger.getUnAckMessages();
        acknowledger.forgetUnAckMessages();
        
        //let's summarize which queues and which message groups we're nacked
        //we have to purge all prefetched messages and queued up callback entries for affected queues and groups
        //if not, we would end up consuming messages out of order
        Map<String, Set<String>> queueToGroupsMapping = getAffectedGroupsPerQueueUrl(unAckedMessages);

        for (SQSMessageConsumer consumer : this.messageConsumers) {
            SQSQueueDestination sqsQueue = (SQSQueueDestination)consumer.getQueue();
            Set<String> affectedGroups = queueToGroupsMapping.get(sqsQueue.getQueueUrl());
            if (affectedGroups != null) {
                unAckedMessages.addAll(consumer.purgePrefetchedMessagesWithGroups(affectedGroups));
            }
        }
        
        unAckedMessages.addAll(sqsSessionRunnable.purgeScheduledCallbacksForQueuesAndGroups(queueToGroupsMapping));
        
        if (!unAckedMessages.isEmpty()) {
            negativeAcknowledger.bulkAction(unAckedMessages, unAckedMessages.size());
        }
    }

    private Map<String, Set<String>> getAffectedGroupsPerQueueUrl(List<SQSMessageIdentifier> messages) {
        Map<String, Set<String>> queueToGroupsMapping = new HashMap<String, Set<String>>();
        for (SQSMessageIdentifier message : messages) {
            String groupId = message.getGroupId();
            if (groupId != null) {
                String queueUrl = message.getQueueUrl();
                if (!queueToGroupsMapping.containsKey(queueUrl)) {
                    queueToGroupsMapping.put(queueUrl, new HashSet<String>());
                }
                queueToGroupsMapping.get(queueUrl).add(groupId);
            }
        }
        return queueToGroupsMapping;
    }

    @Override
    public void run() {
    }
    
    /**
     * Creates a <code>MessageProducer</code> for the specified destination.
     * Only queue destinations are supported at this time.
     * 
     * @param destination
     *            a queue destination
     * @return new message producer
     * @throws JMSException
     *             If session is closed or queue destination is not used
     */
    @Override
    public MessageProducer createProducer(Destination destination) throws JMSException {
        checkClosed();
        if (destination != null && !(destination instanceof SQSQueueDestination)) {
            throw new JMSException("Actual type of Destination/Queue has to be SQSQueueDestination");
        }
        SQSMessageProducer messageProducer;
        synchronized (stateLock) {
            checkClosing();
            messageProducer = new SQSMessageProducer(amazonSQSClient, this, (SQSQueueDestination) destination);
            messageProducers.add(messageProducer);
        }
        return messageProducer;
    }
    
    /**
     * Creates a <code>MessageConsumer</code> for the specified destination.
     * Only queue destinations are supported at this time.
     * 
     * @param destination
     *            a queue destination
     * @return new message consumer
     * @throws JMSException
     *             If session is closed or queue destination is not used
     */
    @Override
    public MessageConsumer createConsumer(Destination destination) throws JMSException {
        checkClosed();
        if (!(destination instanceof SQSQueueDestination)) {
            throw new JMSException("Actual type of Destination/Queue has to be SQSQueueDestination");
        }
        SQSMessageConsumer messageConsumer;
        synchronized (stateLock) {
            checkClosing();
            messageConsumer = createSQSMessageConsumer((SQSQueueDestination) destination);
            messageConsumers.add(messageConsumer);
            if( running ) {
                messageConsumer.startPrefetch();
            }
        }
        return messageConsumer;
    }

    SQSMessageConsumer createSQSMessageConsumer(SQSQueueDestination destination) {
        return new SQSMessageConsumer(
                parentSQSConnection, this, sqsSessionRunnable, (SQSQueueDestination) destination,
                acknowledger,  negativeAcknowledger,
                CONSUMER_PREFETCH_THREAD_FACTORY);
    }

    /**
     *
     * Creates a <code>MessageConsumer</code> for the specified destination.
     * Only queue destinations are supported at this time.
     * It will ignore any argument in messageSelector.
     * 
     * @param destination
     *            a queue destination
     * @param messageSelector           
     * @return new message consumer
     * @throws JMSException
     *             If session is closed or queue destination is not used
     */

    @Override
    public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException {
        if (messageSelector != null) {
            throw new JMSException("SQSSession does not support MessageSelector. This should be null.");
        }
        return createConsumer(destination);
    }
    
    /**
     * Creates a <code>MessageConsumer</code> for the specified destination.
     * Only queue destinations are supported at this time. It will ignore any
     * argument in messageSelector and NoLocal.
     * 
     * @param destination
     *            a queue destination
     * @param messageSelector
     * @param NoLocal
     * @return new message consumer
     * @throws JMSException
     *             If session is closed or queue destination is not used
     */
    @Override
    public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean NoLocal) throws JMSException {
        if (messageSelector != null) {
            throw new JMSException("SQSSession does not support MessageSelector. This should be null.");
        }
        return createConsumer(destination);
    }

    /**
     * This does not create SQS Queue. This method is only to create JMS Queue Object.
     * Make sure the queue exists corresponding to the queueName.
     * @param queueName
     * @return a queue destination
     * @throws JMSException
     *         If session is closed or invalid queue is provided
     */
    @Override
    public Queue createQueue(String queueName) throws JMSException {
        checkClosed();
        return new SQSQueueDestination(queueName, amazonSQSClient.getQueueUrl(queueName).getQueueUrl());
    }
    
    /**
     * This does not create SQS Queue. This method is only to create JMS Queue
     * Object. Make sure the queue exists corresponding to the queueName and
     * ownerAccountId.
     * 
     * @param queueName
     * @param ownerAccountId
     *            the account id, which originally created the queue on SQS
     * @return a queue destination
     * @throws JMSException
     *             If session is closed or invalid queue is provided
     */
    public Queue createQueue(String queueName, String ownerAccountId) throws JMSException {
        checkClosed();
        return new SQSQueueDestination(
                queueName, amazonSQSClient.getQueueUrl(queueName, ownerAccountId).getQueueUrl());
    }

    /**
     * This is used in MessageConsumer. When MessageConsumer is closed
     * it will remove itself from list of consumers.
     */
    void removeConsumer(SQSMessageConsumer consumer) {
        messageConsumers.remove(consumer);
    }

    /**
     * This is used in MessageProducer. When MessageProducer is closed
     * it will remove itself from list of producers.
     */
    void removeProducer(SQSMessageProducer producer) {
        messageProducers.remove(producer);    
    }
    
    void startingCallback(SQSMessageConsumer consumer) throws InterruptedException, JMSException {
        if (closed) {
            return;
        }
        synchronized (stateLock) {
            if (activeConsumerInCallback != null) {
                throw new IllegalStateException("Callback already in progress");
            }
            assert activeCallbackSessionThread == null;

            while (!running && !closing) {
                try {
                    stateLock.wait();
                } catch (InterruptedException e) {
                    LOG.warn("Interrupted while waiting on session start. Continue to wait...", e);
                }
            }
            checkClosing();
            activeConsumerInCallback = consumer;
            activeCallbackSessionThread = Thread.currentThread();
        }
    }
    
    void finishedCallback() throws JMSException {
        synchronized (stateLock) {
            if (activeConsumerInCallback == null) {
                throw new IllegalStateException("Callback not in progress");
            }
            activeConsumerInCallback = null;
            activeCallbackSessionThread = null;
            stateLock.notifyAll();
        }
    }
    
    void waitForConsumerCallbackToComplete(SQSMessageConsumer consumer) throws InterruptedException {
        synchronized (stateLock) {
            while (activeConsumerInCallback == consumer) {
                try {
                    stateLock.wait();
                } catch (InterruptedException e) {
                    LOG.warn(
                            "Interrupted while waiting the active consumer in callback to complete. Continue to wait...",
                            e);
                }
            }
        }
    }
    
    void waitForCallbackComplete() {
        synchronized (stateLock) {
            while (activeConsumerInCallback != null) {
                try {
                    stateLock.wait();
                } catch (InterruptedException e) {
                    LOG.warn("Interrupted while waiting on session callback completion. Continue to wait...", e);
                }
            }
        }
    }

    /** SQS does not support transacted. Transacted will always be false. */
    @Override
    public boolean getTransacted() throws JMSException {
        return false;
    }

    /** This method is not supported. This method is related to transaction which SQS doesn't support */
    @Override
    public void commit() throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. This method is related to transaction which SQS doesn't support */
    @Override
    public void rollback() throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. This method is related to Topic which SQS doesn't support */
    @Override
    public void unsubscribe(String name) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public Topic createTopic(String topicName) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public QueueBrowser createBrowser(Queue queue) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public TemporaryQueue createTemporaryQueue() throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public TemporaryTopic createTemporaryTopic() throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public MessageListener getMessageListener() throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public void setMessageListener(MessageListener listener) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public StreamMessage createStreamMessage() throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    /** This method is not supported. */
    @Override
    public MapMessage createMapMessage() throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }

    static class CallbackEntry {
        private final MessageListener messageListener;
        private final MessageManager messageManager;
        
        CallbackEntry(MessageListener messageListener, MessageManager messageManager) {
            this.messageListener = messageListener;
            this.messageManager = messageManager;
        }
        
        public MessageListener getMessageListener() {
            return messageListener;
        }
        
        public MessageManager getMessageManager() {
            return messageManager;
        }
    }
    
    /**
     * Check if session is closed.
     */
    public void checkClosed() throws IllegalStateException {
        if (closed) {
            throw new IllegalStateException("Session is closed");
        }
    }
    
    /**
     * Check if session is closed or closing.
     */
    public void checkClosing() throws IllegalStateException {
        if (closing) {
            throw new IllegalStateException("Session is closed or closing");
        }
    }
        
    void start() throws IllegalStateException {
        checkClosed();
        synchronized (stateLock) {
            checkClosing();
            running = true;            
            for (SQSMessageConsumer messageConsumer : messageConsumers) {
                messageConsumer.startPrefetch();
            }
            stateLock.notifyAll();
        }
    }
    
    void stop() throws IllegalStateException {
        checkClosed();
        synchronized (stateLock) {
            checkClosing();
            running = false;
            for (SQSMessageConsumer messageConsumer : messageConsumers) {
                messageConsumer.stopPrefetch();
            }
            waitForCallbackComplete();

            stateLock.notifyAll();
        }
    }

    /*
     * Unit Tests Utility Functions
     */

    boolean isCallbackActive() {
        return activeConsumerInCallback != null;
    }
    
    void setActiveConsumerInCallback(SQSMessageConsumer consumer) {
        activeConsumerInCallback = consumer;
    }
    
    Object getStateLock() {
        return stateLock;
    }

    boolean isClosed() {
        return closed;
    }
    
    boolean isClosing() {
        return closing;
    }

    void setClosed(boolean closed) {
        this.closed = closed;
    }

    void setClosing(boolean closing) {
        this.closing = closing;
    }

    void setRunning(boolean running) {
        this.running = running;
    }

    boolean isRunning() {
        return running;
    }

    SQSSessionCallbackScheduler getSqsSessionRunnable() {
        return sqsSessionRunnable;
    }
} public interface Acknowledger {

    /**
     * Generic Acknowledge method. This method will delete message(s) in SQS Queue.
     * 
     * @param message
     *            message to acknowledge.
     * @throws JMSException
     */
    public void acknowledge(SQSMessage message) throws JMSException;

    /**
     * Used when receiving messages. Depending on acknowledge mode this will
     * help create list of message backlog.
     * 
     * @param message
     *            notify acknowledger message is received
     * @throws JMSException
     */
    public void notifyMessageReceived(SQSMessage message) throws JMSException;

    /**
     * Used in negative acknowledge. Gets all delivered but not acknowledged
     * messages.
     */
    public List<SQSMessageIdentifier> getUnAckMessages();

    /**
     * Deletes all not acknowledged delivered messages.
     */
    public void forgetUnAckMessages();

} public class SQSBytesMessage extends SQSMessage implements BytesMessage {
    private static final Log LOG = LogFactory.getLog(SQSBytesMessage.class);

    private byte[] bytes;

    private ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();

    private DataInputStream dataIn;
    
    private DataOutputStream dataOut = new DataOutputStream(bytesOut);

    /**
     * Convert received SQSMessage into BytesMessage.
     */
    public SQSBytesMessage(Acknowledger acknowledger, String queueUrl, Message sqsMessage) throws JMSException {
        super(acknowledger, queueUrl, sqsMessage);
        try {
            /** Bytes is set by the reset() */
            dataOut.write(Base64.decode(sqsMessage.getBody()));
            /** Makes it read-only */
            reset();
        } catch (IOException e) {
            LOG.error("IOException: Message cannot be written", e);
            throw convertExceptionToJMSException(e);
        } catch (Exception e) {
            LOG.error("Unexpected exception: ", e);
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Create new empty BytesMessage to send.
     */
    public SQSBytesMessage() throws JMSException {
        super();
    }
    
    /**
     * Gets the number of bytes of the message body when the message is in
     * read-only mode. The value returned can be used to allocate a byte array.
     * The value returned is the entire length of the message body, regardless
     * of where the pointer for reading the message is currently located.
     * 
     * @return number of bytes in the message
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public long getBodyLength() throws JMSException {
        checkCanRead();
        return bytes.length;
    }
    
    /**
     * Reads a <code>boolean</code> from the bytes message stream.
     * 
     * @return the <code>boolean</code> value
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public boolean readBoolean() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readBoolean();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a signed 8-bit value from the bytes message stream.
     * 
     * @return the next byte from the bytes message stream as a signed 8-bit
     *         byte
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public byte readByte() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readByte();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads an unsigned 8-bit value from the bytes message stream.
     * 
     * @return the next byte from the bytes message stream, interpreted as an
     *         unsigned 8-bit number
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public int readUnsignedByte() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readUnsignedByte();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a signed 16-bit number from the bytes message stream.
     * 
     * @return the next two bytes from the bytes message stream, interpreted as
     *         a signed 16-bit number
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public short readShort() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readShort();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads an unsigned 16-bit number from the bytes message stream.
     * 
     * @return the next two bytes from the bytes message stream, interpreted as
     *         an unsigned 16-bit integer
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public int readUnsignedShort() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readUnsignedShort();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a Unicode character value from the bytes message stream. 
     * 
     * @return a Unicode character from the bytes message stream
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public char readChar() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readChar();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a 32-bit integer from the bytes message stream.
     * 
     * @return the next four bytes from the bytes message stream, interpreted as an int
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public int readInt() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readInt();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a 64-bit integer from the bytes message stream.
     * 
     * @return a 64-bit integer value from the bytes message stream, interpreted as a long
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public long readLong() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readLong();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a <code>float</code> from the bytes message stream.
     * 
     * @return a <code>float</code> value from the bytes message stream
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public float readFloat() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readFloat();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a <code>double</code> from the bytes message stream. 
     * 
     * @return a <code>double</code> value from the bytes message stream
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public double readDouble() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readDouble();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a string that has been encoded using a UTF-8 format from
     * the bytes message stream
     * 
     * @return a Unicode string from the bytes message stream
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageEOFException
     *             If unexpected end of bytes stream has been reached.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public String readUTF() throws JMSException {
        checkCanRead();
        try {
            return dataIn.readUTF();
        } catch (EOFException e) {
            throw new MessageEOFException(e.getMessage());
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Reads a byte array from the bytes message stream.
     * <P>
     * If the length of array value is less than the number of bytes remaining
     * to be read from the stream, the array should be filled. A subsequent call
     * reads the next increment, and so on.
     * <P>
     * If the number of bytes remaining in the stream is less than the length of
     * array value, the bytes should be read into the array. The return value of
     * the total number of bytes read will be less than the length of the array,
     * indicating that there are no more bytes left to be read from the stream.
     * The next read of the stream returns -1.
     * 
     * @param value
     *            The buffer into which the data is read
     * @return the total number of bytes read into the buffer, or -1 if there is
     *         no more data because the end of the stream has been reached
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @Override
    public int readBytes(byte[] value) throws JMSException {
        return readBytes(value, value.length);
    }
    
    
    /**
     * Reads a portion of the bytes message stream.
     * <P>
     * If the length of array value is less than the number of bytes remaining
     * to be read from the stream, the array should be filled. A subsequent call
     * reads the next increment, and so on.
     * <P>
     * If the number of bytes remaining in the stream is less than the length of
     * array value, the bytes should be read into the array. The return value of
     * the total number of bytes read will be less than the length of the array,
     * indicating that there are no more bytes left to be read from the stream.
     * The next read of the stream returns -1.
     * <P>
     * If length is negative, then an <code>IndexOutOfBoundsException</code> is
     * thrown. No bytes will be read from the stream for this exception case.
     * 
     * @param value
     *            The buffer into which the data is read
     * @param length
     *            The number of bytes to read; must be less than or equal to
     *            value.length
     * @return the total number of bytes read into the buffer, or -1 if there is
     *         no more data because the end of the stream has been reached
     * @throws JMSException
     *             If the JMS provider fails to read the message due to some
     *             internal error.
     * @throws MessageNotReadableException
     *             If the message is in write-only mode.
     */
    @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);
        }
        checkCanRead();
        try {
            /**
             * Almost copy of readFully implementation except that EOFException
             * is not thrown if the stream is at the end of file and no byte is
             * available
             */
            int n = 0;
            while (n < length) {
                int count = dataIn.read(value, n, length - n);
                if (count < 0) {
                    break;
                }
                n += count;
            }
            /**
             * JMS specification mentions that the next read of the stream
             * returns -1 if the previous read consumed the byte stream and
             * there are no more bytes left to be read from the stream
             */
            if (n == 0 && length > 0) {
                n = -1;
            }
            return n;
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a <code>boolean</code> to the bytes message stream
     * 
     * @param value
     *            The <code>boolean</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeBoolean(boolean value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeBoolean(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a <code>byte</code> to the bytes message stream
     * 
     * @param value
     *            The <code>byte</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeByte(byte value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeByte(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a <code>short</code> to the bytes message stream
     * 
     * @param value
     *            The <code>short</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeShort(short value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeShort(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a <code>char</code> to the bytes message stream
     * 
     * @param value
     *            The <code>char</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeChar(char value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeChar(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a <code>int</code> to the bytes message stream
     * 
     * @param value
     *            The <code>int</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeInt(int value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeInt(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a <code>long</code> to the bytes message stream
     * 
     * @param value
     *            The <code>long</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeLong(long value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeLong(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a <code>float</code> to the bytes message stream
     * 
     * @param value
     *            The <code>float</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeFloat(float value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeFloat(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a <code>double</code> to the bytes message stream
     * 
     * @param value
     *            The <code>double</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeDouble(double value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeDouble(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a string that has been encoded using a UTF-8 format to the bytes
     * message stream
     * 
     * @param value
     *            The <code>String</code> value to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeUTF(String value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.writeUTF(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a byte array to the bytes
     * message stream
     * 
     * @param value
     *            The byte array value to be written 
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeBytes(byte[] value) throws JMSException {
        checkCanWrite();
        try {
            dataOut.write(value);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes a portion of a byte array to the bytes message stream.
     * 
     * @param value
     *            The portion of byte array value to be written
     * @param offset
     *            The initial offset within the byte array
     * @param length
     *            The number of bytes to use 
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void writeBytes(byte[] value, int offset, int length) throws JMSException {
        checkCanWrite();
        try {
            dataOut.write(value, offset, length);
        } catch (IOException e) {
            throw convertExceptionToJMSException(e);
        }
    }
    
    /**
     * Writes an object to the bytes message stream.
     * <P>
     * This method works only for the boxed primitive object types
     * (Integer, Double, Long ...), String objects, and byte arrays.
     * 
     * @param value
     *            The Java object to be written
     * @throws JMSException
     *             If the JMS provider fails to write the message due to some
     *             internal error.
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     * @throws MessageFormatException
     *             If the object is of an invalid type.
     * @throws NullPointerException
     *             If the object is null.
     */
    @Override
    public void writeObject(Object value) throws JMSException {
        if (value == null) {
            throw new NullPointerException("Cannot write null value of object");
        }
        if (value instanceof Boolean) {
            writeBoolean((Boolean) value);
        } else if (value instanceof Character) {
            writeChar((Character) value);
        } else if (value instanceof Byte) {
            writeByte((Byte) value);
        } else if (value instanceof Short) {
            writeShort((Short) value);
        } else if (value instanceof Integer) {
            writeInt((Integer) value);
        } else if (value instanceof Long) {
            writeLong((Long) value);
        } else if (value instanceof Float) {
            writeFloat((Float) value);
        } else if (value instanceof Double) {
            writeDouble((Double) value);
        } else if (value instanceof String) {
            writeUTF(value.toString());
        } else if (value instanceof byte[]) {
            writeBytes((byte[]) value);
        } else {
            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.
     */
    @Override
    public void reset() throws JMSException {

        if (dataOut != null) {
            bytes = bytesOut.toByteArray();
            dataOut = null;
            bytesOut = null;
        }
        dataIn = new DataInputStream(new ByteArrayInputStream(bytes));
    }
    
    /**
     * When the message is first created, and when clearBody is called, the body
     * of the message is in write-only mode. After the first call to reset is
     * made, the message body goes to read-only mode. when the message is sent,
     * the sender can retain and modify it without affecting the sent message.
     * If clearBody is called on a message, when it is in read-only mode, the
     * message body is cleared and the message goes to write-only mode.
     */
    @Override
    public void clearBody() throws JMSException {
        bytes = null;
        dataIn = null;
        bytesOut = new ByteArrayOutputStream();
        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.
     * 
     * @return value The body returned as byte array
     */
    public byte[] getBodyAsBytes() throws JMSException {
        if (bytes != null) {
            return Arrays.copyOf(bytes, bytes.length);
        } else {
            return bytesOut.toByteArray();
        }
    }

    void checkCanRead() throws JMSException {
        if (bytes == null) {
            throw new MessageNotReadableException("Message is not readable");
        }
    }

    void checkCanWrite() throws JMSException {
        if (dataOut == null) {
            throw new MessageNotWriteableException("Message is not writeable");
        }
    }

    /*
     * Unit Test Utility Function
     */

    void setDataIn(DataInputStream dataIn) {
        this.dataIn = dataIn;
    }

    void setDataOut(DataOutputStream dataOut) {
        this.dataOut = dataOut;
    }
} public class SQSMessage implements Message {
        
    private static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
    
    // Define constant message types.
    public static final String BYTE_MESSAGE_TYPE = "byte";
    public static final String OBJECT_MESSAGE_TYPE = "object";
    public static final String TEXT_MESSAGE_TYPE = "text";
    public static final String JMS_SQS_MESSAGE_TYPE = "JMS_SQSMessageType";
    public static final String JMS_SQS_REPLY_TO_QUEUE_NAME = "JMS_SQSReplyToQueueName";
    public static final String JMS_SQS_REPLY_TO_QUEUE_URL = "JMS_SQSReplyToQueueURL";
    public static final String JMS_SQS_CORRELATION_ID = "JMS_SQSCorrelationID";
    
    // Default JMS Message properties
    private int deliveryMode = Message.DEFAULT_DELIVERY_MODE;
    private int priority = Message.DEFAULT_PRIORITY;
    private long timestamp;
    private boolean redelivered;
    private String correlationID;
    private long expiration = Message.DEFAULT_TIME_TO_LIVE;
    private String messageID;
    private String type;
    private SQSQueueDestination replyTo;
    private Destination destination;

    private final Map<String, JMSMessagePropertyValue> properties = new HashMap<String, JMSMessagePropertyValue>();

    private boolean writePermissionsForProperties;
    private boolean writePermissionsForBody;

    /**
     * Function for acknowledging message. 
     */
    private Acknowledger acknowledger;
    
    /**
     * Original SQS Message ID.
     */
    private String sqsMessageID;

    /**
     * QueueUrl the message came from.
     */
    private String queueUrl;
    /**
     * Original SQS Message receipt handle.
     */
    private String receiptHandle;

    /**
     * This is called at the receiver side to create a
     * JMS message from the SQS message received.
     */
    SQSMessage(Acknowledger acknowledger, String queueUrl, com.amazonaws.services.sqs.model.Message sqsMessage) throws JMSException{
        this.acknowledger = acknowledger;
        this.queueUrl = queueUrl;
        receiptHandle = sqsMessage.getReceiptHandle();
        this.setSQSMessageId(sqsMessage.getMessageId());
        Map<String,String> systemAttributes = sqsMessage.getAttributes();
        int receiveCount = Integer.parseInt(systemAttributes.get(APPROXIMATE_RECEIVE_COUNT));
        
        /**
         * JMSXDeliveryCount is set based on SQS ApproximateReceiveCount
         * attribute.
         */
        properties.put(JMSX_DELIVERY_COUNT, new JMSMessagePropertyValue(
                receiveCount, INT));
        if (receiveCount > 1) {
            setJMSRedelivered(true);
        }
        if (sqsMessage.getMessageAttributes() != null) {
            addMessageAttributes(sqsMessage);
        }

        // map the SequenceNumber, MessageGroupId and MessageDeduplicationId to JMS specific properties
        mapSystemAttributeToJmsMessageProperty(systemAttributes, SEQUENCE_NUMBER, JMS_SQS_SEQUENCE_NUMBER);
        mapSystemAttributeToJmsMessageProperty(systemAttributes, MESSAGE_DEDUPLICATION_ID, JMS_SQS_DEDUPLICATION_ID);
        mapSystemAttributeToJmsMessageProperty(systemAttributes, MESSAGE_GROUP_ID, JMSX_GROUP_ID);

        writePermissionsForBody = false;
        writePermissionsForProperties = false;
    }
    
    private void mapSystemAttributeToJmsMessageProperty(Map<String,String> systemAttributes, String systemAttributeName, String jmsMessagePropertyName) throws JMSException {
        String systemAttributeValue = systemAttributes.get(systemAttributeName);
        if (systemAttributeValue != null) {
            properties.put(jmsMessagePropertyName, new JMSMessagePropertyValue(systemAttributeValue, STRING));
        }
    }

    /**
     * Create new empty Message to send. SQSMessage cannot be sent without any
     * payload. One of SQSTextMessage, SQSObjectMessage, or SQSBytesMessage
     * should be used to add payload.
     */
    SQSMessage() {
        writePermissionsForBody = true;
        writePermissionsForProperties = true;
    }

    private void addMessageAttributes(com.amazonaws.services.sqs.model.Message sqsMessage) throws JMSException {
        for (Entry<String, MessageAttributeValue> entry : sqsMessage.getMessageAttributes().entrySet()) {
            properties.put(entry.getKey(), new JMSMessagePropertyValue(
                    entry.getValue().getStringValue(), entry.getValue().getDataType()));
        }
    }

    protected void checkPropertyWritePermissions() throws JMSException {
        if (!writePermissionsForProperties) {
            throw new MessageNotWriteableException("Message properties are not writable");
        }
    }
    
    protected void checkBodyWritePermissions() throws JMSException {
        if (!writePermissionsForBody) {
            throw new MessageNotWriteableException("Message body is not writable");
        }
    }
    
    protected static JMSException convertExceptionToJMSException(Exception e) {
        JMSException ex = new JMSException(e.getMessage());
        ex.initCause(e);
        return ex;
    }
    
    protected static MessageFormatException convertExceptionToMessageFormatException(Exception e) {
        MessageFormatException ex = new MessageFormatException(e.getMessage());
        ex.initCause(e);
        return ex;
    }
    
    protected void setBodyWritePermissions(boolean enable) {
        writePermissionsForBody = enable;
    }
    
    /**
     * Get SQS Message Group Id (applicable for FIFO queues, available also as JMS property 'JMSXGroupId')
     * @throws JMSException 
     */
    public String getSQSMessageGroupId() throws JMSException {
        return getStringProperty(SQSMessagingClientConstants.JMSX_GROUP_ID);
    }
    
    /**
     * Get SQS Message Deduplication Id (applicable for FIFO queues, available also as JMS property 'JMS_SQS_DeduplicationId')
     * @throws JMSException 
     */
    public String getSQSMessageDeduplicationId() throws JMSException {
        return getStringProperty(SQSMessagingClientConstants.JMS_SQS_DEDUPLICATION_ID);
    }
    
    /**
     * Get SQS Message Sequence Number (applicable for FIFO queues, available also as JMS property 'JMS_SQS_SequenceNumber')
     * @throws JMSException 
     */
    public String getSQSMessageSequenceNumber() throws JMSException {
        return getStringProperty(SQSMessagingClientConstants.JMS_SQS_SEQUENCE_NUMBER);
    }
    
    /**
     * Get SQS Message Id.
     * 
     * @return SQS Message Id.
     */
    public String getSQSMessageId() {
        return sqsMessageID;
    }
    
    /**
     * Set SQS Message Id, used on send.
     * 
     * @param sqsMessageID
     *            messageId assigned by SQS during send.
     */
    public void setSQSMessageId(String sqsMessageID) throws JMSException {
        this.sqsMessageID = sqsMessageID;
        this.setJMSMessageID(String.format(SQSMessagingClientConstants.MESSAGE_ID_FORMAT, sqsMessageID));
    }
        
    /**
     * Get SQS Message receiptHandle.
     * 
     * @return SQS Message receiptHandle.
     */
    public String getReceiptHandle() {
        return receiptHandle;
    }

    /**
     * Get queueUrl the message came from.
     * 
     * @return queueUrl.
     */
    public String getQueueUrl() {
        return queueUrl;
    }
    
    /**
     * Gets the message ID.
     * <P>
     * The JMSMessageID header field contains a value that uniquely identifies
     * each message sent by a provider. It is set to SQS messageId with the
     * prefix 'ID:'.
     * 
     * @return the ID of the message.
     */
    @Override
    public String getJMSMessageID() throws JMSException {
        return messageID;
    }
    
    /**
     * Sets the message ID. It should have prefix 'ID:'.
     * <P>
     * Set when a message is sent. This method can be used to change the value
     * for a message that has been received.
     * 
     * @param id
     *            The ID of the message.
     */
    @Override
    public void setJMSMessageID(String id) throws JMSException {
        messageID = id;
    }

    @Override
    public long getJMSTimestamp() throws JMSException {
        return timestamp;
    }

    @Override
    public void setJMSTimestamp(long timestamp) throws JMSException {
        this.timestamp = timestamp;
    }

    @Override
    public byte[] getJMSCorrelationIDAsBytes() throws JMSException {
        return correlationID != null ? correlationID.getBytes(DEFAULT_CHARSET) : null;
    }

    @Override
    public void setJMSCorrelationIDAsBytes(byte[] correlationID) throws JMSException {
        try {
            this.correlationID = correlationID != null ? new String(correlationID, "UTF-8") : null;
        } catch (UnsupportedEncodingException e) {
            throw new JMSException(e.getMessage());
        }
    }

    @Override
    public void setJMSCorrelationID(String correlationID) throws JMSException {
        this.correlationID = correlationID;
    }

    @Override
    public String getJMSCorrelationID() throws JMSException {
        return correlationID;
    }

    @Override
    public Destination getJMSReplyTo() throws JMSException {
        return replyTo;
    }

    @Override
    public void setJMSReplyTo(Destination replyTo) throws JMSException {
        if (replyTo != null && !(replyTo instanceof SQSQueueDestination)) {
            throw new IllegalArgumentException("The replyTo Destination must be a SQSQueueDestination");
        }
        this.replyTo = (SQSQueueDestination)replyTo;
    }
    
    /**
     * Gets the Destination object for this message.
     * <P>
     * The JMSDestination header field contains the destination to which the
     * message is being sent.
     * <P>
     * When a message is sent, this field is ignored. After completion of the
     * send or publish method, the field holds the destination specified by the
     * method.
     * <P>
     * When a message is received, its JMSDestination value must be equivalent
     * to the value assigned when it was sent.
     * 
     * @return The destination of this message.
     */
    @Override
    public Destination getJMSDestination() throws JMSException {
        return destination;
    }
    
    /**
     * Sets the Destination object for this message.
     * <P>
     * Set when a message is sent. This method can be used to change the value
     * for a message that has been received.
     * 
     * @param destination
     *            The destination for this message.
     */
    @Override
    public void setJMSDestination(Destination destination) throws JMSException {
        this.destination = destination;
    }

    @Override
    public int getJMSDeliveryMode() throws JMSException {
        return deliveryMode;
    }

    @Override
    public void setJMSDeliveryMode(int deliveryMode) throws JMSException {
        this.deliveryMode = deliveryMode;
    }

    @Override
    public boolean getJMSRedelivered() throws JMSException {
        return redelivered;
    }

    @Override
    public void setJMSRedelivered(boolean redelivered) throws JMSException {
        this.redelivered = redelivered;
    }

    @Override
    public String getJMSType() throws JMSException {
        return type;
    }

    @Override
    public void setJMSType(String type) throws JMSException {
        this.type = type;
    }

    @Override
    public long getJMSExpiration() throws JMSException {
        return expiration;
    }

    @Override
    public void setJMSExpiration(long expiration) throws JMSException {
        this.expiration = expiration;
    }

    @Override
    public int getJMSPriority() throws JMSException {
        return priority;
    }

    @Override
    public void setJMSPriority(int priority) throws JMSException {
        this.priority = priority;
    }
    
    /**
     * Clears a message's properties and set the write permissions for
     * properties. The message's header fields and body are not cleared.
     */
    @Override
    public void clearProperties() throws JMSException {
        properties.clear();
        writePermissionsForProperties = true;
    }
    
    /**
     * Indicates whether a property value exists for the given property name.
     * 
     * @param name
     *            The name of the property.
     * @return true if the property exists.
     */
    @Override
    public boolean propertyExists(String name) throws JMSException {
        return properties.containsKey(name);
    }

    /**
     * Get the value for a property that represents a java primitive(e.g. int or
     * long).
     * 
     * @param property
     *            The name of the property to get.
     * @param type
     *            The type of the property.
     * @return the converted value for the property.
     * @throws JMSException
     *             On internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             and NumberFormatException when property name or value is
     *             null. Method throws same exception as primitives
     *             corresponding valueOf(String) method.
     */
    <T> T getPrimitiveProperty(String property, Class<T> type) throws JMSException {
        if (property == null) {
            throw new NullPointerException("Property name is null");
        }
        Object value = getObjectProperty(property);
        if (value == null) {
            return handleNullPropertyValue(property, type);
        }
        T convertedValue = TypeConversionSupport.convert(value, type);
        if (convertedValue == null) {
            throw new MessageFormatException("Property " + property + " was " + value.getClass().getName() +
                                             " and cannot be read as " + type.getName());
        }
        return convertedValue;
    }

    @SuppressWarnings("unchecked")
    private <T> T handleNullPropertyValue(String name, Class<T> clazz) {
        if (clazz == String.class) {
            return null;
        } else if (clazz == Boolean.class) {
            return (T) Boolean.FALSE;
        } else if (clazz == Double.class || clazz == Float.class) {
            throw new NullPointerException("Value of property with name " + name + " is null.");
        } else {
            throw new NumberFormatException("Value of property with name " + name + " is null.");
        }
    }
    
    /**
     * Returns the value of the <code>boolean</code> property with the specified
     * name.
     * 
     * @param name
     *            The name of the property to get.
     * @return the <code>boolean</code> property value for the specified name.
     * @throws JMSException
     *             On internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             When property name is null.
     */
    @Override
    public boolean getBooleanProperty(String name) throws JMSException {
        return getPrimitiveProperty(name, Boolean.class);
    }
    
    /**
     * Returns the value of the <code>byte</code> property with the specified
     * name.
     * 
     * @param name
     *            The name of the property to get.
     * @return the <code>byte</code> property value for the specified name.
     * @throws JMSException
     *             On internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             When property name is null.
     * @throws NumberFormatException
     *             When property value is null.
     */
    @Override
    public byte getByteProperty(String name) throws JMSException {
        return getPrimitiveProperty(name, Byte.class);
    }
    
    /**
     * Returns the value of the <code>short</code> property with the specified
     * name.
     * 
     * @param name
     *            The name of the property to get.
     * @return the <code>short</code> property value for the specified name.
     * @throws JMSException
     *             On internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             When property name is null.
     * @throws NumberFormatException
     *             When property value is null.
     */
    @Override
    public short getShortProperty(String name) throws JMSException {
        return getPrimitiveProperty(name, Short.class);
    }
    
    /**
     * Returns the value of the <code>int</code> property with the specified
     * name.
     * 
     * @param name
     *            The name of the property to get.
     * @return the <code>int</code> property value for the specified name.
     * @throws JMSException
     *             On internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             When property name is null.
     * @throws NumberFormatException
     *             When property value is null.
     */
    @Override
    public int getIntProperty(String name) throws JMSException {
        return getPrimitiveProperty(name, Integer.class);
    }
    
    /**
     * Returns the value of the <code>long</code> property with the specified
     * name.
     * 
     * @param name
     *            The name of the property to get.
     * @return the <code>long</code> property value for the specified name.
     * @throws JMSException
     *             On internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             When property name is null.
     * @throws NumberFormatException
     *             When property value is null.
     */
    @Override
    public long getLongProperty(String name) throws JMSException {
        return getPrimitiveProperty(name, Long.class);
    }
    
    /**
     * Returns the value of the <code>float</code> property with the specified
     * name.
     * 
     * @param name
     *            The name of the property to get.
     * @return the <code>float</code> property value for the specified name.
     * @throws JMSException
     *             Wn internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             When property name or value is null.
     */
    @Override
    public float getFloatProperty(String name) throws JMSException {
        return getPrimitiveProperty(name, Float.class);
    }
    
    /**
     * Returns the value of the <code>double</code> property with the specified
     * name.
     * 
     * @param name
     *            The name of the property to get.
     * @return the <code>double</code> property value for the specified name.
     * @throws JMSException
     *             On internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             When property name or value is null.
     */
    @Override
    public double getDoubleProperty(String name) throws JMSException {
        return getPrimitiveProperty(name, Double.class);
    }
    
    /**
     * Returns the value of the <code>String</code> property with the specified
     * name.
     * 
     * @param name
     *            The name of the property to get.
     * @return the <code>String</code> property value for the specified name.
     * @throws JMSException
     *             On internal error.
     * @throws MessageFormatException
     *             If the property cannot be converted to the specified type.
     * @throws NullPointerException
     *             When property name is null.
     */
    @Override
    public String getStringProperty(String name) throws JMSException {
        return getPrimitiveProperty(name, String.class);
    }
    
    /**
     * Returns the value of the Java object property with the specified name.
     * <P>
     * This method can be used to return, in boxed format, an object that has
     * been stored as a property in the message with the equivalent
     * <code>setObjectProperty</code> method call, or its equivalent primitive
     * setter method.
     * 
     * @param name
     *            The name of the property to get.
     * @return the Java object property value with the specified name, in boxed
     *         format (for example, if the property was set as an
     *         <code>int</code>, an <code>Integer</code> is returned); if there
     *         is no property by this name, a null value is returned.
     * @throws JMSException
     *             On internal error.
     */
    @Override
    public Object getObjectProperty(String name) throws JMSException {
        JMSMessagePropertyValue propertyValue = getJMSMessagePropertyValue(name);
        if (propertyValue != null) {
            return propertyValue.getValue();
        }
        return null;
    }
    
    /**
     * Returns the property value with message attribute to object property
     * conversions took place.
     * <P>
     * 
     * @param name
     *            The name of the property to get.
     * @return <code>JMSMessagePropertyValue</code> with object value and
     *         corresponding SQS message attribute type and message attribute
     *         string value.
     * @throws JMSException
     *             On internal error.
     */
    public JMSMessagePropertyValue getJMSMessagePropertyValue(String name) throws JMSException {
        return properties.get(name);
    }

    private static class PropertyEnum implements Enumeration<String> {
        private final Iterator<String> propertyItr;

        public PropertyEnum(Iterator<String> propertyItr) {
            this.propertyItr = propertyItr;
        }

        @Override
        public boolean hasMoreElements() {
            return propertyItr.hasNext();
        }

        @Override
        public String nextElement() {
            return propertyItr.next();
        }
    }
    
    /**
     * Returns an <code>Enumeration</code> of all the property names.
     * <P>
     * Note that JMS standard header fields are not considered properties and
     * are not returned in this enumeration.
     * 
     * @return an enumeration of all the names of property values.
     * @throws JMSException
     *             On internal error.
     */
    @Override
    public Enumeration<String> getPropertyNames() throws JMSException {
        return new PropertyEnum(properties.keySet().iterator());
    }
    
    /**
     * Sets a <code>boolean</code> property value with the specified name into
     * the message.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The <code>boolean</code> value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setBooleanProperty(String name, boolean value) throws JMSException {
        setObjectProperty(name, value);
    }
    
    /**
     * Sets a <code>byte</code> property value with the specified name into
     * the message.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The <code>byte</code> value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setByteProperty(String name, byte value) throws JMSException {
        setObjectProperty(name, value);
    }
    
    /**
     * Sets a <code>short</code> property value with the specified name into
     * the message.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The <code>short</code> value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setShortProperty(String name, short value) throws JMSException {
        setObjectProperty(name, value);
    }
    
    /**
     * Sets a <code>int</code> property value with the specified name into
     * the message.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The <code>int</code> value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setIntProperty(String name, int value) throws JMSException {
        setObjectProperty(name, value);
    }
    
    /**
     * Sets a <code>long</code> property value with the specified name into
     * the message.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The <code>long</code> value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setLongProperty(String name, long value) throws JMSException {
        setObjectProperty(name, value);
    }
    
    /**
     * Sets a <code>float</code> property value with the specified name into
     * the message.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The <code>float</code> value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setFloatProperty(String name, float value) throws JMSException {
        setObjectProperty(name, value);
    }
    
    /**
     * Sets a <code>double</code> property value with the specified name into
     * the message.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The <code>double</code> value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setDoubleProperty(String name, double value) throws JMSException {
        setObjectProperty(name, value);
    }
    
    /**
     * Sets a <code>String</code> property value with the specified name into
     * the message.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The <code>String</code> value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setStringProperty(String name, String value) throws JMSException {
        setObjectProperty(name, value);
    }
    
    /**
     * Sets a Java object property value with the specified name into the
     * message.
     * <P>
     * Note that this method works only for the boxed primitive object types
     * (Integer, Double, Long ...) and String objects.
     * 
     * @param name
     *            The name of the property to set.
     * @param value
     *            The object value of the property to set.
     * @throws JMSException
     *             On internal error.
     * @throws IllegalArgumentException
     *             If the name or value is null or empty string.
     * @throws MessageFormatException
     *             If the object is invalid type.
     * @throws MessageNotWriteableException
     *             If properties are read-only.
     */
    @Override
    public void setObjectProperty(String name, Object value) throws JMSException {
        if (name == null || name.isEmpty()) {
            throw new IllegalArgumentException("Property name can not be null or empty.");
        }
        if (value == null || "".equals(value)) {
            throw new IllegalArgumentException("Property value can not be null or empty.");
        }
        if(!isValidPropertyValueType(value)) {
            throw new MessageFormatException("Value of property with name " + name + " has incorrect type " + value.getClass().getName() + ".");
        }
        checkPropertyWritePermissions();
        properties.put(name, new JMSMessagePropertyValue(value));
    }
    
    /**
     * <P>
     * Acknowledges message(s).
     * <P>
     * A client may individually acknowledge each message as it is consumed, or
     * it may choose to acknowledge multiple messages based on acknowledge mode,
     * which in turn might might acknowledge all messages consumed by the
     * session.
     * <P>
     * Messages that have been received but not acknowledged may be redelivered.
     * <P>
     * If the session is closed, messages cannot be acknowledged.
     * <P>
     * If only the consumer is closed, messages can still be acknowledged.
     * 
     * @see com.amazon.sqs.javamessaging.acknowledge.AcknowledgeMode
     * @throws JMSException
     *             On Internal error
     * @throws IllegalStateException
     *             If this method is called on a closed session.
     */
    @Override
    public void acknowledge() throws JMSException {
        if (acknowledger != null) {
            acknowledger.acknowledge(this);
        }
    }
    
    /**
     * <P>
     * Clears out the message body. Clearing a message's body does not clear its
     * header values or property entries.
     * <P>
     * This method cannot be called directly instead the implementation on the
     * subclasses should be used.
     * 
     * @throws JMSException
     *             If directly called
     */
    @Override
    public void clearBody() throws JMSException {
        throw new JMSException("SQSMessage does not have any body");
    }

    private boolean isValidPropertyValueType(Object value) {
        return value instanceof Boolean || value instanceof Byte || value instanceof Short ||
               value instanceof Integer || value instanceof Long || value instanceof Float ||
               value instanceof Double || value instanceof String;
    }

    /**
     * Copied from org.apache.activemq.util.TypeConversionSupport to provide the
     * same property support provided by activemq without creating a dependency
     * on activemq.
     */
    public static class TypeConversionSupport {

        static class ConversionKey {
            final Class<?> from;

            final Class<?> to;

            public ConversionKey(Class<?> from, Class<?> to) {
                this.from = from;
                this.to = to;
            }
            
            @Override
            public int hashCode() {
                final int prime = 31;
                int result = 1;
                result = prime * result + ((from == null) ? 0 : from.hashCode());
                result = prime * result + ((to == null) ? 0 : to.hashCode());
                return result;
            }

            @Override
            public boolean equals(Object obj) {
                if (this == obj)
                    return true;
                if (obj == null)
                    return false;
                if (getClass() != obj.getClass())
                    return false;
                ConversionKey other = (ConversionKey) obj;
                if (from == null) {
                    if (other.from != null)
                        return false;
                } else if (!from.equals(other.from))
                    return false;
                if (to == null) {
                    if (other.to != null)
                        return false;
                } else if (!to.equals(other.to))
                    return false;
                return true;
            }
        }

        interface Converter {
            Object convert(Object value);
        }

        static final private Map<ConversionKey, Converter> CONVERSION_MAP = new HashMap<ConversionKey, Converter>();
        static {
            Converter toStringConverter = new Converter() {
                public Object convert(Object value) {
                    return value.toString();
                }
            };
            CONVERSION_MAP.put(new ConversionKey(Boolean.class, String.class), toStringConverter);
            CONVERSION_MAP.put(new ConversionKey(Byte.class, String.class), toStringConverter);
            CONVERSION_MAP.put(new ConversionKey(Short.class, String.class), toStringConverter);
            CONVERSION_MAP.put(new ConversionKey(Integer.class, String.class), toStringConverter);
            CONVERSION_MAP.put(new ConversionKey(Long.class, String.class), toStringConverter);
            CONVERSION_MAP.put(new ConversionKey(Float.class, String.class), toStringConverter);
            CONVERSION_MAP.put(new ConversionKey(Double.class, String.class), toStringConverter);

            CONVERSION_MAP.put(new ConversionKey(String.class, Boolean.class), new Converter() {
                public Object convert(Object value) {
                    String stringValue = (String) value;
                    if (Boolean.valueOf(stringValue) || INT_TRUE.equals((String) value)) {
                        return Boolean.TRUE;
                    }
                    return Boolean.FALSE;
                }
            });
            CONVERSION_MAP.put(new ConversionKey(String.class, Byte.class), new Converter() {
                public Object convert(Object value) {
                    return Byte.valueOf((String) value);
                }
            });
            CONVERSION_MAP.put(new ConversionKey(String.class, Short.class), new Converter() {
                public Object convert(Object value) {
                    return Short.valueOf((String) value);
                }
            });
            CONVERSION_MAP.put(new ConversionKey(String.class, Integer.class), new Converter() {
                public Object convert(Object value) {
                    return Integer.valueOf((String) value);
                }
            });
            CONVERSION_MAP.put(new ConversionKey(String.class, Long.class), new Converter() {
                public Object convert(Object value) {
                    return Long.valueOf((String) value);
                }
            });
            CONVERSION_MAP.put(new ConversionKey(String.class, Float.class), new Converter() {
                public Object convert(Object value) {
                    return Float.valueOf((String) value);
                }
            });
            CONVERSION_MAP.put(new ConversionKey(String.class, Double.class), new Converter() {
                public Object convert(Object value) {
                    return Double.valueOf((String) value);
                }
            });

            Converter longConverter = new Converter() {
                public Object convert(Object value) {
                    return Long.valueOf(((Number) value).longValue());
                }
            };
            CONVERSION_MAP.put(new ConversionKey(Byte.class, Long.class), longConverter);
            CONVERSION_MAP.put(new ConversionKey(Short.class, Long.class), longConverter);
            CONVERSION_MAP.put(new ConversionKey(Integer.class, Long.class), longConverter);
            CONVERSION_MAP.put(new ConversionKey(Date.class, Long.class), new Converter() {
                public Object convert(Object value) {
                    return Long.valueOf(((Date) value).getTime());
                }
            });

            Converter intConverter = new Converter() {
                public Object convert(Object value) {
                    return Integer.valueOf(((Number) value).intValue());
                }
            };
            CONVERSION_MAP.put(new ConversionKey(Byte.class, Integer.class), intConverter);
            CONVERSION_MAP.put(new ConversionKey(Short.class, Integer.class), intConverter);

            CONVERSION_MAP.put(new ConversionKey(Byte.class, Short.class), new Converter() {
                public Object convert(Object value) {
                    return Short.valueOf(((Number) value).shortValue());
                }
            });

            CONVERSION_MAP.put(new ConversionKey(Float.class, Double.class), new Converter() {
                public Object convert(Object value) {
                    return Double.valueOf(((Number) value).doubleValue());
                }
            });
        }

        @SuppressWarnings("unchecked")
        static public <T> T convert(Object value, Class<T> clazz) {

            assert value != null && clazz != null;

            if (value.getClass() == clazz)
                return (T) value;

            Converter c = (Converter) CONVERSION_MAP.get(new ConversionKey(value.getClass(), clazz));
            if (c == null)
                return null;
            return (T) c.convert(value);

        }
    }
    
    /**
     * This class is used fulfill object value, corresponding SQS message
     * attribute type and message attribute string value.
     */
    public static class JMSMessagePropertyValue {
        
        private final Object value;

        private final String type;
        
        private final String stringMessageAttributeValue;
        
        public JMSMessagePropertyValue(String stringValue, String type) throws JMSException{
            this.type = type;
            this.value = getObjectValue(stringValue, type);
            this.stringMessageAttributeValue = stringValue;
        }
        
        public JMSMessagePropertyValue(Object value) throws JMSException {
            this.type = getType(value);
            this.value = value;
            if (BOOLEAN.equals(type)) {
                if((Boolean) value) {
                    stringMessageAttributeValue = INT_TRUE;
                } else {
                    stringMessageAttributeValue = INT_FALSE;
                }
            } else {
                stringMessageAttributeValue = value.toString();
            }
        }
        
        public JMSMessagePropertyValue(Object value, String type) throws JMSException {
            this.value = value;
            this.type = type;
            if (BOOLEAN.equals(type)) {
                if((Boolean) value) {
                    stringMessageAttributeValue = INT_TRUE;
                } else {
                    stringMessageAttributeValue = INT_FALSE;
                }
            } else {
                stringMessageAttributeValue = value.toString();
            }
        }
        
        private static String getType(Object value) throws JMSException {
            if (value instanceof String) {
                return STRING;
            } else if (value instanceof Integer) {
                return INT;
            } else if (value instanceof Long) {
                return LONG;
            } else if (value instanceof Boolean) {
                return BOOLEAN;
            } else if (value instanceof Byte) {
                return BYTE;
            } else if (value instanceof Double) {
                return DOUBLE;
            } else if (value instanceof Float) {
                return FLOAT;
            } else if (value instanceof Short) {
                return SHORT;
            } else {
                throw new JMSException("Not a supported JMS property type");
            }
        }

        private static Object getObjectValue(String value, String type) throws JMSException {
            if (INT.equals(type)) {
                return Integer.valueOf(value);
            } else if (LONG.equals(type)) {
                return Long.valueOf(value);
            } else if (BOOLEAN.equals(type)) {
                if (INT_TRUE.equals(value)) {
                    return Boolean.TRUE;
                }
                return Boolean.FALSE;
            } else if (BYTE.equals(type)) {
                return Byte.valueOf(value);
            } else if (DOUBLE.equals(type)) {
                return Double.valueOf(value);
            } else if (FLOAT.equals(type)) {
                return Float.valueOf(value);
            } else if (SHORT.equals(type)) {
                return Short.valueOf(value);
            } else if (type != null && (type.startsWith(STRING) || type.startsWith(NUMBER))) {
                return value;
            } else {
                throw new JMSException(type + " is not a supported JMS property type");
            }
        }        

        public String getType() {
            return type;
        }

        public Object getValue() {
            return value;
        }
        
        public String getStringMessageAttributeValue() {
            return stringMessageAttributeValue;
        }
    }


    /**
     * This method sets the JMS_SQS_SEQUENCE_NUMBER property on the message. It is exposed explicitly here, so that
     * it can be invoked even on read-only message object obtained through receing a message. 
     * This support the use case of send a received message by using the same JMSMessage object.
     * 
     * @param sequenceNumber Sequence number to set. If null or empty, the stored sequence number will be removed.
     * @throws JMSException
     */
    public void setSequenceNumber(String sequenceNumber) throws JMSException {
        if (sequenceNumber == null || sequenceNumber.isEmpty()) {
            properties.remove(SQSMessagingClientConstants.JMS_SQS_SEQUENCE_NUMBER);
        } else {
            properties.put(SQSMessagingClientConstants.JMS_SQS_SEQUENCE_NUMBER, new JMSMessagePropertyValue(sequenceNumber));
        }
    }

    /*
     * Unit Test Utility Functions
     */
    void setWritePermissionsForProperties(boolean writePermissionsForProperties) {
        this.writePermissionsForProperties = writePermissionsForProperties;
    }
} public class SQSObjectMessage extends SQSMessage implements ObjectMessage {
    private static final Log LOG = LogFactory.getLog(SQSObjectMessage.class);

    /**
     * Serialized message body
     */
    private String body;

    /**
     * Convert received SQSMessage into ObjectMessage
     */
    public SQSObjectMessage(Acknowledger acknowledger, String queueUrl, Message sqsMessage) throws JMSException {
        super(acknowledger, queueUrl, sqsMessage);
        body = sqsMessage.getBody();
    }
    
    /**
     * Create new empty ObjectMessage to send.
     */
    public SQSObjectMessage() throws JMSException {
        super();
    }

    /**
     * Create new ObjectMessage with payload to send.
     */
    public SQSObjectMessage(Serializable payload) throws JMSException {
        super();
        body = serialize(payload);
    }
    
    /**
     * Sets the <code>Serializable</code> containing this message's body
     * 
     * @param payload
     *            The <code>Serializable</code> containing the message's body
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     * @throws MessageFormatException
     *             If object serialization fails.
     */
    @Override
    public void setObject(Serializable payload) throws JMSException {
        checkBodyWritePermissions();
        body = serialize(payload);
    }
    
    /**
     * Gets the <code>Serializable</code> containing this message's body
     * 
     * @throws MessageFormatException
     *             If object deserialization fails.
     */
    @Override
    public Serializable getObject() throws JMSException {
        return deserialize(body);
    }
    
    /**
     * Sets the message body to write mode, and sets the object body to null
     */
    @Override
    public void clearBody() throws JMSException {
        body = null;
        setBodyWritePermissions(true);
    }
    
    /**
     * Deserialize the <code>String</code> into <code>Serializable</code>
     * object.
     */
    protected static Serializable deserialize(String serialized) throws JMSException {
        if (serialized == null) {
            return null;
        }
        Serializable deserializedObject;
        ObjectInputStream objectInputStream = null;
        try {
            byte[] bytes = Base64.decode(serialized);
            objectInputStream = new ObjectInputStream(new ByteArrayInputStream(bytes));
            deserializedObject = (Serializable) objectInputStream.readObject();
        } catch (IOException e) {
            LOG.error("IOException: Message cannot be written", e);
            throw convertExceptionToMessageFormatException(e);
        } catch (Exception e) {
            LOG.error("Unexpected exception: ", e);
            throw convertExceptionToMessageFormatException(e);
        } finally {
            if (objectInputStream != null) {
                try {
                    objectInputStream.close();
                } catch (IOException e) {
                    LOG.warn(e.getMessage());
                }
            }
        }
        return deserializedObject;
    }
    
    /**
     * Serialize the <code>Serializable</code> object to <code>String</code>.
     */
    protected static String serialize(Serializable serializable) throws JMSException {
        if (serializable == null) {
            return null;
        }
        String serializedString;
        ObjectOutputStream objectOutputStream = null;
        try {
            ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
            objectOutputStream = new ObjectOutputStream(bytesOut);
            objectOutputStream.writeObject(serializable);
            objectOutputStream.flush();
            serializedString = Base64.encodeAsString(bytesOut.toByteArray());
        } catch (IOException e) {
            LOG.error("IOException: cannot serialize objectMessage", e);
            throw convertExceptionToMessageFormatException(e);
        } finally {
            if (objectOutputStream != null) {
                try {
                    objectOutputStream.close();
                } catch (IOException e) {
                    LOG.warn(e.getMessage());
                }
            }
        }
        return serializedString;
    }
    
    public String getMessageBody() {
        return body;
    }

    void setMessageBody(String body) {
        this.body = body;
    }
} public class SQSTextMessage extends SQSMessage implements TextMessage {

    /**
     * Text of the message. Assume this is safe from SQS invalid characters.
     */
    private String text;

    /**
     * Convert received SQSMessage into TextMessage.
     */
    public SQSTextMessage(Acknowledger acknowledger, String queueUrl, Message sqsMessage) throws JMSException{
        super(acknowledger, queueUrl, sqsMessage);
        this.text = sqsMessage.getBody();
    }
    
    /**
     * Create new empty TextMessage to send.
     */
    public SQSTextMessage() throws JMSException {
        super();
    }

    /**
     * Create new TextMessage with payload to send.
     */
    public SQSTextMessage(String payload) throws JMSException {
        super();
        this.text = payload;
    }
     
    /**
     * Sets the text containing this message's body.
     * 
     * @param string
     *            The <code>String</code> containing the message's body
     * @throws MessageNotWriteableException
     *             If the message is in read-only mode.
     */
    @Override
    public void setText(String string) throws JMSException {
        checkBodyWritePermissions();
        this.text = string;
    }
    
    /**
     * Gets the text containing this message's body.
     * 
     * @return The <code>String</code> containing the message's body
     */
    @Override
    public String getText() throws JMSException {
        return text;
    }
    
    /**
     * Sets the message body to write mode, and sets the text to null
     */
    @Override
    public void clearBody() throws JMSException {
        text = null;
        setBodyWritePermissions(true);
    }
}
import com.amazon.sqs.javamessaging.AmazonSQSMessagingClientWrapper;
import com.amazon.sqs.javamessaging.SQSConnection;
import com.amazon.sqs.javamessaging.SQSMessageProducer;
import com.amazon.sqs.javamessaging.SQSQueueDestination;
import com.amazon.sqs.javamessaging.SQSSession;
import com.amazon.sqs.javamessaging.acknowledge.Acknowledger;
import com.amazon.sqs.javamessaging.message.SQSBytesMessage;
import com.amazon.sqs.javamessaging.message.SQSMessage;
import com.amazon.sqs.javamessaging.message.SQSObjectMessage;
import com.amazon.sqs.javamessaging.message.SQSTextMessage;
import com.amazonaws.services.sqs.model.DeleteMessageRequest;
import com.amazonaws.services.sqs.model.MessageAttributeValue;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageResult;
import com.amazonaws.util.Base64;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.IllegalStateException;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.Destination;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatcher;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.argThat;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/*
 * Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 * or in the "license" file accompanying this file. This file is distributed
 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 * express or implied. See the License for the specific language governing
 * permissions and limitations under the License.
 */
package com.amazon.sqs.javamessaging;







/**
 * Test the SQSMessageProducerTest class
 */
public class SQSMessageProducerTest {

    public static final String QUEUE_URL = "QueueUrl";
    public static final String QUEUE_NAME = "QueueName";
    public static final String MESSAGE_ID_1 = "MessageId1";
    public static final String MESSAGE_ID_2 = "MessageId2";

    private SQSMessageProducer producer;
    private SQSQueueDestination destination;
    private SQSSession sqsSession;
    private SQSConnection sqsConnection;