diff --git a/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamAutoConfiguration.java b/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamAutoConfiguration.java index a1f7a360..805a0a76 100644 --- a/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamAutoConfiguration.java +++ b/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 - * 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. + * 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 + * hystrix.stream.queue.enabled. 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 diff --git a/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamProperties.java b/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamProperties.java index 6e66a20d..282c5631 100644 --- a/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamProperties.java +++ b/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"; + } diff --git a/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamTask.java b/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamTask.java index 7c67e645..84c21b47 100644 --- a/spring-cloud-netflix-hystrix-stream/src/main/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamTask.java +++ b/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", diff --git a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamAutoConfiguration.java b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamAutoConfiguration.java index 3dd6cf95..84270b83 100644 --- a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamAutoConfiguration.java +++ b/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 turbine.stream.enabled. + * 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 + * turbine.stream.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 turbine and another for business messages, you - * need to create both, and annotate them @TurbineConnectionFactory and - * @Primary respectively. + * 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 turbine and + * another for business messages, you need to create both, and annotate them + * @TurbineConnectionFactory and @Primary 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()); } } diff --git a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamProperties.java b/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamProperties.java index cd2895fd..6c919ab5 100644 --- a/spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/TurbineStreamProperties.java +++ b/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; }