Merging bug23616 to default

Author: Matthew Sackman

src/com/rabbitmq/client/Channel.java

@@ -688,10 +688,9 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
     Confirm.SelectOk confirmSelect() throws IOException;
 
     /**
-     * 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; if the channel is not in confirm mode, -1 is
-     * returned */
-    long getPublishedMessageCount();
+     * When in confirm mode, returns the sequence number of the next
+     * message to be published.
+     * @return the sequence number of the next message to be published
+     */
+    long getNextPublishSeqNo();
 }

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

@@ -105,9 +105,9 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      */
     public volatile AckListener ackListener = null;
 
-    /** Current published message count (used by publisher acknowledgements)
+    /** Sequence number of next published message requiring confirmation.
      */
-    private final AtomicLong publishedMessageCount = new AtomicLong(-1);
+    private long nextPublishSeqNo = 0L;
 
     /** Reference to the currently-active default consumer, or null if there is
      *  none.
@@ -498,8 +498,7 @@ public void basicPublish(String exchange, String routingKey,
                              BasicProperties props, byte[] body)
         throws IOException
     {
-        if (publishedMessageCount.get() >= 0)
-            publishedMessageCount.incrementAndGet();
+        if (nextPublishSeqNo > 0) nextPublishSeqNo++;
         BasicProperties useProps = props;
         if (props == null) {
             useProps = MessageProperties.MINIMAL_BASIC;
@@ -868,8 +867,7 @@ public Tx.RollbackOk txRollback()
     public Confirm.SelectOk confirmSelect()
         throws IOException
     {
-        if (publishedMessageCount.get() == -1)
-            publishedMessageCount.set(0);
+        if (nextPublishSeqNo == 0) nextPublishSeqNo = 1;
         return (Confirm.SelectOk)
             exnWrappingRpc(new Confirm.Select(false)).getMethod();
 
@@ -886,7 +884,7 @@ public Channel.FlowOk getFlow() {
     }
 
     /** Public API - {@inheritDoc} */
-    public long getPublishedMessageCount() {
-        return publishedMessageCount.longValue();
+    public long getNextPublishSeqNo() {
+        return nextPublishSeqNo;
     }
 }

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

@@ -254,7 +254,7 @@ private void publishN(String exchangeName, String queueName,
         throws IOException
     {
         for (long i = 0; i < NUM_MESSAGES; i++) {
-            ackSet.add(i);
+            ackSet.add(channel.getNextPublishSeqNo());
             publish(exchangeName, queueName, persistent, mandatory, immediate);
         }
     }

test/src/com/rabbitmq/examples/MulticastMain.java

@@ -344,7 +344,7 @@ private void publish(byte[] msg)
         }
 
         private boolean throttlePubAck() {
-            return ((pubAckCount > 0) && (channel.getPublishedMessageCount() - mostRecentAcked > pubAckCount));
+            return ((pubAckCount > 0) && (channel.getNextPublishSeqNo() - mostRecentAcked > pubAckCount));
         }
 
         private void delay(long now)