Add @BusConnectionFactory to qualify the preferred instance

If you are using AMQP there needs to be a ConnectionFactory (from
Spring Rabbit) in the application context. If there is a single
ConnectionFactory it will be used, or if there is a one qualified as
@BusConnectionFactory it will be preferred over others, otherwise
the @Primary one will be used. If there are multiple unqualified
connection factories there will be an error.

Fixes gh-13

docs/src/main/asciidoc/spring-cloud-bus.adoc

@@ -6,3 +6,17 @@ include::intro.adoc[]
 == Quick Start
 
 include::quickstart.adoc[]
+
+== Customizing the AMQP ConnectionFactory
+
+If you are using AMQP there needs to be a `ConnectionFactory` (from
+Spring Rabbit) in the application context. If there is a single
+`ConnectionFactory` it will be used, or if there is a one qualified as
+`@BusConnectionFactory` it will be preferred over others, otherwise
+the `@Primary` one will be used. If there are multiple unqualified
+connection factories there will be an error.
+
+Note that Spring Boot (as of 1.2.2) creates a `ConnectionFactory` that
+is _not_ `@Primary`, so if you want to use one connection factory for
+the bus and another for business messages, you need to create both,
+and annotate them `@BusConnectionFactory` and `@Primary` respectively.

spring-cloud-bus/src/main/java/org/springframework/cloud/bus/amqp/AmqpBusAutoConfiguration.java

@@ -1,8 +1,11 @@
 package org.springframework.cloud.bus.amqp;
 
-import javax.annotation.PostConstruct;
-
-import org.springframework.amqp.core.*;
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.AnonymousQueue;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
@@ -21,29 +24,49 @@ import org.springframework.messaging.MessageChannel;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
+ * Autoconfiguration for a Spring Cloud Bus on AMQP. Enabled by default if spring-rabbit
+ * is on the classpath, and can be switched off with
+ * <code>spring.cloud.bus.amqp.enabled</code>. If there is a single
+ * {@link ConnectionFactory} in the context it will be used, or if there is a one
+ * qualified as <code>@BusConnectionFactory</code> it will be preferred over others,
+ * otherwise the <code>@Primary</code> one will be used. If there are multiple unqualified
+ * connection factories there will be an autowiring error. Note that Spring Boot (as of
+ * 1.2.2) creates a ConnectionFactory that is <i>not</i> <code>@Primary</code>, so if you
+ * want to use one connection factory for the bus and another for business messages, you
+ * need to create both, and annotate them <code>@BusConnectionFactory</code> and
+ * <code>@Primary</code> respectively.
+ * 
  * @author Spencer Gibb
+ * @author Dave Syer
  */
 @Configuration
 @ConditionalOnBusEnabled
