refactor

Author: Alexandru Scvortov

src/com/rabbitmq/client/Channel.java

@@ -649,7 +649,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
 
     /**
      * Enables publisher acknowledgements on this channel.
-     * @param many determines whether the broker can acknowledge
+     * @param multiple determines whether the broker can acknowledge
      * multiple messages at the same time
      * @see com.rabbitmq.client.AMQP.Confirm.Select
      * @throws java.io.IOException if an error is encountered
@@ -660,6 +660,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * Returns the number of messages published since the channel was
      * put in confirm mode.
      * @return the number of messages published since the first
-     * confirm.select */
+     * confirm.select; if the channel is not in confirm mode, -1 is
+     * returned */
     long getPublishedMessageCount();
 }

src/com/rabbitmq/client/impl/ChannelN.java

@@ -56,6 +56,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.TimeoutException;
 
 
@@ -106,7 +107,7 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
 
     /** Current published message count (used by publisher acknowledgements)
      */
-    private long publishedMessageCount;
+    private AtomicLong publishedMessageCount = new AtomicLong(-1);
 
     /** Reference to the currently-active default consumer, or null if there is
      *  none.
@@ -497,10 +498,8 @@ public void basicPublish(String exchange, String routingKey,
                              BasicProperties props, byte[] body)
         throws IOException
     {
-        synchronized (_channelMutex) {
-            if (publishedMessageCount >= 0)
-                ++publishedMessageCount;
-        }
+        if (publishedMessageCount.get() >= 0)
+            publishedMessageCount.incrementAndGet();
         BasicProperties useProps = props;
         if (props == null) {
             useProps = MessageProperties.MINIMAL_BASIC;
@@ -847,10 +846,8 @@ public Tx.RollbackOk txRollback()
     public Confirm.SelectOk confirmSelect(boolean multiple)
         throws IOException
     {
-        synchronized (_channelMutex) {
-            if (publishedMessageCount == -1)
-                publishedMessageCount = 0;
-        }
+        if (publishedMessageCount.get() == -1)
+            publishedMessageCount.set(0);
         return (Confirm.SelectOk)
             exnWrappingRpc(new Confirm.Select(multiple, false)).getMethod();
 
@@ -868,6 +865,6 @@ public Channel.FlowOk getFlow() {
 
     /** Public API - {@inheritDoc} */
     public long getPublishedMessageCount() {
-        return publishedMessageCount;
+        return publishedMessageCount.longValue();
     }
 }

test/src/com/rabbitmq/client/test/functional/Confirm.java

@@ -53,15 +53,14 @@ public class Confirm extends BrokerTestCase
     protected void setUp() throws IOException {
         super.setUp();
         ackSet = new TreeSet<Long>();
-        final Confirm This = this;
         channel.setAckListener(new AckListener() {
                 public void handleAck(long seqNo,
                                       boolean multiple) {
                     if (multiple) {
                         for (int i = 0; i <= seqNo; ++i)
-                            This.gotAckFor(i);
+                            Confirm.this.gotAckFor(i);
                     } else {
-                        This.gotAckFor(seqNo);
+                        Confirm.this.gotAckFor(seqNo);
                     }
                 }
             });