better naming for publish sequence number

also, there is on need to make the var atomic. It is only ever
accessed by application threads and our general rules for Channel
usage dictate that there must only ever be one of these per channel.

Author: rade

src/com/rabbitmq/client/Channel.java

@@ -688,10 +688,11 @@ 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();
+     * Returns the sequence number of the next message to be published
+     * that requires confirmation.
+     * @return the sequence number of the next message to be published
+     * that requires confirmation. 0 if the channel is not in confirm mode.
+     * not in confirm mode
+     */
+    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 = 0;
         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/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)