Use concurrent utilities in RMQSession

Instead of synchronized blocks and Object monitors.

(cherry picked from commit 67b80b2)

Author: acogoluegnes

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

@@ -13,6 +13,8 @@
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiFunction;
 
 import javax.jms.BytesMessage;
@@ -133,8 +135,10 @@ public class RMQSession implements Session, QueueSession, TopicSession {
 
     /** The main RabbitMQ channel we use under the hood */
     private final Channel channel;
+
     /** Set to true if close() has been called and completed */
-    private volatile boolean closed = false;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+
     /** The message listener for this session. */
     private volatile MessageListener messageListener;
     /** A list of all the producers created by this session.
@@ -146,22 +150,18 @@ public class RMQSession implements Session, QueueSession, TopicSession {
     /** We keep an ordered set of the message tags (acknowledgement tags) for all messages received and unacknowledged.
      * Each message acknowledgement must ACK all (unacknowledged) messages received up to this point, and
      * we must never acknowledge a message more than once (nor acknowledge a message that doesn't exist). */
-    private final SortedSet<Long> unackedMessageTags = Collections.synchronizedSortedSet(
-        new TreeSet<>());
-
-    /* Holds the uncommited tags to commit a nack on rollback */
-    private final List<Long> uncommittedMessageTags = new ArrayList<>(); // GuardedBy("commitLock");
+    private final SortedSet<Long> unackedMessageTags =
+        Collections.synchronizedSortedSet(new TreeSet<>()); // GuardedBy("unackedMessageTagsLock")
+    private final Lock unackedMessageTagsLock = new ReentrantLock();
 
     /** List of all our topic subscriptions so we can track them */
-    private final Subscriptions subscriptions;
-
-    /** Lock for waiting for close */
-    private final Object closeLock = new Object();
+    private final Subscriptions subscriptions; // GuardedBy("subscriptionsLock")
+    private final Lock subscriptionsLock = new ReentrantLock();
 
-    /** Lock and parms for commit and rollback blocking of other commands */
-    private final Object commitLock = new Object();
-    private static final long COMMIT_WAIT_MAX = 2000L; // 2 seconds
-    private boolean committing = false; // GuardedBy("commitLock");
+    /* Holds the uncommited tags to commit a nack on rollback */
+    private final List<Long> uncommittedMessageTags = new ArrayList<>(); // GuardedBy("commitLock");
+    /** Lock commit and rollback blocking of other commands */
+    private final Lock commitLock = new ReentrantLock();
 
     /** Client version obtained from compiled class. */
     private static final GenericVersion CLIENT_VERSION = new GenericVersion(RMQSession.class.getPackage().getImplementationVersion());
@@ -182,14 +182,13 @@ public class RMQSession implements Session, QueueSession, TopicSession {
     private static final Map<String, Object> RJMS_SELECTOR_EXCHANGE_ARGS
         = Collections.singletonMap(RJMS_VERSION_ARG, RJMS_CLIENT_VERSION);
 
-
     private static final String JMS_TOPIC_SELECTOR_EXCHANGE_TYPE = "x-jms-topic";
 
     private final DeliveryExecutor deliveryExecutor;
 
     /** The channels we use for browsing queues (there may be more than one in operation at a time) */
-    private Set<Channel> browsingChannels = new HashSet<>(); // @GuardedBy(bcLock)
-    private final Object bcLock = new Object();
+    private final Set<Channel> browsingChannels = new HashSet<>(); // @GuardedBy(bcLock)
+    private final Lock bcLock = new ReentrantLock();
 
     /**
      * Classes in these packages can be transferred via ObjectMessage.
@@ -344,7 +343,7 @@ public BytesMessage createBytesMessage() throws JMSException {
     }
 
     private void illegalStateExceptionIfClosed() throws IllegalStateException {
-        if (this.closed) throw new IllegalStateException("Session is closed");
+        if (this.closed.get()) throw new IllegalStateException("Session is closed");
     }
 
     /**
@@ -470,25 +469,18 @@ int getAcknowledgeModeNoException() {
     }
 
     private boolean enterCommittingBlock() {
-        synchronized(this.commitLock){
-            try {
-                while(this.committing) {
-                    this.commitLock.wait(COMMIT_WAIT_MAX);
-                }
-                this.committing = true;
-                return true;
-            } catch(InterruptedException ie) {
-                Thread.currentThread().interrupt();
-                return false;
-            }
+        try {
+            this.commitLock.lockInterruptibly();
+            return true;
+        } catch (InterruptedException ie) {
+            this.commitLock.unlock();
+            Thread.currentThread().interrupt();
+            return false;
         }
     }
 
     private void leaveCommittingBlock() {
-        synchronized(this.commitLock){
-            this.committing = false;
-            this.commitLock.notifyAll();
-        }
+        this.commitLock.unlock();
     }
 
     /**
@@ -583,40 +575,33 @@ public void close() throws JMSException {
     }
 
     void internalClose() throws JMSException {
-        if (this.closed) return;
-        logger.trace("close session {}", this);
-
-        synchronized (this.closeLock) {
-            try {
-                // 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();
-                }
-
-                //clear up potential executor
-                this.deliveryExecutor.close();
-
-                //close all producers created by this session
-                for (RMQMessageProducer producer : this.producers) {
-                    producer.internalClose();
-                }
-                this.producers.clear();
+        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();
+            }
 
-                //now commit anything done during close
-                if (this.getTransactedNoException()) {
-                    this.commit();
-                }
+            //clear up potential executor
+            this.deliveryExecutor.close();
 
-                this.closeRabbitChannels();
+            //close all producers created by this session
+            for (RMQMessageProducer producer : this.producers) {
+                producer.internalClose();
+            }
+            this.producers.clear();
 
-            } finally {
-                this.closed = true;
+            //now commit anything done during close
+            if (this.getTransactedNoException()) {
+                this.commit();
             }
+
+            this.closeRabbitChannels();
         }
     }
 
@@ -670,7 +655,8 @@ public void recover() throws JMSException {
         if (getTransactedNoException()) {
             throw new javax.jms.IllegalStateException("Session is transacted.");
         } else {
-            synchronized (this.unackedMessageTags) {
+            try {
+               this.unackedMessageTagsLock.lock();
                 /* If we have messages to recover */
                 if (!this.unackedMessageTags.isEmpty()) {
                     try {
@@ -681,6 +667,8 @@ public void recover() throws JMSException {
                     }
                     this.unackedMessageTags.clear();
                 }
+            } finally {
+               this.unackedMessageTagsLock.unlock();
             }
         }
     }
