SpringCloud/spring-cloud-openfeign
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add @*ConnectionFactory to qualify the preferred instance

Browse Source 
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
@[Hystrix,Turbine]ConnectionFactory 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.

See https://github.com/spring-cloud/spring-cloud-bus/issues/13
This commit is contained in:
Dave Syer
2015-02-27 12:27:04 +00:00
parent a954e88fc6
commit 8ac43cca52
6 changed files with 173 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
docs/src/main/asciidoc/spring-cloud-netflix.adoc
View File

@@ -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

38
spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixConnectionFactory.java Normal file
View File

@@ -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 {
}

45
spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamAutoConfiguration.java
View File

@@ -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
* <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>@HystrixConnectionFactory</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>@HystrixConnectionFactory</code> and
* <code>@Primary</code> 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) {

12
spring-cloud-netflix-hystrix-amqp/src/main/java/org/springframework/cloud/netflix/hystrix/amqp/HystrixStreamTask.java
View File

@@ -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);

50
spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineAmqpAutoConfiguration.java
View File

@@ -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
* <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>@TurbineConnectionFactory</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>@TurbineConnectionFactory</code> and
* <code>@Primary</code> 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<String, Object> 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) {

38
spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineConnectionFactory.java Normal file
View File

@@ -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 {
}