Consolidate test set up code in the BrokerRunning rule

spring-projects · Nov 11, 2010 · 59fa5c6 · 59fa5c6
1 parent 3e13076
commit 59fa5c6
Show file tree

Hide file tree

Showing 8 changed files with 110 additions and 149 deletions.
diff --git a/...-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/...-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
@@ -6,11 +6,11 @@
 import java.util.concurrent.TimeUnit;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.ShutdownSignalException;
-import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.utility.Utility;
 /**
  * Variation on QueueingConsumer in RabbitMQ, uses 'put' instead of 'add' and stored a reference to the consumerTag that
@@ -20,8 +20,6 @@
  */
 public class BlockingQueueConsumer extends DefaultConsumer {
 
-		private String consumerTag;
-
 	    private final BlockingQueue<Delivery> queue;
 
 	    // When this is non-null the queue is in shutdown mode and nextDelivery should
@@ -43,18 +41,7 @@ public BlockingQueueConsumer(Channel ch, BlockingQueue<Delivery> q)
 	        super(ch);
 	        this.queue = q;
 	    }
-
-
-
-	    public String getConsumerTag() {
-			return consumerTag;
-		}
-
-	    public void setConsumerTag(String consumerTag) 
-	    {
-	    	this.consumerTag = consumerTag;
-	    }
-
+
 		@Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
 	        shutdown = sig; 
 	        try {

diff --git a/...rc/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/...rc/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
@@ -260,8 +260,7 @@ protected BlockingQueueConsumer createBlockingQueueConsumer(final Channel channe
 		String[] queue = StringUtils.commaDelimitedListToStringArray(queueNames);
 		for (int i = 0; i < queue.length; i++) {
 			channel.queueDeclarePassive(queue[i]);
-			String consumerTag = channel.basicConsume(queue[i], !isChannelTransacted(), consumer);
-			consumer.setConsumerTag(consumerTag);
+			channel.basicConsume(queue[i], !isChannelTransacted(), consumer);
 		}
 		return consumer;
 	}

diff --git a/...bit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java b/...bit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
@@ -3,10 +3,8 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 
-import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
-import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
@@ -23,27 +21,12 @@
 
 public class RabbitBindingIntegrationTests {
 
-	private Queue queue;
+	private Queue queue = new Queue("test.queue");
 
-	private RabbitTemplate template = new RabbitTemplate(
-			new CachingConnectionFactory());
+	private RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
 
 	@Rule
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
-
-	@Before
-	public void declareQueue() {
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue("test.queue");
-		} catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		queue = new Queue("test.queue");
-		// Idempotent, so no problem to do this for every test
-		admin.declareQueue(queue);
-		admin.purgeQueue("test.queue", false);
-	}
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue);
 
 	@Test
 	public void testSendAndReceiveWithTopicSingleCallback() throws Exception {
@@ -53,25 +36,29 @@ public void testSendAndReceiveWithTopicSingleCallback() throws Exception {
 		admin.declareExchange(exchange);
 		template.setExchange(exchange.getName());
 
-		admin.declareBinding(BindingBuilder.from(queue).to(exchange)
-				.with("*.end"));
+		admin.declareBinding(BindingBuilder.from(queue).to(exchange).with("*.end"));
 
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(
-						channel);
-				String tag = channel.basicConsume(queue.getName(), true,
-						consumer);
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
 				assertNotNull(tag);
 
 				template.convertAndSend("foo", "message");
-				String result = getResult(consumer);
-				assertEquals(null, result);
 
-				template.convertAndSend("foo.end", "message");
-				result = getResult(consumer);
-				assertEquals("message", result);
+				try {
+
+					String result = getResult(consumer);
+					assertEquals(null, result);
+
+					template.convertAndSend("foo.end", "message");
+					result = getResult(consumer);
+					assertEquals("message", result);
+
+				} finally {
+					channel.basicCancel(tag);
+				}
 
 				return null;
 
@@ -82,36 +69,29 @@ public Void doInRabbit(Channel channel) throws Exception {
 
 	@Test
 	// @Ignore("Not sure yet if we need to support a use case like this")
-	public void testSendAndReceiveWithTopicConsumeInBackground()
-			throws Exception {
+	public void testSendAndReceiveWithTopicConsumeInBackground() throws Exception {
 
 		RabbitAdmin admin = new RabbitAdmin(template);
 		TopicExchange exchange = new TopicExchange("topic");
 		admin.declareExchange(exchange);
 		template.setExchange(exchange.getName());
 
-		admin.declareBinding(BindingBuilder.from(queue).to(exchange)
-				.with("*.end"));
+		admin.declareBinding(BindingBuilder.from(queue).to(exchange).with("*.end"));
 
-		final RabbitTemplate template = new RabbitTemplate(
-				new CachingConnectionFactory());
+		final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
 		template.setExchange(exchange.getName());
 
-		BlockingQueueConsumer consumer = template
-				.execute(new ChannelCallback<BlockingQueueConsumer>() {
-					public BlockingQueueConsumer doInRabbit(Channel channel)
-							throws Exception {
+		BlockingQueueConsumer consumer = template.execute(new ChannelCallback<BlockingQueueConsumer>() {
+			public BlockingQueueConsumer doInRabbit(Channel channel) throws Exception {
 
-						BlockingQueueConsumer consumer = new BlockingQueueConsumer(
-								channel);
-						String tag = channel.basicConsume(queue.getName(),
-								true, consumer);
-						assertNotNull(tag);
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
+				assertNotNull(tag);
 
-						return consumer;
+				return consumer;
 
-					}
-				});
+			}
+		});
 
 		template.convertAndSend("foo", "message");
 		String result = getResult(consumer);
@@ -121,6 +101,8 @@ public BlockingQueueConsumer doInRabbit(Channel channel)
 		result = getResult(consumer);
 		assertEquals("message", result);
 
+		consumer.getChannel().basicCancel(consumer.getConsumerTag());
+
 	}
 
 	@Test
@@ -131,16 +113,13 @@ public void testSendAndReceiveWithTopicTwoCallbacks() throws Exception {
 		admin.declareExchange(exchange);
 		template.setExchange(exchange.getName());
 
-		admin.declareBinding(BindingBuilder.from(queue).to(exchange)
-				.with("*.end"));
+		admin.declareBinding(BindingBuilder.from(queue).to(exchange).with("*.end"));
 
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(
-						channel);
-				String tag = channel.basicConsume(queue.getName(), true,
-						consumer);
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
 				assertNotNull(tag);
 
 				try {
@@ -159,10 +138,8 @@ public Void doInRabbit(Channel channel) throws Exception {
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(
-						channel);
-				String tag = channel.basicConsume(queue.getName(), true,
-						consumer);
+				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel);
+				String tag = channel.basicConsume(queue.getName(), true, consumer);
 				assertNotNull(tag);
 
 				try {
@@ -180,15 +157,13 @@ public Void doInRabbit(Channel channel) throws Exception {
 
 	}
 
-	private String getResult(final BlockingQueueConsumer consumer)
-			throws InterruptedException {
+	private String getResult(final BlockingQueueConsumer consumer) throws InterruptedException {
 		Delivery response = consumer.nextDelivery(200L);
 		if (response == null) {
 			return null;
 		}
-		MessageProperties messageProps = RabbitUtils.createMessageProperties(
-				response.getProperties(), response.getEnvelope(), "UTF-8");
-		return (String) new SimpleMessageConverter().fromMessage(new Message(
-				response.getBody(), messageProps));
+		MessageProperties messageProps = RabbitUtils.createMessageProperties(response.getProperties(),
+				response.getEnvelope(), "UTF-8");
+		return (String) new SimpleMessageConverter().fromMessage(new Message(response.getBody(), messageProps));
 	}
 }
diff --git a/...it/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/...it/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
@@ -11,14 +11,11 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
-import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.UncategorizedAmqpException;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
 import org.springframework.amqp.rabbit.support.RabbitUtils;
@@ -42,20 +39,7 @@ public class RabbitTemplateIntegrationTests {
 	private RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory());
 
 	@Rule
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
-
-	@Before
-	public void declareQueue() {
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue(ROUTE);
-		}
-		catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		admin.declareQueue(new Queue(ROUTE));
-		admin.purgeQueue(ROUTE, false);
-	}
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(ROUTE);
 
 	@Test
 	public void testSendAndReceive() throws Exception {

diff --git a/.../java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java b/.../java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java
@@ -7,8 +7,6 @@
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
-import org.springframework.amqp.AmqpIOException;
-import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
@@ -30,7 +28,7 @@ public class RabbitTemplatePerformanceIntegrationTests {
 
 	@Rule
 	// After the repeat processor, so it only runs once
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(ROUTE);
 
 	private CachingConnectionFactory connectionFactory;
 
@@ -44,17 +42,6 @@ public void declareQueue() {
 		connectionFactory.setChannelCacheSize(repeat.getConcurrency());
 		// connectionFactory.setPort(5673);
 		template.setConnectionFactory(connectionFactory);
-		// TODO: investigate the effects of these flags...
-		// template.setMandatoryPublish(true);
-		// template.setImmediatePublish(true);
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue(ROUTE);
-		} catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		admin.declareQueue(new Queue(ROUTE));
-		admin.purgeQueue(ROUTE, false);
 	}
 
 	@After

diff --git a/...ringframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/...ringframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java
@@ -18,10 +18,8 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
-import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
@@ -31,7 +29,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(MessageListenerContainerLifecycleIntegrationTests.class);
 
-	private Queue queue;
+	private Queue queue = new Queue("test.queue");
 
 	private RabbitTemplate template = new RabbitTemplate();
 
@@ -40,7 +38,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 	private final boolean transactional;
 
 	@Rule
-	public static BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue);
 
 	private final int messageCount;
 
@@ -62,17 +60,6 @@ public void declareQueue() {
 		connectionFactory.setChannelCacheSize(concurrentConsumers);
 		connectionFactory.setPort(5673);
 		template.setConnectionFactory(connectionFactory);
-		RabbitAdmin admin = new RabbitAdmin(template);
-		try {
-			admin.deleteQueue("test.queue");
-		}
-		catch (AmqpIOException e) {
-			// Ignore (queue didn't exist)
-		}
-		queue = new Queue("test.queue");
-		// Idempotent, so no problem to do this for every test
-		admin.declareQueue(queue);
-		admin.purgeQueue("test.queue", false);
 	}
 
 	@Test