Ensure content type is application/json

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

@@ -33,17 +33,12 @@ import org.springframework.scheduling.annotation.EnableScheduling;
 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.
+ * Autoconfiguration for a Spring Cloud Hystrix on Spring Cloud Stream. Enabled by default
+ * if spring-cloud-stream is on the classpath, and can be switched off with
+ * <code>hystrix.stream.queue.enabled</code>. There are some high level configuration
+ * options in {@link HystrixStreamProperties}. The binding name for Spring Cloud Stream is
+ * {@link HystrixStreamClient#OUTPUT} so you can configure stream other properties through
+ * that.
  *
  * @author Spencer Gibb
  * @author Dave Syer
@@ -64,20 +59,26 @@ public class HystrixStreamAutoConfiguration {
 
 	@Bean
 	public HasFeatures hystrixStreamQueueFeature() {
-		return HasFeatures.namedFeature("Hystrix Stream (Queue)", HystrixStreamAutoConfiguration.class);
+		return HasFeatures.namedFeature("Hystrix Stream (Queue)",
+				HystrixStreamAutoConfiguration.class);
 	}
 
 	@PostConstruct
 	public void init() {
-		BindingProperties outputBinding = this.bindings.getBindings().get(HystrixStreamClient.OUTPUT);
+		BindingProperties outputBinding = this.bindings.getBindings()
+				.get(HystrixStreamClient.OUTPUT);
 		if (outputBinding == null) {
 			this.bindings.getBindings().put(HystrixStreamClient.OUTPUT,
 					new BindingProperties());
 		}
-		BindingProperties output = this.bindings.getBindings().get(HystrixStreamClient.OUTPUT);
+		BindingProperties output = this.bindings.getBindings()
+				.get(HystrixStreamClient.OUTPUT);
 		if (output.getDestination() == null) {
 			output.setDestination(this.properties.getDestination());
 		}
+		if (output.getContentType() == null) {
+			output.setContentType(this.properties.getContentType());
+		}
 	}
 
 	@Bean

spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamProperties.java

@@ -16,11 +16,11 @@
 
 package org.springframework.cloud.netflix.hystrix.stream;
 
-import lombok.Data;
-
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cloud.netflix.hystrix.HystrixConstants;
 
+import lombok.Data;
+
 /**
  * @author Spencer Gibb
  */
@@ -36,4 +36,6 @@ public class HystrixStreamProperties {
 
 	private String destination = HystrixConstants.HYSTRIX_STREAM_DESTINATION;
 
+	private String contentType = "application/json";
+
 }

spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamTask.java

@@ -22,8 +22,6 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.concurrent.LinkedBlockingQueue;
 
-import lombok.extern.apachecommons.CommonsLog;
-
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.client.ServiceInstance;
@@ -32,6 +30,7 @@ import org.springframework.cloud.stream.annotation.Output;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.scheduling.annotation.Scheduled;
 
@@ -45,6 +44,8 @@ import com.netflix.hystrix.HystrixThreadPoolKey;
 import com.netflix.hystrix.HystrixThreadPoolMetrics;
 import com.netflix.hystrix.util.HystrixRollingNumberEvent;
 
+import lombok.extern.apachecommons.CommonsLog;
+
 /**
  * @author Spencer Gibb
  *
@@ -90,7 +91,11 @@ public class HystrixStreamTask implements ApplicationContextAware {
 			for (String json : metrics) {
 				// TODO: batch all metrics to one message
 				try {
-					this.outboundChannel.send(MessageBuilder.withPayload(json).build());
+					// TODO: remove the explicit content type when s-c-stream can handle that for us
+					this.outboundChannel.send(MessageBuilder.withPayload(json)
+							.setHeader(MessageHeaders.CONTENT_TYPE,
+									"application/json")
+							.build());
 				}
 				catch (Exception ex) {
 					if (log.isTraceEnabled()) {
@@ -157,8 +162,8 @@ public class HystrixStreamTask implements ApplicationContextAware {
 						.getRollingCount(HystrixRollingNumberEvent.COLLAPSED));
 				json.writeNumberField("rollingCountExceptionsThrown", commandMetrics
 						.getRollingCount(HystrixRollingNumberEvent.EXCEPTION_THROWN));
-				json.writeNumberField("rollingCountFailure",
-						commandMetrics.getRollingCount(HystrixRollingNumberEvent.FAILURE));
+				json.writeNumberField("rollingCountFailure", commandMetrics
+						.getRollingCount(HystrixRollingNumberEvent.FAILURE));
 				json.writeNumberField("rollingCountFallbackFailure", commandMetrics
 						.getRollingCount(HystrixRollingNumberEvent.FALLBACK_FAILURE));
 				json.writeNumberField("rollingCountFallbackRejection", commandMetrics
@@ -171,12 +176,12 @@ public class HystrixStreamTask implements ApplicationContextAware {
 						.getRollingCount(HystrixRollingNumberEvent.SEMAPHORE_REJECTED));
 				json.writeNumberField("rollingCountShortCircuited", commandMetrics
 						.getRollingCount(HystrixRollingNumberEvent.SHORT_CIRCUITED));
-				json.writeNumberField("rollingCountSuccess",
-						commandMetrics.getRollingCount(HystrixRollingNumberEvent.SUCCESS));
+				json.writeNumberField("rollingCountSuccess", commandMetrics
+						.getRollingCount(HystrixRollingNumberEvent.SUCCESS));
 				json.writeNumberField("rollingCountThreadPoolRejected", commandMetrics
 						.getRollingCount(HystrixRollingNumberEvent.THREAD_POOL_REJECTED));
-				json.writeNumberField("rollingCountTimeout",
-						commandMetrics.getRollingCount(HystrixRollingNumberEvent.TIMEOUT));
+				json.writeNumberField("rollingCountTimeout", commandMetrics
+						.getRollingCount(HystrixRollingNumberEvent.TIMEOUT));
 
 				json.writeNumberField("currentConcurrentExecutionCount",
 						commandMetrics.getCurrentConcurrentExecutionCount());
@@ -186,12 +191,18 @@ public class HystrixStreamTask implements ApplicationContextAware {
 						commandMetrics.getExecutionTimeMean());
 				json.writeObjectFieldStart("latencyExecute");
 				json.writeNumberField("0", commandMetrics.getExecutionTimePercentile(0));
-				json.writeNumberField("25", commandMetrics.getExecutionTimePercentile(25));
-				json.writeNumberField("50", commandMetrics.getExecutionTimePercentile(50));
-				json.writeNumberField("75", commandMetrics.getExecutionTimePercentile(75));
-				json.writeNumberField("90", commandMetrics.getExecutionTimePercentile(90));
-				json.writeNumberField("95", commandMetrics.getExecutionTimePercentile(95));
-				json.writeNumberField("99", commandMetrics.getExecutionTimePercentile(99));
+				json.writeNumberField("25",
+						commandMetrics.getExecutionTimePercentile(25));
+				json.writeNumberField("50",
+						commandMetrics.getExecutionTimePercentile(50));
+				json.writeNumberField("75",
+						commandMetrics.getExecutionTimePercentile(75));
+				json.writeNumberField("90",
+						commandMetrics.getExecutionTimePercentile(90));
+				json.writeNumberField("95",
+						commandMetrics.getExecutionTimePercentile(95));
+				json.writeNumberField("99",
+						commandMetrics.getExecutionTimePercentile(99));
 				json.writeNumberField("99.5",
 						commandMetrics.getExecutionTimePercentile(99.5));
 				json.writeNumberField("100",
@@ -208,7 +219,8 @@ public class HystrixStreamTask implements ApplicationContextAware {
 				json.writeNumberField("90", commandMetrics.getTotalTimePercentile(90));
 				json.writeNumberField("95", commandMetrics.getTotalTimePercentile(95));
 				json.writeNumberField("99", commandMetrics.getTotalTimePercentile(99));
-				json.writeNumberField("99.5", commandMetrics.getTotalTimePercentile(99.5));
+				json.writeNumberField("99.5",
+						commandMetrics.getTotalTimePercentile(99.5));
 				json.writeNumberField("100", commandMetrics.getTotalTimePercentile(100));
 				json.writeEndObject();
 
@@ -222,7 +234,8 @@ public class HystrixStreamTask implements ApplicationContextAware {
 						commandProperties.circuitBreakerRequestVolumeThreshold().get());
 				json.writeNumberField(
 						"propertyValue_circuitBreakerSleepWindowInMilliseconds",
-						commandProperties.circuitBreakerSleepWindowInMilliseconds().get());
+						commandProperties.circuitBreakerSleepWindowInMilliseconds()
+								.get());
 				json.writeNumberField(
 						"propertyValue_circuitBreakerErrorThresholdPercentage",
 						commandProperties.circuitBreakerErrorThresholdPercentage().get());
@@ -245,11 +258,13 @@ public class HystrixStreamTask implements ApplicationContextAware {
 								.get());
 				json.writeStringField(
 						"propertyValue_executionIsolationThreadPoolKeyOverride",
-						commandProperties.executionIsolationThreadPoolKeyOverride().get());
+						commandProperties.executionIsolationThreadPoolKeyOverride()
+								.get());
 				json.writeNumberField(
 						"propertyValue_executionIsolationSemaphoreMaxConcurrentRequests",
 						commandProperties
-								.executionIsolationSemaphoreMaxConcurrentRequests().get());
+								.executionIsolationSemaphoreMaxConcurrentRequests()
+								.get());
 				json.writeNumberField(
 						"propertyValue_fallbackIsolationSemaphoreMaxConcurrentRequests",
 						commandProperties
@@ -309,22 +324,22 @@ public class HystrixStreamTask implements ApplicationContextAware {
 				json.writeStringField("name", key.name());
 				json.writeNumberField("currentTime", System.currentTimeMillis());
 
-				json.writeNumberField("currentActiveCount", threadPoolMetrics
-						.getCurrentActiveCount().intValue());
-				json.writeNumberField("currentCompletedTaskCount", threadPoolMetrics
-						.getCurrentCompletedTaskCount().longValue());
-				json.writeNumberField("currentCorePoolSize", threadPoolMetrics
-						.getCurrentCorePoolSize().intValue());
-				json.writeNumberField("currentLargestPoolSize", threadPoolMetrics
-						.getCurrentLargestPoolSize().intValue());
-				json.writeNumberField("currentMaximumPoolSize", threadPoolMetrics
-						.getCurrentMaximumPoolSize().intValue());
-				json.writeNumberField("currentPoolSize", threadPoolMetrics
-						.getCurrentPoolSize().intValue());
-				json.writeNumberField("currentQueueSize", threadPoolMetrics
-						.getCurrentQueueSize().intValue());
-				json.writeNumberField("currentTaskCount", threadPoolMetrics
-						.getCurrentTaskCount().longValue());
+				json.writeNumberField("currentActiveCount",
+						threadPoolMetrics.getCurrentActiveCount().intValue());
+				json.writeNumberField("currentCompletedTaskCount",
+						threadPoolMetrics.getCurrentCompletedTaskCount().longValue());
+				json.writeNumberField("currentCorePoolSize",
+						threadPoolMetrics.getCurrentCorePoolSize().intValue());
+				json.writeNumberField("currentLargestPoolSize",
+						threadPoolMetrics.getCurrentLargestPoolSize().intValue());
+				json.writeNumberField("currentMaximumPoolSize",
+						threadPoolMetrics.getCurrentMaximumPoolSize().intValue());
+				json.writeNumberField("currentPoolSize",
+						threadPoolMetrics.getCurrentPoolSize().intValue());
+				json.writeNumberField("currentQueueSize",
+						threadPoolMetrics.getCurrentQueueSize().intValue());
+				json.writeNumberField("currentTaskCount",
+						threadPoolMetrics.getCurrentTaskCount().longValue());
 				json.writeNumberField("rollingCountThreadsExecuted",
 						threadPoolMetrics.getRollingCountThreadsExecuted());
 				json.writeNumberField("rollingMaxActiveThreads",

spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamAutoConfiguration.java

@@ -28,19 +28,18 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 /**
- * Autoconfiguration for a Spring Cloud Turbine using Spring Cloud Stream.
- * Enabled by default if spring-cloud-stream is on the classpath, and can be
- * switched off with <code>turbine.stream.enabled</code>.
+ * Autoconfiguration for a Spring Cloud Turbine using Spring Cloud Stream. Enabled by
+ * default if spring-cloud-stream is on the classpath, and can be switched off with
+ * <code>turbine.stream.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 turbine and another for business messages, you
- * need to create both, and annotate them <code>@TurbineConnectionFactory</code> and
- * <code>@Primary</code> respectively.
+ * 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 turbine 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
@@ -59,14 +58,19 @@ public class TurbineStreamAutoConfiguration {
 
 	@PostConstruct
 	public void init() {
-		BindingProperties inputBinding = this.bindings.getBindings().get(TurbineStreamClient.INPUT);
+		BindingProperties inputBinding = this.bindings.getBindings()
+				.get(TurbineStreamClient.INPUT);
 		if (inputBinding == null) {
 			this.bindings.getBindings().put(TurbineStreamClient.INPUT,
 					new BindingProperties());
 		}
-		BindingProperties input = this.bindings.getBindings().get(TurbineStreamClient.INPUT);
+		BindingProperties input = this.bindings.getBindings()
+				.get(TurbineStreamClient.INPUT);
 		if (input.getDestination() == null) {
-			input.setDestination(properties.getDestination());
+			input.setDestination(this.properties.getDestination());
+		}
+		if (input.getContentType() == null) {
+			input.setContentType(this.properties.getContentType());
 		}
 	}
 

spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamProperties.java

@@ -16,11 +16,12 @@
 
 package org.springframework.cloud.netflix.turbine.stream;
 
-import lombok.Data;
-
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cloud.netflix.hystrix.HystrixConstants;
+import org.springframework.http.MediaType;
+
+import lombok.Data;
 
 /**
  * @author Dave Syer
@@ -33,4 +34,6 @@ public class TurbineStreamProperties {
 	private int port = 8989;
 
 	private String destination = HystrixConstants.HYSTRIX_STREAM_DESTINATION;
+
+	private String contentType = MediaType.APPLICATION_JSON_VALUE;
 }