'Version 1.0.0 of the Amazon SQS Java Messaging Library'

Author: Patel, Mihir

NOTICE.txt

@@ -8,6 +8,6 @@ Amazon Technologies, Inc (http://www.amazon.com/).
 THIRD PARTY COMPONENTS
 **********************
 This software includes third party software subject to the following copyrights:
-- Apache ActiveMQ (see: org.apache.activemq.util.TypeConversionSupport), Copyright 2005-2013 Apache Software Foundation.
+- Apache ActiveMQ (see: org.apache.activemq.util.TypeConversionSupport, org.apache.activemq.command.ActiveMQStreamMessage). Copyright 2005-2013 Apache Software Foundation
 
-The licenses for these third party components are included in LICENSE.txt
+The licenses for these third party components are included in LICENSE.txt

README.md

@@ -1,7 +1,7 @@
 Amazon SQS Java Messaging Library
 ========================================
 The **Amazon SQS Java Messaging Library** holds the Java Message Service compatible classes, that are used
-for communicating with Amazon Simple Queue Service. This project builds on top of the AWS SDK for Java to use Amazon SQS as the JMS (as defined in 1.1 specification) provider for the messaging applications without running any additional software.
+for communicating with Amazon Simple Queue Service. This project builds on top of the AWS SDK for Java to use Amazon SQS as the JMS (as defined in 1.1 specification) provider for the messaging applications without running any additional software. 
 
 * You can download release builds through the [releases section of this](https://github.com/awslabs/amazon-sqs-java-messaging-lib) project.
 * For more information on using the amazon-sqs-java-messaging-lib, see our getting started guide to SQS [here](http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSGettingStartedGuide/Welcome.html).
@@ -11,7 +11,7 @@ for communicating with Amazon Simple Queue Service. This project builds on top o
 * **Sign up for AWS** — Before you begin, you need an AWS account. For more information about creating an AWS account and retrieving your AWS credentials, see [AWS Account and Credentials](http://docs.aws.amazon.com/AWSSdkDocsJava/latest/DeveloperGuide/java-dg-setup.html) in the AWS SDK for Java Developer Guide.
 * **Sign up for Amazon SQS** — Go to the Amazon [SQS console](https://console.aws.amazon.com/sqs/home?region=us-east-1) to sign up for the service.
 * **Minimum requirements** — To use the sample application, you'll need Java 1.7+ and [Maven 3](http://maven.apache.org/). For more information about the requirements, see the [Getting Started](http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSGettingStartedGuide/Welcome.html) section of the Amazon SQS Developer Guide.
-* **Download** — Download the [latest preview release](https://github.com/awslabs/amazon-sqs-java-messaging-lib/releases) or pick it up from Maven:
+* **Download** — Download the [latest preview release or](https://github.com/awslabs/amazon-sqs-java-messaging-lib/releases) pick it up from Maven:
 ```xml
   <dependency>
     <groupId>com.amazonaws</groupId>
@@ -25,3 +25,4 @@ for communicating with Amazon Simple Queue Service. This project builds on top o
 ##Feedback
 * Give us feedback [here](https://github.com/awslabs/amazon-sqs-java-messaging-lib/issues).
 * If you'd like to contribute a new feature or bug fix, we'd love to see Github pull requests from you.
+

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

@@ -25,4 +25,6 @@ public interface PrefetchManager {
      * called.
      */
     public void messageDispatched();
+    
+    public SQSMessageConsumer getMessageConsumer();
 }

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

@@ -194,11 +194,13 @@ public void start() throws JMSException {
     @Override
     public void stop() throws JMSException {
         checkClosed();
-        actionOnConnectionTaken = true;
+                
         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");

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

@@ -60,9 +60,6 @@ public class SQSMessageConsumer implements MessageConsumer, QueueReceiver {
      */
     private final SQSMessageConsumerPrefetch sqsMessageConsumerPrefetch;
 
-    /** Used to synchronize on calls from message listener on its own consumer */
-    private final Object callBackSynchronizer;
-    
     SQSMessageConsumer(SQSConnection parentSQSConnection, SQSSession parentSQSSession,
                        SQSSessionCallbackScheduler sqsSessionRunnable, SQSDestination destination,
                        Acknowledger acknowledger, NegativeAcknowledger negativeAcknowledger, ThreadFactory threadFactory) {
@@ -84,12 +81,11 @@ public class SQSMessageConsumer implements MessageConsumer, QueueReceiver {
         this.acknowledger = acknowledger;
         this.sqsSessionRunnable = sqsSessionRunnable;
         this.sqsMessageConsumerPrefetch = sqsMessageConsumerPrefetch;
+        this.sqsMessageConsumerPrefetch.setMessageConsumer(this);
         this.negativeAcknowledger = negativeAcknowledger;
 
         prefetchExecutor = Executors.newSingleThreadExecutor(threadFactory);
         prefetchExecutor.execute(sqsMessageConsumerPrefetch);
-
-        callBackSynchronizer = sqsSessionRunnable.getSynchronizer();
     }
 
 
@@ -139,15 +135,13 @@ public void close() throws JMSException {
         if (closed) {
             return;
         }
-        synchronized (callBackSynchronizer) {
-            if (Thread.currentThread() == sqsSessionRunnable.getCurrentThread()) {
-                sqsSessionRunnable.setConsumerCloseAfterCallback(this);
-                return;
-            }
-
-            doClose();
-            callBackSynchronizer.notifyAll();
+        
+        if (parentSQSSession.isActiveCallbackSessionThread()) {
+            sqsSessionRunnable.setConsumerCloseAfterCallback(this);
+            return;
         }
+        
+        doClose();
     }
 
     void doClose() {
@@ -160,21 +154,24 @@ void doClose() {
         parentSQSSession.removeConsumer(this);
 
         try {
-            LOG.info("Shutting down " + SQSSession.CONSUMER_PREFETCH_EXECUTER_NAME + " executor");
-
-            /** Shut down executor. */
-            prefetchExecutor.shutdown();
+            if (!prefetchExecutor.isShutdown()) {
+                LOG.info("Shutting down " + SQSSession.CONSUMER_PREFETCH_EXECUTER_NAME + " executor");
+                /** Shut down executor. */
+                prefetchExecutor.shutdown();
+            }
+            
+            parentSQSSession.waitForConsumerCallbackToComplete(this);
 
             if (!prefetchExecutor.awaitTermination(PREFETCH_EXECUTOR_GRACEFUL_SHUTDOWN_TIME, TimeUnit.SECONDS)) {
 
-                LOG.warn("Can't terminate executor service " +
-                         SQSSession.CONSUMER_PREFETCH_EXECUTER_NAME + " after " + 60 +
-                         " seconds, some running threads will be shutdown immediately");
+                LOG.warn("Can't terminate executor service " + SQSSession.CONSUMER_PREFETCH_EXECUTER_NAME +
+                         " after " + 60 + " seconds, some running threads will be shutdown immediately");
                 prefetchExecutor.shutdownNow();
             }
         } catch (InterruptedException e) {
             LOG.error("Interrupted while closing the consumer.", e);
         }
+
         closed = true;
     }
 
@@ -196,22 +193,16 @@ public String getMessageSelector() throws JMSException {
         throw new JMSException(SQSJMSClientConstants.UNSUPPORTED_METHOD);
     }
 
-    /** This call blocks until message listener in progress have completed. */
-    protected void stop() {
+    /** This stops the prefetching */
+    protected void stopPrefetch() {
         if (!closed) {
-            synchronized (callBackSynchronizer) {
-                sqsMessageConsumerPrefetch.stop();
-                callBackSynchronizer.notifyAll();
-            }
+            sqsMessageConsumerPrefetch.stop();
         }
     }
 
-    protected void start() {
+    protected void startPrefetch() {
         if (!closed) {
-            synchronized (callBackSynchronizer) {
-                sqsMessageConsumerPrefetch.start();
-                callBackSynchronizer.notifyAll();
-            }
+            sqsMessageConsumerPrefetch.start();
         }
     }
 

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

@@ -55,6 +55,8 @@ public class SQSMessageConsumerPrefetch implements Runnable, PrefetchManager {
     private final Acknowledger acknowledger;
     private final NegativeAcknowledger negativeAcknowledger;
     private volatile MessageListener messageListener;
+    
+    private SQSMessageConsumer messageConsumer;
 
     private final SQSSessionCallbackScheduler sqsSessionRunnable;
 
@@ -100,6 +102,15 @@ public class SQSMessageConsumerPrefetch implements Runnable, PrefetchManager {
     MessageListener getMessageListener() {
         return messageListener;
     }
+    
+    void setMessageConsumer(SQSMessageConsumer messageConsumer) {
+        this.messageConsumer = messageConsumer;
+    }
+    
+    @Override
+    public SQSMessageConsumer getMessageConsumer() {
+        return messageConsumer;
+    }
 
     protected void setMessageListener(MessageListener messageListener) {
         this.messageListener = messageListener;