Merge bug23802 into default.

Author: Steve Powell

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

@@ -82,20 +82,20 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
     /** The ConfirmListener collection. */
     private final Collection<ConfirmListener> confirmListeners = new CopyOnWriteArrayList<ConfirmListener>();
 
-    /** Sequence number of next published message requiring confirmation. */
-    private long nextPublishSeqNo = 0L;
-
     /** The current default consumer, or null if there is none. */
     private volatile Consumer defaultConsumer = null;
 
     /** Set of currently unconfirmed messages (i.e. messages that have
-     *  not been ack'd or nack'd by the server yet. */
-    private volatile SortedSet<Long> unconfirmedSet =
+     *  not been ack'd or nack'd by the server yet.
+     *  Used as monitor and protects nextPublishSeqNo and onlyAcksReceived. */
+    private SortedSet<Long> unconfirmedSet =
             Collections.synchronizedSortedSet(new TreeSet<Long>());
-
+    /** Sequence number of next published message requiring confirmation.
+     * 0 means no confirmations. */
+    private long nextPublishSeqNo = 0L;
     /** Whether any nacks have been received since the last
      * waitForConfirms(). */
-    private volatile boolean onlyAcksReceived = true;
+    private boolean noNacksReceived = true;
 
     /**
      * Construct a new channel on the given connection with the given
@@ -167,8 +167,8 @@ public boolean waitForConfirms()
                     throw Utility.fixStackTrace(getCloseReason());
                 }
                 if (unconfirmedSet.isEmpty()) {
-                    boolean aux = onlyAcksReceived;
-                    onlyAcksReceived = true;
+                    boolean aux = noNacksReceived;
+                    noNacksReceived = true;
                     return aux;
                 }
                 unconfirmedSet.wait();
@@ -182,7 +182,7 @@ public void waitForConfirmsOrDie()
     {
         if (!waitForConfirms()) {
             close(AMQP.REPLY_SUCCESS, "NACKS RECEIVED", true, null, false);
-            throw new IOException("nacks received");
+            throw new IOException("nacks received", getCloseReason());
         }
     }
 
@@ -321,7 +321,7 @@ public void releaseChannelNumber() {
             } else if (method instanceof Basic.Nack) {
                 Basic.Nack nack = (Basic.Nack) method;
                 callConfirmListeners(command, nack);
-                handleAckNack(nack.getDeliveryTag(), nack.getMultiple(), false);
+                handleAckNack(nack.getDeliveryTag(), nack.getMultiple(), true);
                 return true;
             } else if (method instanceof Basic.RecoverOk) {
                 for (Consumer callback: _consumers.values()) {
@@ -551,9 +551,11 @@ public void basicPublish(String exchange, String routingKey,
                              BasicProperties props, byte[] body)
         throws IOException
     {
-        if (nextPublishSeqNo > 0) {
-            unconfirmedSet.add(getNextPublishSeqNo());
-            nextPublishSeqNo++;
+        synchronized(unconfirmedSet) {
+            if (nextPublishSeqNo > 0) {
+                unconfirmedSet.add(nextPublishSeqNo);
+                nextPublishSeqNo++;
+            }
         }
         BasicProperties useProps = props;
         if (props == null) {
@@ -994,7 +996,9 @@ public Tx.RollbackOk txRollback()
     public Confirm.SelectOk confirmSelect()
         throws IOException
     {
-        if (nextPublishSeqNo == 0) nextPublishSeqNo = 1;
+        synchronized(unconfirmedSet) {
+            if (nextPublishSeqNo == 0) nextPublishSeqNo = 1;
+        }
         return (Confirm.SelectOk)
             exnWrappingRpc(new Confirm.Select(false)).getMethod();
 
@@ -1012,7 +1016,9 @@ public Channel.FlowOk getFlow() {
 
     /** Public API - {@inheritDoc} */
     public long getNextPublishSeqNo() {
-        return nextPublishSeqNo;
+        synchronized(unconfirmedSet) {
+            return nextPublishSeqNo;
+        }
     }
 
     public void asyncRpc(Method method) throws IOException {
@@ -1024,13 +1030,13 @@ public AMQCommand rpc(Method method) throws IOException {
     }
 
     protected void handleAckNack(long seqNo, boolean multiple, boolean nack) {
-        if (multiple) {
-            unconfirmedSet.headSet(seqNo + 1).clear();
-        } else {
-            unconfirmedSet.remove(seqNo);
-        }
         synchronized (unconfirmedSet) {
-            onlyAcksReceived = onlyAcksReceived && !nack;
+            if (multiple) {
+                unconfirmedSet.headSet(seqNo + 1).clear();
+            } else {
+                unconfirmedSet.remove(seqNo);
+            }
+            noNacksReceived &= !nack;
             if (unconfirmedSet.isEmpty())
                 unconfirmedSet.notifyAll();
         }

test/src/com/rabbitmq/client/test/Bug20004Test.java

@@ -25,9 +25,9 @@
  * tests.
  */
 public class Bug20004Test extends BrokerTestCase {
-    public Exception caughtException = null;
-    public boolean completed = false;
-    public boolean created = false;
+    private volatile Exception caughtException = null;
+    private volatile boolean completed = false;
+    private volatile boolean created = false;
 
     protected void releaseResources()
         throws IOException
@@ -37,6 +37,7 @@ protected void releaseResources()
         }
     }
 
+    @SuppressWarnings("deprecation")
     public void testBug20004()
         throws IOException
     {

test/src/com/rabbitmq/client/test/ConfirmBase.java

@@ -0,0 +1,61 @@
+//  The contents of this file are subject to the Mozilla Public License
+//  Version 1.1 (the "License"); you may not use this file except in
+//  compliance with the License. You may obtain a copy of the License
+//  at http://www.mozilla.org/MPL/
+//
+//  Software distributed under the License is distributed on an "AS IS"
+//  basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+//  the License for the specific language governing rights and
+//  limitations under the License.
+//
+//  The Original Code is RabbitMQ.
+//
+//  The Initial Developer of the Original Code is VMware, Inc.
+//  Copyright (c) 2011 VMware, Inc.  All rights reserved.
+//
+
+package com.rabbitmq.client.test;
+
+import com.rabbitmq.client.ShutdownSignalException;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.TimeUnit;
+
+import junit.framework.AssertionFailedError;
+
+public class ConfirmBase extends BrokerTestCase {
+    protected void waitForConfirms() throws Exception
+    {
+        waitForConfirms("ConfirmBase.waitForConfirms");
+    }
+
+    protected void waitForConfirms(final String testTitle) throws Exception
+    {
+        try {
+            FutureTask<?> waiter = new FutureTask<Object>(new Runnable() {
+                    public void run() {
+                        try {
+                            channel.waitForConfirmsOrDie();
+                        } catch (IOException e) {
+                            throw (ShutdownSignalException)e.getCause();
+                        } catch (InterruptedException _) {
+                            fail(testTitle + ": interrupted");
+                        }
+                    }
+                }, null);
+            (Executors.newSingleThreadExecutor()).execute(waiter);
+            waiter.get(10, TimeUnit.SECONDS);
+        } catch (ExecutionException ee) {
+            Throwable t = ee.getCause();
+            if (t instanceof ShutdownSignalException) throw (ShutdownSignalException) t;
+            if (t instanceof AssertionFailedError) throw (AssertionFailedError) t;
+            throw (Exception)t;
+        } catch (TimeoutException _) {
+            fail(testTitle + ": timeout");
+        }
+    }
+}

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

@@ -56,6 +56,7 @@ public void testQueuePurge() throws IOException {
      * longer purged, even if the channel they were sent down is not
      * (Tx-)transacted."
      */
+    @SuppressWarnings("deprecation")
     public void testUnackedPurge() throws IOException {
         Binding binding = setupExchangeBindings(false);
         channel.basicPublish(binding.x, binding.k, null, payload);