@@ -1074,13 +1062,14 @@ public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JM
      */
     Channel getBrowsingChannel() throws JMSException {
         try {
-            synchronized (this.bcLock) {
-                Channel chan = this.getConnection().createRabbitChannel(false); // not transactional
-                this.browsingChannels.add(chan);
-                return chan;
-            }
+            this.bcLock.lock();
+            Channel chan = this.getConnection().createRabbitChannel(false); // not transactional
+            this.browsingChannels.add(chan);
+            return chan;
         } catch (Exception e) { // includes unchecked exceptions, e.g. ShutdownSignalException
             throw new RMQJMSException("Cannot create browsing channel", e);
+        } finally {
+            this.bcLock.unlock();
         }
     }
 
@@ -1101,7 +1090,8 @@ private static Map<String, Object> merge(Map<String, Object> m1, Map<String, Obj
      * Silently close and discard browsing channels, if any.
      */
     private void clearBrowsingChannels() {
-        synchronized (this.bcLock) {
+        try {
+            this.bcLock.lock();
             for (Channel chan : this.browsingChannels) {
                 try {
                     if (chan.isOpen())
@@ -1111,6 +1101,8 @@ private void clearBrowsingChannels() {
                 }
             }
             this.browsingChannels.clear();
+        } finally {
+            this.bcLock.unlock();
         }
     }
 
@@ -1119,15 +1111,16 @@ private void clearBrowsingChannels() {
      */
     void closeBrowsingChannel(Channel chan) {
         try {
-            synchronized (this.bcLock) {
-                if (this.browsingChannels.remove(chan)) {
-                    if (chan.isOpen())
-                        chan.close();
+            this.bcLock.lock();
+            if (this.browsingChannels.remove(chan)) {
+                if (chan.isOpen()) {
+                    chan.close();
                 }
             }
         } catch (Exception e) {
-            // throw new RMQJMSException("Cannot close browsing channel", _);
             // ignore errors in clearing up
+        } finally {
+            this.bcLock.unlock();
         }
     }
 
@@ -1303,8 +1296,11 @@ void resume() throws JMSException {
 
     void unackedMessageReceived(long dTag) {
         if (!getTransactedNoException()) {
-            synchronized (this.unackedMessageTags) {
+            try {
+                this.unackedMessageTagsLock.lock();
                 this.unackedMessageTags.add(dTag);
+            } finally {
+                this.unackedMessageTagsLock.unlock();
             }
         }
     }
@@ -1344,32 +1340,33 @@ private void acknowledge(long messageTag) throws JMSException {
              * The individualAck option is set by session mode (CLIENT_INDIVIDUAL_ACKNOWLEDGE) and overrides groupAck (default) and acknowledges at most a single message.
              * </p>
              */
-            synchronized (this.unackedMessageTags) {
-                try {
-                    if (individualAck) {
-                        if (!this.unackedMessageTags.contains(messageTag)) return; // this message already acknowledged
-                        /* ACK a single message */
-                        this.getChannel().basicAck(messageTag, false); // we ack the single message with this tag
-                        this.unackedMessageTags.remove(messageTag);
-                    } else if (groupAck) {
-                        /** The tags that precede the given one, and the given one, if unacknowledged */
-                        SortedSet<Long> previousTags = this.unackedMessageTags.headSet(messageTag+1);
-                        if (previousTags.isEmpty()) return; // no message to acknowledge
-                        /* ack multiple message up until the existing tag */
-                        this.getChannel().basicAck(previousTags.last(), // we ack the latest one (which might be this one, but might not be)
-                                              true);               // and everything prior to that
-                        // now remove all the tags <= messageTag
-                        previousTags.clear();
-                    } else {
-                        // this block is no longer possible (groupAck == true) after RJMS 1.2.0
-                        this.getChannel().basicAck(this.unackedMessageTags.last(), // we ack the highest tag
-                                              true);                          // and everything prior to that
-                        this.unackedMessageTags.clear();
-                    }
-                } catch (IOException x) {
-                    this.logger.error("RabbitMQ exception on basicAck of message {}; on session '{}'", messageTag, this, x);
-                    throw new RMQJMSException(x);
+            try {
+                this.unackedMessageTagsLock.lock();
+                if (individualAck) {
+                    if (!this.unackedMessageTags.contains(messageTag)) return; // this message already acknowledged
+                    /* ACK a single message */
+                    this.getChannel().basicAck(messageTag, false); // we ack the single message with this tag
+                    this.unackedMessageTags.remove(messageTag);
+                } else if (groupAck) {
+                    /** The tags that precede the given one, and the given one, if unacknowledged */
+                    SortedSet<Long> previousTags = this.unackedMessageTags.headSet(messageTag+1);
+                    if (previousTags.isEmpty()) return; // no message to acknowledge
+                    /* ack multiple message up until the existing tag */
+                    this.getChannel().basicAck(previousTags.last(), // we ack the latest one (which might be this one, but might not be)
+                        true);               // and everything prior to that
+                    // now remove all the tags <= messageTag
+                    previousTags.clear();
+                } else {
+                    // this block is no longer possible (groupAck == true) after RJMS 1.2.0
+                    this.getChannel().basicAck(this.unackedMessageTags.last(), // we ack the highest tag
+                        true);                          // and everything prior to that
+                    this.unackedMessageTags.clear();
                 }
+            } catch (IOException x) {
+                this.logger.error("RabbitMQ exception on basicAck of message {}; on session '{}'", messageTag, this, x);
+                throw new RMQJMSException(x);
+            } finally {
+                this.unackedMessageTagsLock.unlock();
             }
         }
     }
@@ -1444,7 +1441,8 @@ private MessageConsumer createTopicConsumer(Topic topic, String name, boolean du
 
         RMQDestination topicDest = (RMQDestination) topic;
         String queueName = durable ? name : generateJmsConsumerQueueName();
-        synchronized (this.subscriptions) {
+        try {
+            this.subscriptionsLock.lock();
             Subscription subscription = this.subscriptions.register(name, queueName, durable, shared,
                 messageSelector, noLocal);
             PostAction postAction = subscription.validateNewConsumer(topic, durable, shared,
@@ -1459,6 +1457,8 @@ private MessageConsumer createTopicConsumer(Topic topic, String name, boolean du
             consumer.setNoLocal(noLocal);
             subscription.add(consumer);
             return consumer;
+        } finally {
+            this.subscriptionsLock.unlock();
         }
     }