Merge branch 'master' into prevent-socket-write-blocking

Author: michaelklishin

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

@@ -26,6 +26,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -145,7 +146,7 @@ public class ConnectionRecovery extends BrokerTestCase {
 
     // see https://github.com/rabbitmq/rabbitmq-java-client/issues/135
     @Test public void thatShutdownHooksOnConnectionFireBeforeRecoveryStarts() throws IOException, InterruptedException {
-        final List<String> events = new ArrayList<String>();
+        final List<String> events = new CopyOnWriteArrayList<String>();
         final CountDownLatch latch = new CountDownLatch(1);
         connection.addShutdownListener(new ShutdownListener() {
             public void shutdownCompleted(ShutdownSignalException cause) {

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

@@ -583,7 +583,7 @@ private void sleep(long millis) {
        publication time + TTL + latency */
     private void checkPromptArrival(QueueingConsumer c,
                                     int count, long latency) throws Exception {
-        long epsilon = TTL / 20;
+        long epsilon = TTL / 15;
         for (int i = 0; i < count; i++) {
             Delivery d = c.nextDelivery(TTL + TTL + latency + epsilon);
             assertNotNull("message #" + i + " did not expire", d);

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

@@ -156,7 +156,7 @@ protected void releaseResources() throws IOException {
         closeChannel();
         openChannel();
 
-        Thread.sleep(150);
+        Thread.sleep(110);
         expectBodyAndRemainingMessages(MSG[1], 1);
         expectBodyAndRemainingMessages(MSG[2], 0);
     }

src/test/java/com/rabbitmq/client/test/server/XDeathHeaderGrowth.java

@@ -28,6 +28,8 @@
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.junit.Test;
 
@@ -39,25 +41,37 @@
 
 class RejectingConsumer extends DefaultConsumer {
     private CountDownLatch latch;
-    private Map<String, Object> headers;
+    private volatile Map<String, Object> headers;
+    private AtomicLong counter;
+
 
     public RejectingConsumer(Channel channel, CountDownLatch latch) {
         super(channel);
         this.latch = latch;
+        this.counter = new AtomicLong(latch.getCount());
     }
 
     @Override
     public void handleDelivery(String consumerTag, Envelope envelope,
                                AMQP.BasicProperties properties, byte[] body)
             throws IOException {
+
         if(this.latch.getCount() > 0) {
             this.getChannel().basicReject(envelope.getDeliveryTag(), false);
         } else {
             if(this.getChannel().isOpen()) {
                 this.getChannel().basicAck(envelope.getDeliveryTag(), false);
             }
         }
-        this.headers = properties.getHeaders();
+        if(this.counter.decrementAndGet() == 0) {
+            // get headers only when the message has been redelivered
+            // the expected number of times.
+            // it looks like the message can be redelivered because
+            // of the reject when handleDelivery isn't done yet or
+            // before the latch releases the main thread. There's then
+            // an additional delivery and the checks (transiently) fail.
+            this.headers = properties.getHeaders();
+        }
         latch.countDown();
     }
 
@@ -88,7 +102,7 @@ public class XDeathHeaderGrowth extends BrokerTestCase {
         this.channel.queueBind(q3, x2, "");
 
         final String qz = "issues.rabbitmq-server-78.destination";
-        declareTransientQueue(qz, argumentsForDeadLetteringTo(x3));
+        declareTransientQueue(qz, argumentsForDeadLetteringWithoutTtlTo(x3));
         this.channel.queueBind(qz, x3, "");
 
         CountDownLatch latch = new CountDownLatch(10);
@@ -143,7 +157,7 @@ private void cleanUpQueues(String... qs) throws IOException {
         this.channel.queueBind(q2, x1, "");
 
         final String qz = "issues.rabbitmq-server-152.destination";
-        declareTransientQueue(qz, argumentsForDeadLetteringTo(x2));
+        declareTransientQueue(qz, argumentsForDeadLetteringWithoutTtlTo(x2));
         this.channel.queueBind(qz, x2, "");
 
         CountDownLatch latch = new CountDownLatch(10);
@@ -215,11 +229,17 @@ private Map<String, Object> argumentsForDeadLetteringTo(String dlx) {
         return argumentsForDeadLetteringTo(dlx, 1);
     }
 
+    private Map<String, Object> argumentsForDeadLetteringWithoutTtlTo(String dlx) {
+        return argumentsForDeadLetteringTo(dlx, -1);
+    }
+
     private Map<String, Object> argumentsForDeadLetteringTo(String dlx, int ttl) {
         Map<String, Object> m = new HashMap<String, Object>();
         m.put("x-dead-letter-exchange", dlx);
         m.put("x-dead-letter-routing-key", "some-routing-key");
-        m.put("x-message-ttl", ttl);
+        if(ttl > 0) {
+            m.put("x-message-ttl", ttl);
+        }
         return m;
     }
 }