INT-3282: Add BrokerRunning Rule

JIRA: https://jira.springsource.org/browse/INT-3282

Remove `@Ignore` from integration tests and some polishing

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java

@@ -21,13 +21,14 @@ import java.util.Collection;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Ignore;
+import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.integration.amqp.rule.BrokerRunning;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
@@ -45,6 +46,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @RunWith(SpringJUnit4ClassRunner.class)
 public class ChannelTests {
 
+	@ClassRule
+	public static final BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+
 	@Autowired
 	private PublishSubscribeAmqpChannel channel;
 
@@ -55,7 +59,6 @@ public class ChannelTests {
 	private ConfigurableApplicationContext context;
 
 	@Test
-	@Ignore
 	public void pubSubLostConnectionTest() throws Exception {
 		final CyclicBarrier latch = new CyclicBarrier(2);
 		channel.subscribe(new MessageHandler() {

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml

@@ -16,24 +16,21 @@
 
 	<rabbit:direct-exchange name="si.test.exchange">
 		<rabbit:bindings>
-			<rabbit:binding queue="si.test.queue" key="si.test.binding"></rabbit:binding>
+			<rabbit:binding queue="si.test.queue" key="si.test.binding"/>
 		</rabbit:bindings>
 	</rabbit:direct-exchange>
 
-	<console:stdout-channel-adapter id="fromRabbit" append-newline="true"/>
-
-	<console:stdin-channel-adapter channel="toRabbit">
-		<poller fixed-delay="1000" max-messages-per-poll="1"/>
-	</console:stdin-channel-adapter>
-
 	<channel id="toRabbit"/>
 
+	<channel id="fromRabbit">
+		<queue/>
+	</channel>
+
 	<amqp:outbound-gateway request-channel="toRabbit"
 		reply-channel="fromRabbit"
 		exchange-name="si.test.exchange"
 		routing-key="si.test.binding"
-		amqp-template="amqpTemplate"
-		order="1"/>
+		amqp-template="amqpTemplate"/>
 
 	<amqp:inbound-gateway request-channel="amqpIn"
 		connection-factory="connectionFactory"

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2014 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,29 +16,47 @@
 
 package org.springframework.integration.amqp.config;
 
-import org.junit.Ignore;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
-import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.amqp.rule.BrokerRunning;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Mark Fisher
+ * @author Artem Bilan
  * @since 2.1
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
 public class OutboundGatewayIntegrationTests {
 
+	@ClassRule
+	public static final BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues("si.test.queue");
+
+	@Autowired
+	private MessageChannel toRabbit;
+
+	@Autowired
+	private PollableChannel fromRabbit;
+
 	@Test
-	@Ignore // comment this out to test when a RabbitMQ broker is available
-	public void run() throws Exception {
-		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
-		RabbitTemplate template = new RabbitTemplate(connectionFactory);
-		template.convertAndSend("si.test.exchange", "si.test.binding", "foo");
+	public void testOutboundInboundGateways() throws Exception {
+		String payload = "foo";
+		this.toRabbit.send(new GenericMessage<String>(payload));
+		Message<?> receive = this.fromRabbit.receive(1000);
+		assertNotNull(receive);
+		assertEquals(payload.toUpperCase(), receive.getPayload());
 	}
 
 

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.amqp.rule;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Assume;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.util.Assert;
+
+/**
+ * <p>
+ * A rule that prevents integration tests from failing if the Rabbit broker application is not running or not
+ * accessible. If the Rabbit broker is not running in the background all the tests here will simply be skipped because
+ * of a violated assumption (showing as successful).
+ * <p>
+ * The rule can be declared as static so that it only has to check once for all tests in the enclosing test case, but
+ * there isn't a lot of overhead in making it non-static.
+ *
+ * @author Dave Syer
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class BrokerRunning extends TestWatcher {
+
+	public static final int PORT = 5672;
+
+	private static final Log logger = LogFactory.getLog(BrokerRunning.class);
+
+	private static final Queue DEFAULT_QUEUE_NAME = new Queue(BrokerRunning.class.getName());
+
+	private final Queue[] queues;
+
+	/**
+	 * Ensure the broker is running and has an empty queue (which can be addressed via the default exchange).
+	 *
+	 * @return a new rule that assumes an existing running broker
+	 */
+	public static BrokerRunning isRunningWithEmptyQueues(Queue... queues) {
+		Assert.notNull(queues);
+		Assert.noNullElements(queues);
+		return new BrokerRunning(queues);
+	}
+
+	/**
+	 * Ensure the broker is running and has an empty queue (which can be addressed via the default exchange).
+	 *
+	 * @return a new rule that assumes an existing running broker
+	 */
+	public static BrokerRunning isRunningWithEmptyQueues(String... queues) {
+		Assert.notNull(queues);
+		Assert.noNullElements(queues);
+		return new BrokerRunning(queues);
+	}
+
+	/**
+	 * @return a new rule that assumes an existing running broker
+	 */
+	public static BrokerRunning isRunning() {
+		return new BrokerRunning(DEFAULT_QUEUE_NAME);
+	}
+
+
+	private BrokerRunning(Queue... queues) {
+		this.queues = queues;
+	}
+
+	private BrokerRunning(String... queues) {
+		List<Queue> queueList = new ArrayList<Queue>(queues.length);
+		for (String queue : queues) {
+			queueList.add(new Queue(queue));
+		}
+		this.queues = queueList.toArray(new Queue[queues.length]);
+	}
+
+	@Override
+	public Statement apply(Statement base, Description description) {
+		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
+
+		try {
+
+			connectionFactory.setPort(PORT);
+
+			RabbitAdmin admin = new RabbitAdmin(connectionFactory);
+
+			for (Queue queue : queues) {
+				String queueName = queue.getName();
+				logger.info("Deleting queue: " + queueName);
+				// Delete completely - gets rid of consumers and bindings as well
+				admin.deleteQueue(queueName);
+
+				if (!DEFAULT_QUEUE_NAME.getName().equals(queueName)) {
+					admin.declareQueue(queue);
+				}
+			}
+
+
+		}
+		catch (final Exception e) {
+			logger.warn("Not executing tests because basic connectivity test failed", e);
+			Assume.assumeNoException(e);
+		}
+		finally {
+			connectionFactory.destroy();
+		}
+
+		return super.apply(base, description);
+	}
+
+}