diff --git a/docs/src/main/asciidoc/spring-cloud-netflix.adoc b/docs/src/main/asciidoc/spring-cloud-netflix.adoc index 42bf13fd..7693ba89 100644 --- a/docs/src/main/asciidoc/spring-cloud-netflix.adoc +++ b/docs/src/main/asciidoc/spring-cloud-netflix.adoc @@ -495,6 +495,21 @@ You can then point the Hystrix Dashboard to the Turbine AMQP Server instead of i Spring Cloud provides a `spring-cloud-starter-turbine-amqp` that has all the dependencies you need to get a Turbine AMQP server running. You need Java 8 to run the app because it is Netty-based. +== 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 +`@HystrixConnectionFactory` (on the client) and +`@TurbineConnectionFactory` (on the server) 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 `@*ConnectionFactory` and `@Primary` respectively. + [[spring-cloud-ribbon]] == Client Side Load Balancer: Ribbon diff --git a/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixConnectionFactory.java b/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixConnectionFactory.java new file mode 100644 index 00000000..09aa35a5 --- /dev/null +++ b/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixConnectionFactory.java @@ -0,0 +1,38 @@ +/* + * 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.netflix.hystrix.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; + +/** + * Annotation to mark a bean as the Rabbit ConnectionFactory for Spring Cloud Hystrix + */ +@Target({ ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD }) +@Retention(RetentionPolicy.RUNTIME) +@Documented +@Inherited +@Qualifier +public @interface HystrixConnectionFactory { + +} diff --git a/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamAutoConfiguration.java b/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamAutoConfiguration.java index 6fc1c48d..43bea4f5 100644 --- a/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamAutoConfiguration.java +++ b/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamAutoConfiguration.java @@ -16,9 +16,8 @@ package org.springframework.cloud.netflix.hystrix.amqp; -import javax.annotation.PostConstruct; - import org.springframework.amqp.core.DirectExchange; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.beans.factory.annotation.Autowired; @@ -39,7 +38,20 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.netflix.hystrix.HystrixCircuitBreaker; /** + * Autoconfiguration for a Spring Cloud Hystrix 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 @HystrixConnectionFactory 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 @HystrixConnectionFactory and + * @Primary respectively. + * * @author Spencer Gibb + * @author Dave Syer */ @Configuration @ConditionalOnClass({ HystrixCircuitBreaker.class, RabbitTemplate.class }) @@ -49,16 +61,26 @@ import com.netflix.hystrix.HystrixCircuitBreaker; @EnableScheduling public class HystrixStreamAutoConfiguration { - @Autowired - private RabbitTemplate amqpTemplate; + @Autowired(required = false) + @HystrixConnectionFactory + private ConnectionFactory hystrixConnectionFactory; + + @Autowired(required = false) + private ConnectionFactory primaryConnectionFactory; @Autowired(required = false) private ObjectMapper objectMapper; - @PostConstruct - public void init() { - Jackson2JsonMessageConverter converter = messageConverter(); - this.amqpTemplate.setMessageConverter(converter); + private RabbitTemplate amqpTemplate; + + public RabbitTemplate amqpTemplate() { + if (this.amqpTemplate == null) { + RabbitTemplate amqpTemplate = new RabbitTemplate(connectionFactory()); + Jackson2JsonMessageConverter converter = messageConverter(); + amqpTemplate.setMessageConverter(converter); + this.amqpTemplate = amqpTemplate; + } + return this.amqpTemplate; } @Bean @@ -97,6 +119,13 @@ public class HystrixStreamAutoConfiguration { HystrixConstants.HYSTRIX_STREAM_NAME)).get(); } + private ConnectionFactory connectionFactory() { + if (this.hystrixConnectionFactory != null) { + return this.hystrixConnectionFactory; + } + return this.primaryConnectionFactory; + } + private Jackson2JsonMessageConverter messageConverter() { Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter(); if (this.objectMapper != null) { diff --git a/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamTask.java b/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamTask.java index 93077942..1e462f77 100644 --- a/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamTask.java +++ b/spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamTask.java @@ -22,8 +22,6 @@ import java.util.ArrayList; import java.util.Collection; import java.util.concurrent.LinkedBlockingQueue; -import com.fasterxml.jackson.core.JsonFactory; -import com.fasterxml.jackson.core.JsonGenerator; import lombok.extern.apachecommons.CommonsLog; import org.springframework.beans.BeansException; @@ -34,6 +32,8 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.scheduling.annotation.Scheduled; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; import com.netflix.hystrix.HystrixCircuitBreaker; import com.netflix.hystrix.HystrixCommandKey; import com.netflix.hystrix.HystrixCommandMetrics; @@ -44,7 +44,9 @@ import com.netflix.hystrix.util.HystrixRollingNumberEvent; /** * @author Spencer Gibb - * @see com.netflix.hystrix.contrib.metrics.eventstream.HystrixMetricsPoller.MetricsPoller + * + * @see com.netflix.hystrix.contrib.metrics.eventstream.HystrixMetricsPoller (nested + * private class MetricsPoller) */ @CommonsLog public class HystrixStreamTask implements ApplicationContextAware { @@ -113,7 +115,7 @@ public class HystrixStreamTask implements ApplicationContextAware { .getInstance(key); StringWriter jsonString = new StringWriter(); - JsonGenerator json = this.jsonFactory.createJsonGenerator(jsonString); + JsonGenerator json = this.jsonFactory.createGenerator(jsonString); json.writeStartObject(); @@ -293,7 +295,7 @@ public class HystrixStreamTask implements ApplicationContextAware { HystrixThreadPoolKey key = threadPoolMetrics.getThreadPoolKey(); StringWriter jsonString = new StringWriter(); - JsonGenerator json = this.jsonFactory.createJsonGenerator(jsonString); + JsonGenerator json = this.jsonFactory.createGenerator(jsonString); json.writeStartObject(); addServiceData(json, localService); diff --git a/spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineAmqpAutoConfiguration.java b/spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineAmqpAutoConfiguration.java index 6274c5ba..b3c3576b 100644 --- a/spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineAmqpAutoConfiguration.java +++ b/spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineAmqpAutoConfiguration.java @@ -19,8 +19,6 @@ package org.springframework.cloud.netflix.turbine.amqp; import java.util.HashMap; import java.util.Map; -import javax.annotation.PostConstruct; - import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; @@ -42,26 +40,46 @@ import org.springframework.integration.dsl.amqp.Amqp; import com.fasterxml.jackson.databind.ObjectMapper; /** + * Autoconfiguration for a Spring Cloud Turbine 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 @TurbineConnectionFactory 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 @TurbineConnectionFactory and + * @Primary respectively. + * * @author Spencer Gibb + * @author Dave Syer */ @Configuration @ConditionalOnClass(AmqpTemplate.class) @ConditionalOnProperty(value = "turbine.amqp.enabled", matchIfMissing = true) public class TurbineAmqpAutoConfiguration { - @Autowired - private ConnectionFactory connectionFactory; + @Autowired(required = false) + @TurbineConnectionFactory + private ConnectionFactory turbineConnectionFactory; - @Autowired - private RabbitTemplate amqpTemplate; + @Autowired(required = false) + private ConnectionFactory primaryConnectionFactory; @Autowired(required = false) private ObjectMapper objectMapper; - @PostConstruct - public void init() { - Jackson2JsonMessageConverter converter = messageConverter(); - this.amqpTemplate.setMessageConverter(converter); + private RabbitTemplate amqpTemplate; + + public RabbitTemplate amqpTemplate() { + if (this.amqpTemplate == null) { + RabbitTemplate amqpTemplate = new RabbitTemplate(connectionFactory()); + Jackson2JsonMessageConverter converter = messageConverter(); + amqpTemplate.setMessageConverter(converter); + this.amqpTemplate = amqpTemplate; + } + return this.amqpTemplate; } @Bean @@ -80,14 +98,15 @@ public class TurbineAmqpAutoConfiguration { public Queue hystrixStreamQueue() { Map args = new HashMap<>(); args.put("x-message-ttl", 60000); // TODO: configure TTL - Queue queue = new Queue(HystrixConstants.HYSTRIX_STREAM_NAME, false, false, false, args); + Queue queue = new Queue(HystrixConstants.HYSTRIX_STREAM_NAME, false, false, + false, args); return queue; } @Bean public IntegrationFlow hystrixStreamAggregatorInboundFlow() { return IntegrationFlows - .from(Amqp.inboundAdapter(this.connectionFactory, hystrixStreamQueue()) + .from(Amqp.inboundAdapter(connectionFactory(), hystrixStreamQueue()) .messageConverter(messageConverter())) .channel("hystrixStreamAggregator").get(); } @@ -97,6 +116,13 @@ public class TurbineAmqpAutoConfiguration { return new Aggregator(); } + private ConnectionFactory connectionFactory() { + if (this.turbineConnectionFactory != null) { + return this.turbineConnectionFactory; + } + return this.primaryConnectionFactory; + } + private Jackson2JsonMessageConverter messageConverter() { Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter(); if (this.objectMapper != null) { diff --git a/spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineConnectionFactory.java b/spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineConnectionFactory.java new file mode 100644 index 00000000..a429d96e --- /dev/null +++ b/spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineConnectionFactory.java @@ -0,0 +1,38 @@ +/* + * 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.netflix.turbine.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; + +/** + * Annotation to mark a bean as the Rabbit ConnectionFactory for Spring Cloud Turbine + */ +@Target({ ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD }) +@Retention(RetentionPolicy.RUNTIME) +@Documented +@Inherited +@Qualifier +public @interface TurbineConnectionFactory { + +}