diff --git a/docs/src/main/asciidoc/spring-cloud-bus.adoc b/docs/src/main/asciidoc/spring-cloud-bus.adoc index 6011879..15a36a2 100644 --- a/docs/src/main/asciidoc/spring-cloud-bus.adoc +++ b/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. \ No newline at end of file diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/amqp/AmqpBusAutoConfiguration.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/amqp/AmqpBusAutoConfiguration.java index e1c5d5c..406f7f2 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/amqp/AmqpBusAutoConfiguration.java +++ b/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 + * spring.cloud.bus.amqp.enabled. If there is a single + * {@link ConnectionFactory} in the context 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 autowiring 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. + * * @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) { diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/amqp/BusConnectionFactory.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/amqp/BusConnectionFactory.java new file mode 100644 index 0000000..551c437 --- /dev/null +++ b/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 { + +} diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/amqp/AmqpBusAutoConfigurationTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/amqp/AmqpBusAutoConfigurationTests.java index 3256078..e827434 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/amqp/AmqpBusAutoConfigurationTests.java +++ b/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; + } + } + }