update .hgignore for idea

Author: Rob Harrop

.hgignore

@@ -9,3 +9,5 @@ syntax: glob
 
 syntax: regexp
 ^build/
+^out/
+^.idea/

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

@@ -32,6 +32,7 @@
 package com.rabbitmq.client.test.functional;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.test.BrokerTestCase;
 
 import java.io.IOException;
@@ -43,14 +44,26 @@
  */
 public class PerQueueTTL extends BrokerTestCase {
 
+    private static final String TTL_EXCHANGE = "ttl.exchange";
+
     private static final String TTL_ARG = "x-message-ttl";
 
     private static final String TTL_QUEUE_NAME = "queue.ttl";
 
     private static final String TTL_INVALID_QUEUE_NAME = "invalid.queue.ttl";
 
+    @Override
+    protected void createResources() throws IOException {
+        this.channel.exchangeDeclare(TTL_EXCHANGE, "direct");
+    }
+
+    @Override
+    protected void releaseResources() throws IOException {
+        this.channel.exchangeDelete(TTL_EXCHANGE);
+    }
+
     public void testCreateQueueWithTTL() throws IOException {
-        AMQP.Queue.DeclareOk declareOk = declareQueue(TTL_QUEUE_NAME, 2000);
+        AMQP.Queue.DeclareOk declareOk = declareQueue(TTL_QUEUE_NAME, 2000L);
         assertNotNull(declareOk);
     }
 
@@ -72,9 +85,70 @@ public void testCreateQueueWithZeroTTL() throws Exception {
         }
     }
 
+    /*
+     * Test messages expire when using basic get.
+     */
+    public void testPublishAndGetWithExpiry() throws Exception {
+        long ttl = 2000;
+        declareQueue(TTL_QUEUE_NAME, ttl);
+        this.channel.queueBind(TTL_QUEUE_NAME, TTL_EXCHANGE, TTL_QUEUE_NAME);
+
+        byte[] msg1 = "one".getBytes();
+        byte[] msg2 = "two".getBytes();
+        byte[] msg3 = "three".getBytes();
+
+        this.channel.basicPublish(TTL_EXCHANGE, TTL_QUEUE_NAME, null, msg1);
+        Thread.sleep(1500);
+
+        this.channel.basicPublish(TTL_EXCHANGE, TTL_QUEUE_NAME, null, msg2);
+        Thread.sleep(1000);
+
+        this.channel.basicPublish(TTL_EXCHANGE, TTL_QUEUE_NAME, null, msg3);
+
+        assertEquals("two", new String(get()));
+        assertEquals("three", new String(get()));
+
+    }
+    
+    /*
+     * Test get expiry for messages sent under a transaction
+     */
+    public void testTransactionalPublishWithGet() throws Exception {
+        long ttl = 1000;
+        declareQueue(TTL_QUEUE_NAME, ttl);
+        this.channel.queueBind(TTL_QUEUE_NAME, TTL_EXCHANGE, TTL_QUEUE_NAME);
+
+        byte[] msg1 = "one".getBytes();
+        byte[] msg2 = "two".getBytes();
+
+        this.channel.txSelect();
+
+        this.channel.basicPublish(TTL_EXCHANGE, TTL_QUEUE_NAME, null, msg1);
+        Thread.sleep(1500);
+
+        this.channel.basicPublish(TTL_EXCHANGE, TTL_QUEUE_NAME, null, msg2);
+        this.channel.txCommit();
+        Thread.sleep(500);
+
+        assertEquals("one", new String(get()));
+        Thread.sleep(800);
+
+        assertNull(get());
+    }
+
+
+    private byte[] get() throws IOException {
+        GetResponse response = this.channel.basicGet(TTL_QUEUE_NAME, false);
+        if(response == null) {
+            return null;
+        }
+        return response.getBody();
+    }
+
     private AMQP.Queue.DeclareOk declareQueue(String name, Object ttlValue) throws IOException {
-        AMQP.Queue.DeclareOk declareOk = this.channel.queueDeclare(name, false, false, false,
-                Collections.<String, Object>singletonMap(TTL_ARG, ttlValue));
-        return declareOk;
+        Map<String, Object> argMap = Collections.singletonMap(TTL_ARG, ttlValue);
+        return this.channel.queueDeclare(name, false, true, false, argMap);
     }
+
+
 }

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

@@ -49,7 +49,6 @@
 import org.apache.commons.cli.ParseException;
 
 import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Address;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -58,7 +57,6 @@
 import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.ReturnListener;
 import com.rabbitmq.client.ShutdownSignalException;
-import com.rabbitmq.client.AMQP.Queue;
 import com.rabbitmq.client.QueueingConsumer.Delivery;