Add checkClosed() to all three receive*() methods

As per #77,
closed consumers should only return null from concurrent in-flight
receive calls, not future ones, and returning null doesn't play nice
with SpringJMS. Checking in SQSMessageConsumer is more consistent than
in SQSMessageConsumerPrefetch, and easier to test to boot.

Author: robin-aws

src/main/java/com/amazon/sqs/javamessaging/SQSMessageConsumer.java

@@ -140,12 +140,13 @@ public void setMessageListener(MessageListener listener) throws JMSException {
      * paused.
      * 
      * @return the next message produced for this message consumer, or null if
-     *         this message consumer is closed
+     *         this message consumer is closed during the receive call
      * @throws JMSException
      *             On internal error
      */
     @Override
     public Message receive() throws JMSException {
+        checkClosed();
         return sqsMessageConsumerPrefetch.receive();
     }
 
@@ -157,12 +158,13 @@ public Message receive() throws JMSException {
      * @param timeout
      *            the timeout value (in milliseconds)
      * @return the next message produced for this message consumer, or null if
-     *         the timeout expires or this message consumer is closed
+     *         the timeout expires or this message consumer is closed during the receive call
      * @throws JMSException
      *             On internal error
      */
     @Override
     public Message receive(long timeout) throws JMSException {
+        checkClosed();
         return sqsMessageConsumerPrefetch.receive(timeout);
     }
 
@@ -176,6 +178,7 @@ public Message receive(long timeout) throws JMSException {
      */
     @Override
     public Message receiveNoWait() throws JMSException {
+        checkClosed();
         return sqsMessageConsumerPrefetch.receiveNoWait();
     }
 

src/main/java/com/amazon/sqs/javamessaging/SQSMessageConsumerPrefetch.java

@@ -590,11 +590,11 @@ private javax.jms.Message messageHandler(MessageManager messageManager) throws J
 
     private boolean cannotDeliver() throws JMSException {
         if (!running) {
-          return true;
+            return true;
         }
 
         if (isClosed()) {
-          throw new JMSException("Cannot receive messages when the consumer is closed");
+            throw new JMSException("Cannot receive messages when the consumer is closed");
         }
 
         if (messageListener != null) {

src/test/java/com/amazon/sqs/javamessaging/SQSMessageConsumerTest.java

@@ -347,6 +347,88 @@ public void testSetMessageListenerAlreadyClosed() throws InterruptedException, J
         }
     }
 
+    /**
+     * Test receive fails when consumer is already closed
+     */
+    @Test
+    public void testReceiveAlreadyClosed() throws InterruptedException, JMSException {
+
+        /*
+         * Set up consumer
+         */
+        consumer = spy(new SQSMessageConsumer(sqsConnection, sqsSession, sqsSessionRunnable,
+                destination, acknowledger, negativeAcknowledger, threadFactory, sqsMessageConsumerPrefetch));
+
+        consumer.close();
+
+        /*
+         * Call receive
+         */
+        try {
+            consumer.receive();
+            fail();
+        } catch (JMSException ex) {
+            assertEquals("Consumer is closed", ex.getMessage());
+        }
+
+
+    }
+
+    /**
+     * Test set message listener fails when consumer is already closed
+     */
+    @Test
+    public void testReceiveWithTimeoutAlreadyClosed() throws InterruptedException, JMSException {
+
+        /*
+         * Set up consumer
+         */
+        consumer = spy(new SQSMessageConsumer(sqsConnection, sqsSession, sqsSessionRunnable,
+                destination, acknowledger, negativeAcknowledger, threadFactory, sqsMessageConsumerPrefetch));
+
+        consumer.close();
+
+        long timeout = 10;
+
+        /*
+         * Call receive with timeout
+         */
+        try {
+            consumer.receive(timeout);
+            fail();
+        } catch (JMSException ex) {
+            assertEquals("Consumer is closed", ex.getMessage());
+        }
+
+
+    }
+
+    /**
+     * Test set message listener fails when consumer is already closed
+     */
+    @Test
+    public void testReceiveNoWaitAlreadyClosed() throws InterruptedException, JMSException {
+
+        /*
+         * Set up consumer
+         */
+        consumer = spy(new SQSMessageConsumer(sqsConnection, sqsSession, sqsSessionRunnable,
+                destination, acknowledger, negativeAcknowledger, threadFactory, sqsMessageConsumerPrefetch));
+
+        consumer.close();
+
+        /*
+         * Call receive no wait
+         */
+        try {
+
+            consumer.receiveNoWait();
+            fail();
+        } catch (JMSException ex) {
+            assertEquals("Consumer is closed", ex.getMessage());
+        }
+    }
+
     /**
      * Test set message listener
      */