Fix session closing sequence

The refactoring in #606 broke the closing sequence for transacted
sessions, as a closed flag was set to true on closing and a method
called in the closing checks the flag.

This commit adds a closing flag to make the closing idempotent and sets
the closed flag to true when the session is closed.

References #606

(cherry picked from commit 3a227a8)

Author: acogoluegnes

src/main/java/com/rabbitmq/jms/client/RMQSession.java

@@ -138,6 +138,7 @@ public class RMQSession implements Session, QueueSession, TopicSession {
 
     /** Set to true if close() has been called and completed */
     private final AtomicBoolean closed = new AtomicBoolean(false);
+    private final AtomicBoolean closing = new AtomicBoolean(false);
 
     /** The message listener for this session. */
     private volatile MessageListener messageListener;
@@ -575,33 +576,37 @@ public void close() throws JMSException {
     }
 
     void internalClose() throws JMSException {
-        if (this.closed.compareAndSet(false, true)) {
-            logger.trace("close session {}", this);
-            // close consumers first (to prevent requeues being consumed)
-            closeAllConsumers();
-
-            // rollback anything not committed already
-            if (this.getTransactedNoException()) {
-                // don't nack messages on close
-                this.clearUncommittedTags();
-                this.rollback();
-            }
+        if (this.closing.compareAndSet(false, true)) {
+            try {
+                logger.trace("close session {}", this);
+                // close consumers first (to prevent requeues being consumed)
+                closeAllConsumers();
 
-            //clear up potential executor
-            this.deliveryExecutor.close();
+                // rollback anything not committed already
+                if (this.getTransactedNoException()) {
+                    // don't nack messages on close
+                    this.clearUncommittedTags();
+                    this.rollback();
+                }
 
-            //close all producers created by this session
-            for (RMQMessageProducer producer : this.producers) {
-                producer.internalClose();
-            }
-            this.producers.clear();
+                //clear up potential executor
+                this.deliveryExecutor.close();
 
-            //now commit anything done during close
-            if (this.getTransactedNoException()) {
-                this.commit();
-            }
+                //close all producers created by this session
+                for (RMQMessageProducer producer : this.producers) {
+                    producer.internalClose();
+                }
+                this.producers.clear();
+
+                //now commit anything done during close
+                if (this.getTransactedNoException()) {
+                    this.commit();
+                }
 
-            this.closeRabbitChannels();
+                this.closeRabbitChannels();
+            } finally {
+                this.closed.set(true);
+            }
         }
     }