-@ConditionalOnClass(AmqpTemplate.class)
+@ConditionalOnClass({ AmqpTemplate.class, RabbitTemplate.class })
 @ConditionalOnProperty(value = "spring.cloud.bus.amqp.enabled", matchIfMissing = true)
 public class AmqpBusAutoConfiguration {
 
 	public static final String SPRING_CLOUD_BUS = "spring.cloud.bus";
 
-	@Autowired
-	private ConnectionFactory connectionFactory;
+	@Autowired(required = false)
+	@BusConnectionFactory
+	private ConnectionFactory busConnectionFactory;
+
+	@Autowired(required = false)
+	private ConnectionFactory primaryConnectionFactory;
 
-	@Autowired
 	private RabbitTemplate amqpTemplate;
 
 	@Autowired(required = false)
 	private ObjectMapper objectMapper;
 
-	@PostConstruct
-	public void init() {
-		Jackson2JsonMessageConverter converter = messageConverter();
-		amqpTemplate.setMessageConverter(converter);
+	public RabbitTemplate amqpTemplate() {
+		if (this.amqpTemplate == null) {
+			RabbitTemplate amqpTemplate = new RabbitTemplate(connectionFactory());
+			Jackson2JsonMessageConverter converter = messageConverter();
+			amqpTemplate.setMessageConverter(converter);
+			this.amqpTemplate = amqpTemplate;
+		}
+		return amqpTemplate;
 	}
 
 	@Bean
@@ -68,7 +91,7 @@ public class AmqpBusAutoConfiguration {
 			@Qualifier("cloudBusOutboundChannel") MessageChannel cloudBusOutboundChannel) {
 		return IntegrationFlows
 				.from(cloudBusOutboundChannel)
-				.handle(Amqp.outboundAdapter(this.amqpTemplate).exchangeName(
+				.handle(Amqp.outboundAdapter(amqpTemplate()).exchangeName(
 						SPRING_CLOUD_BUS)).get();
 	}
 
@@ -76,11 +99,18 @@ public class AmqpBusAutoConfiguration {
 	public IntegrationFlow cloudBusAmqpInboundFlow(
 			@Qualifier("cloudBusInboundChannel") MessageChannel cloudBusInboundChannel) {
 		return IntegrationFlows
-				.from(Amqp.inboundAdapter(connectionFactory, localCloudBusQueue())
+				.from(Amqp.inboundAdapter(connectionFactory(), localCloudBusQueue())
 						.messageConverter(messageConverter()))
 				.channel(cloudBusInboundChannel).get();
 	}
 
+	private ConnectionFactory connectionFactory() {
+		if (busConnectionFactory != null) {
+			return busConnectionFactory;
+		}
+		return primaryConnectionFactory;
+	}
+
 	private Jackson2JsonMessageConverter messageConverter() {
 		Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
 		if (objectMapper != null) {

spring-cloud-bus/src/main/java/org/springframework/cloud/bus/amqp/BusConnectionFactory.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2013-2015 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.cloud.bus.amqp;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+
+import com.rabbitmq.client.ConnectionFactory;
+
+/**
+ * Annotation to mark a bean as the {@link ConnectionFactory} for Spring Cloud Bus
+ */
+@Target({ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@Qualifier
+public @interface BusConnectionFactory {
+
+}

spring-cloud-bus/src/test/java/org/springframework/cloud/bus/amqp/AmqpBusAutoConfigurationTests.java

@@ -1,16 +1,22 @@
 package org.springframework.cloud.bus.amqp;
 
-import org.junit.Test;
-import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
-import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
-import org.springframework.boot.test.EnvironmentTestUtils;
-import org.springframework.cloud.bus.BusAutoConfiguration;
-import org.springframework.context.annotation.AnnotationConfigApplicationContext;
-
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.springframework.cloud.bus.BusAutoConfiguration.SPRING_CLOUD_BUS_ENABLED;
 
+import org.junit.Test;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
+import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
+import org.springframework.boot.test.EnvironmentTestUtils;
+import org.springframework.cloud.bus.BusAutoConfiguration;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+
 /**
  * @author Dave Syer
  */
@@ -24,6 +30,30 @@ public class AmqpBusAutoConfigurationTests {
 		context.close();
 	}
 
+	@Test
+	public void qualifiedConnectionFactory() throws Exception {
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(
+				getConfigClasses(QualifiedConnectionFactory.class));
+		assertTrue(context.containsBean("cloudBusExchange"));
+		context.close();
+	}
+
+	@Test
+	public void unqualifiedConnectionFactory() throws Exception {
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(
+				getConfigClasses(UnqualifiedConnectionFactory.class));
+		assertTrue(context.containsBean("cloudBusExchange"));
+		context.close();
+	}
+
+	@Test
+	public void twoConnectionFactories() throws Exception {
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(
+				getConfigClasses(TwoConnectionFactories.class));
+		assertTrue(context.containsBean("cloudBusExchange"));
+		context.close();
+	}
+
 	@Test
 	public void notStartedIfBusDisabled() {
 		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
@@ -36,8 +66,48 @@ public class AmqpBusAutoConfigurationTests {
 		context.close();
 	}
 
-	private Class[] getConfigClasses() {
-		return new Class[]{AmqpBusAutoConfiguration.class, RabbitAutoConfiguration.class,
+	private Class<?>[] getConfigClasses(Class<?>... extras) {
+		Class<?>[] defaults = new Class<?>[]{AmqpBusAutoConfiguration.class, RabbitAutoConfiguration.class,
 				PropertyPlaceholderAutoConfiguration.class, BusAutoConfiguration.class};
+		Class<?>[] result = new Class<?>[extras.length + defaults.length];
+		System.arraycopy(extras, 0, result, 0, extras.length);
+		System.arraycopy(defaults, 0, result, extras.length, defaults.length);
+		return result;
 	}
+	
+	@Configuration
+	protected static class UnqualifiedConnectionFactory {
+		@Bean
+		public ConnectionFactory rabbitConnectionFactory(RabbitProperties config) {
+			CachingConnectionFactory factory = new CachingConnectionFactory();
+			return factory;
+		}		
+	}
+
+	@Configuration
+	protected static class QualifiedConnectionFactory {
+		@Bean
+		@BusConnectionFactory
+		public ConnectionFactory rabbitConnectionFactory(RabbitProperties config) {
+			CachingConnectionFactory factory = new CachingConnectionFactory();
+			return factory;
+		}		
+	}
+	
+	@Configuration
+	protected static class TwoConnectionFactories {
+		@Bean
+		@BusConnectionFactory
+		public ConnectionFactory busConnectionFactory(RabbitProperties config) {
+			CachingConnectionFactory factory = new CachingConnectionFactory();
+			return factory;
+		}		
+		@Bean
+		@Primary
+		public ConnectionFactory rabbitConnectionFactory(RabbitProperties config) {
+			CachingConnectionFactory factory = new CachingConnectionFactory();
+			return factory;
+		}		
+	}
+	
 }