Use JSON for AMQP Hystrix data

spring-cloud-netflix-hystrix-amqp/pom.xml

@@ -48,6 +48,10 @@
             <optional>true</optional>				
         </dependency>
         <dependency>
+        	<groupId>com.fasterxml.jackson.core</groupId>
+        	<artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
             <groupId>com.netflix.hystrix</groupId>
             <artifactId>hystrix-core</artifactId>
         </dependency>

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

 package org.springframework.netflix.hystrix.amqp;
 
-import com.netflix.hystrix.HystrixCircuitBreaker;
-import org.springframework.amqp.core.AmqpTemplate;
+import javax.annotation.PostConstruct;
+
 import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
@@ -16,50 +18,68 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.amqp.Amqp;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.netflix.hystrix.HystrixCircuitBreaker;
+
 /**
  * @author Spencer Gibb
  */
 @Configuration
-@ConditionalOnClass({HystrixCircuitBreaker.class, AmqpTemplate.class})
+@ConditionalOnClass({ HystrixCircuitBreaker.class, RabbitTemplate.class })
+@ConditionalOnExpression("${hystrix.stream.amqp.enabled:true}")
+@IntegrationComponentScan(basePackageClasses = HystrixStreamChannel.class)
+@EnableScheduling
 public class HystrixStreamAutoConfiguration {
-    @Configuration
-    @ConditionalOnExpression("${hystrix.stream.amqp.enabled:true}")
-    @IntegrationComponentScan(basePackageClasses = HystrixStreamChannel.class)
-    @EnableScheduling
-    protected static class HystrixStreamAmqpAutoConfiguration {
 
-        @Autowired
-        private AmqpTemplate amqpTemplate;
+	@Autowired
+	private RabbitTemplate amqpTemplate;
+
+	@Autowired(required = false)
+	private ObjectMapper objectMapper;
+
+	@PostConstruct
+	public void init() {
+		Jackson2JsonMessageConverter converter = messageConverter();
+		amqpTemplate.setMessageConverter(converter);
+	}
+
+	@Bean
+	public HystrixStreamTask hystrixStreamTask() {
+		return new HystrixStreamTask();
+	}
+
+	@Bean
+	public DirectChannel hystrixStream() {
+		return new DirectChannel();
+	}
 
-        @Bean
-        public HystrixStreamTask hystrixStreamTask() {
-            return new HystrixStreamTask();
-        }
+	@Bean
+	public DirectExchange hystrixStreamExchange() {
+		DirectExchange exchange = new DirectExchange(Constants.HYSTRIX_STREAM_NAME);
+		return exchange;
+	}
 
-        @Bean
-        public DirectChannel hystrixStream() {
-            return new DirectChannel();
-        }
-        
-        @Bean
-        public DirectExchange hystrixStreamExchange() {
-            DirectExchange exchange = new DirectExchange(Constants.HYSTRIX_STREAM_NAME);
-            return exchange;
-        }
+	@Bean
+	public IntegrationFlow hystrixStreamOutboundFlow() {
+		return IntegrationFlows
+				.from("hystrixStream")
+				// TODO: set content type
+				/*
+				 * .enrichHeaders(new ComponentConfigurer<HeaderEnricherSpec>() {
+				 * 
+				 * @Override public void configure(HeaderEnricherSpec spec) {
+				 * spec.header("content-type", "application/json", true); } })
+				 */
+				.handle(Amqp.outboundAdapter(this.amqpTemplate).exchangeName(
+						Constants.HYSTRIX_STREAM_NAME)).get();
+	}
 
-        @Bean
-        public IntegrationFlow hystrixStreamOutboundFlow() {
-            return IntegrationFlows.from("hystrixStream")
-                    //TODO: set content type
-                    /*.enrichHeaders(new ComponentConfigurer<HeaderEnricherSpec>() {
-                        @Override
-                        public void configure(HeaderEnricherSpec spec) {
-                            spec.header("content-type", "application/json", true);
-                        }
-                    })*/
-                    .handle(Amqp.outboundAdapter(this.amqpTemplate).exchangeName(Constants.HYSTRIX_STREAM_NAME))
-                    .get();
-        }
-    }
+	private Jackson2JsonMessageConverter messageConverter() {
+		Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
+		if (objectMapper != null) {
+			converter.setJsonObjectMapper(objectMapper);
+		}
+		return converter;
+	}
 
 }

spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/netflix/turbine/amqp/Aggregator.java → spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/Aggregator.java

-package org.springframework.netflix.turbine.amqp;
+package org.springframework.cloud.netflix.turbine.amqp;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 

spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/netflix/turbine/amqp/EnableTurbineAmqp.java → spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/EnableTurbineAmqp.java

-package org.springframework.netflix.turbine.amqp;
+package org.springframework.cloud.netflix.turbine.amqp;
 
 import org.springframework.context.annotation.Import;
 

spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/netflix/turbine/amqp/TurbineAmqpAutoConfiguration.java → spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineAmqpAutoConfiguration.java

-package org.springframework.netflix.turbine.amqp;
+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;
 import org.springframework.amqp.core.DirectExchange;
 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;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
@@ -19,52 +23,70 @@ import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.amqp.Amqp;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 /**
  * @author Spencer Gibb
  */
 @Configuration
 @ConditionalOnClass(AmqpTemplate.class)
+@ConditionalOnExpression("${turbine.amqp.enabled:true}")
 public class TurbineAmqpAutoConfiguration {
 
-    @Configuration
-    @ConditionalOnExpression("${turbine.amqp.enabled:true}")
-    protected static class HystrixStreamAggregatorAutoConfiguration {
-
-        @Autowired
-        private ConnectionFactory connectionFactory;
-
-        //TODO: how to fail gracefully if no rabbit?
-        @Bean
-        public DirectExchange hystrixStreamExchange() {
-            DirectExchange exchange = new DirectExchange(Constants.HYSTRIX_STREAM_NAME);
-            return exchange;
-        }
-
-    	@Bean
-    	protected Binding localTurbineAmqpQueueBinding() {
-    		return BindingBuilder.bind(hystrixStreamQueue()).to(hystrixStreamExchange()).with("");		
-    	}
-
-        @Bean
-        public Queue hystrixStreamQueue() {
-            Map<String, Object> args = new HashMap<>();
-            args.put("x-message-ttl", 60000); //TODO: configure TTL
-            Queue queue = new Queue(Constants.HYSTRIX_STREAM_NAME, false, false, false, args);
-            return queue;
-        }
-
-        @Bean
-        public IntegrationFlow hystrixStreamAggregatorInboundFlow() {
-            return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, hystrixStreamQueue()))
-                    .channel("hystrixStreamAggregator")
-                    .get();
-        }
-
-        @Bean
-        public Aggregator hystrixStreamAggregator() {
-            return new Aggregator();
-        }
-    }
+	@Autowired
+	private ConnectionFactory connectionFactory;
+
+	@Autowired
+	private RabbitTemplate amqpTemplate;
+
+	@Autowired(required = false)
+	private ObjectMapper objectMapper;
+
+	@PostConstruct
+	public void init() {
+		Jackson2JsonMessageConverter converter = messageConverter();
+		amqpTemplate.setMessageConverter(converter);
+	}
+
+	@Bean
+	public DirectExchange hystrixStreamExchange() {
+		DirectExchange exchange = new DirectExchange(Constants.HYSTRIX_STREAM_NAME);
+		return exchange;
+	}
+
+	@Bean
+	protected Binding localTurbineAmqpQueueBinding() {
+		return BindingBuilder.bind(hystrixStreamQueue()).to(hystrixStreamExchange())
+				.with("");
+	}
+
+	@Bean
+	public Queue hystrixStreamQueue() {
+		Map<String, Object> args = new HashMap<>();
+		args.put("x-message-ttl", 60000); // TODO: configure TTL
+		Queue queue = new Queue(Constants.HYSTRIX_STREAM_NAME, false, false, false, args);
+		return queue;
+	}
+
+	@Bean
+	public IntegrationFlow hystrixStreamAggregatorInboundFlow() {
+		return IntegrationFlows
+				.from(Amqp.inboundAdapter(connectionFactory, hystrixStreamQueue())
+						.messageConverter(messageConverter()))
+				.channel("hystrixStreamAggregator").get();
+	}
+
+	@Bean
+	public Aggregator hystrixStreamAggregator() {
+		return new Aggregator();
+	}
 
+	private Jackson2JsonMessageConverter messageConverter() {
+		Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
+		if (objectMapper != null) {
+			converter.setJsonObjectMapper(objectMapper);
+		}
+		return converter;
+	}
 
 }

spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/netflix/turbine/amqp/TurbineAmqpConfiguration.java → spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineAmqpConfiguration.java

-package org.springframework.netflix.turbine.amqp;
+package org.springframework.cloud.netflix.turbine.amqp;
 
-import com.netflix.turbine.aggregator.InstanceKey;
-import com.netflix.turbine.aggregator.StreamAggregator;
-import com.netflix.turbine.internal.JsonUtility;
+import static io.reactivex.netty.pipeline.PipelineConfigurators.sseServerConfigurator;
 import io.netty.buffer.ByteBuf;
 import io.reactivex.netty.RxNetty;
 import io.reactivex.netty.protocol.http.server.HttpServer;
 import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+
+import java.util.Map;
+
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+
 import rx.Observable;
 import rx.subjects.PublishSubject;
 
-import java.util.Map;
-
-import static io.reactivex.netty.pipeline.PipelineConfigurators.sseServerConfigurator;
+import com.netflix.turbine.aggregator.InstanceKey;
+import com.netflix.turbine.aggregator.StreamAggregator;
+import com.netflix.turbine.internal.JsonUtility;
 
 /**
  * @author Spencer Gibb
  */
 @Configuration
 @Slf4j
