Corrected start/stop logic for AsyncLogger.

Author: Steve Powell

src/com/rabbitmq/client/ConsumerCancelledException.java

@@ -24,7 +24,6 @@ public class ConsumerCancelledException extends RuntimeException implements
     /** Default for non-checking. */
     private static final long serialVersionUID = 1L;
 
-    @Override
     public ConsumerCancelledException sensibleClone() {
         try {
             return (ConsumerCancelledException) super.clone();

src/com/rabbitmq/tools/Tracer.java

@@ -31,6 +31,7 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.impl.AMQCommand;
@@ -388,7 +389,7 @@ public static class AsyncLogger implements Logger {
 
         private final Runnable loggerRunnable;
 
-        private final AtomicBoolean started;
+        private final SafeCounter countStarted;
         private volatile Thread loggerThread = null;
 
         /**
@@ -430,7 +431,7 @@ public AsyncLogger(OutputStream os, int flushInterval) {
                 throw new IllegalArgumentException("Flush interval ("
                         + flushInterval + "ms) must be positive and at least "
                         + MIN_FLUSH_INTERVAL + "ms.");
-            this.started = new AtomicBoolean(false);
+            this.countStarted = new SafeCounter();
 
             PrintStream printStream = new PrintStream(new BufferedOutputStream(
                     os, BUFFER_SIZE), false);
@@ -448,8 +449,17 @@ public void log(String message) {
             }
         }
 
+        public boolean start() {
+            if (this.countStarted.testZeroAndIncrement()) {
+                this.loggerThread = new Thread(this.loggerRunnable);
+                this.loggerThread.start();
+                return true;
+            }
+            return false; // meaning already started
+        }
+
         public boolean stop() {
-            if (this.started.compareAndSet(true, false)) {
+            if (this.countStarted.decrementAndTestZero()) {
                 if (this.loggerThread != null) {
                     try {
                         this.queue.put(new Pr<String, LogCmd>(null, LogCmd.STOP));
@@ -464,16 +474,7 @@ public boolean stop() {
             return false; // meaning already stopped
         }
 
-        public boolean start() {
-            if (this.started.compareAndSet(false, true)) {
-                this.loggerThread = new Thread(this.loggerRunnable);
-                this.loggerThread.start();
-                return true;
-            }
-            return false; // meaning already started
-        }
-
-        private static class AsyncLoggerRunnable implements Runnable {
+        private class AsyncLoggerRunnable implements Runnable {
             private final int flushInterval;
             private final PrintStream ps;
             private final BlockingQueue<Pr<String, LogCmd> > queue;
@@ -515,6 +516,7 @@ public void run() {
                     this.ps.flush();
 
                 } catch (InterruptedException ie) {
+                    AsyncLogger.this.countStarted.reset();
                     drainCurrentQueue();
                     this.ps.println("Interrupted.");
                     this.ps.flush();
@@ -531,4 +533,31 @@ private void drainCurrentQueue() {
             }
         }
     }
+    
+    private static class SafeCounter {
+        private final Object countMonitor = new Object();
+        private int count;
+        public SafeCounter() {
+            this.count = 0;
+        }
+        public boolean testZeroAndIncrement() {
+            synchronized (this.countMonitor) {
+                int val = this.count;
+                this.count++;
+                return (val == 0);
+            }
+        }
+        public boolean decrementAndTestZero() {
+            synchronized (this.countMonitor) {
+                if (this.count == 0) return false;
+                --this.count;
+                return (0 == this.count);
+            }
+        }
+        public void reset() {
+            synchronized (this.countMonitor) {
+                this.count = 0;
+            }
+        }
+    }
 }

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

@@ -57,7 +57,6 @@ public void testConsumerCancellationInterruptsQueuingConsumerWait()
         final QueueingConsumer consumer = new QueueingConsumer(channel);
         Runnable receiver = new Runnable() {
 
-            @Override
             public void run() {
                 try {
                     try {

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

@@ -108,9 +108,10 @@ public void testTTLMustBePositive() throws Exception {
     public void testQueueRedeclareEquivalence() throws Exception {
         declareQueue(TTL_QUEUE_NAME, 10);
         try {
-             declareQueue(TTL_QUEUE_NAME, 20);
+            declareQueue(TTL_QUEUE_NAME, 20);
+            fail("Should not be able to redeclare with different TTL");
         } catch(IOException ex) {
-            checkShutdownSignal(AMQP.NOT_ALLOWED, ex);
+            checkShutdownSignal(AMQP.PRECONDITION_FAILED, ex);
         }
     }