-@ConfigurationProperties("turbine.amqp")
+@EnableConfigurationProperties(TurbineAmqpProperties.class)
 public class TurbineAmqpConfiguration implements SmartLifecycle {
 
 	private boolean running = false;
 
-	private int port = 8989;
+	@Autowired
+	private TurbineAmqpProperties turbine;
 
 	@Bean
 	public PublishSubject<Map<String, Object>> hystrixSubject() {
@@ -39,19 +44,26 @@ public class TurbineAmqpConfiguration implements SmartLifecycle {
 	@Bean
 	public HttpServer<ByteBuf, ServerSentEvent> aggregatorServer() {
 		// multicast so multiple concurrent subscribers get the same stream
-		Observable<Map<String, Object>> publishedStreams = StreamAggregator.aggregateGroupedStreams(hystrixSubject()
-				.groupBy(data -> InstanceKey.create((String) data.get("instanceId"))))
-				.doOnUnsubscribe(() -> log.info("AmqpTurbine => Unsubscribing aggregation."))
-				.doOnSubscribe(() -> log.info("AmqpTurbine => Starting aggregation"))
-				.flatMap(o -> o).publish().refCount();
-
-		HttpServer<ByteBuf, ServerSentEvent> httpServer = RxNetty.createHttpServer(port, (request, response) -> {
-			log.info("AmqpTurbine => SSE Request Received");
-			response.getHeaders().setHeader("Content-Type", "text/event-stream");
-			return publishedStreams
-					.doOnUnsubscribe(() -> log.info("AmqpTurbine => Unsubscribing RxNetty server connection"))
-					.flatMap(data -> response.writeAndFlush(new ServerSentEvent(null, null, JsonUtility.mapToJson(data))));
-		}, sseServerConfigurator());
+		Observable<Map<String, Object>> publishedStreams = StreamAggregator
+				.aggregateGroupedStreams(
+						hystrixSubject().groupBy(
+								data -> InstanceKey.create((String) data
+										.get("instanceId"))))
+				.doOnUnsubscribe(() -> log.info("Unsubscribing aggregation."))
+				.doOnSubscribe(() -> log.info("Starting aggregation")).flatMap(o -> o)
+				.publish().refCount();
+
+		HttpServer<ByteBuf, ServerSentEvent> httpServer = RxNetty.createHttpServer(
+				turbine.getPort(),
+				(request, response) -> {
+					log.info("SSE Request Received");
+					response.getHeaders().setHeader("Content-Type", "text/event-stream");
+					return publishedStreams.doOnUnsubscribe(
+							() -> log.info("Unsubscribing RxNetty server connection"))
+							.flatMap(
+									data -> response.writeAndFlush(new ServerSentEvent(
+											null, null, JsonUtility.mapToJson(data))));
+				}, sseServerConfigurator());
 		return httpServer;
 	}
 
@@ -75,7 +87,8 @@ public class TurbineAmqpConfiguration implements SmartLifecycle {
 	public void stop() {
 		try {
 			aggregatorServer().shutdown();
-		} catch (InterruptedException e) {
+		}
+		catch (InterruptedException e) {
 			log.error("Error shutting down", e);
 		}
 		running = false;

spring-cloud-netflix-turbine-amqp/src/main/java/org/springframework/cloud/netflix/turbine/amqp/TurbineAmqpProperties.java

+package org.springframework.cloud.netflix.turbine.amqp;
+
+import lombok.Data;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * @author Dave Syer
+ */
+@ConfigurationProperties("turbine.amqp")
+@Data
+public class TurbineAmqpProperties {
+
+	private int port = 8989;
+
+}

spring-cloud-netflix-turbine-amqp/src/main/resources/META-INF/spring.factories

 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-org.springframework.netflix.turbine.amqp.TurbineAmqpAutoConfiguration
+org.springframework.cloud.netflix.turbine.amqp.TurbineAmqpAutoConfiguration

spring-cloud-netflix-turbine-amqp/src/test/java/org/springframework/netflix/turbine/amqp/AggregatorTest.java

@@ -12,8 +12,7 @@ import com.netflix.turbine.internal.JsonUtility;
 
 import rx.Observable;
 import rx.observables.GroupedObservable;
-
-import static org.springframework.netflix.turbine.amqp.Aggregator.getPayloadData;
+import static org.springframework.cloud.netflix.turbine.amqp.Aggregator.getPayloadData;
 
 public class AggregatorTest {
 

spring-cloud-netflix-turbine-amqp/src/test/java/org/springframework/netflix/turbine/amqp/SampleTurbineAmqpApplication.java

@@ -2,6 +2,7 @@ package org.springframework.netflix.turbine.amqp;
 
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.netflix.turbine.amqp.EnableTurbineAmqp;
 
 /**
  * @author Spencer Gibb