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 805a0a76..791514e2 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 @@ -23,11 +23,14 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.client.actuator.HasFeatures; +import org.springframework.cloud.client.discovery.DiscoveryClient; import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Output; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.MessageChannel; import org.springframework.scheduling.annotation.EnableScheduling; import com.netflix.hystrix.HystrixCircuitBreaker; @@ -57,6 +60,10 @@ public class HystrixStreamAutoConfiguration { @Autowired private HystrixStreamProperties properties; + @Autowired + @Output(HystrixStreamClient.OUTPUT) + private MessageChannel outboundChannel; + @Bean public HasFeatures hystrixStreamQueueFeature() { return HasFeatures.namedFeature("Hystrix Stream (Queue)", @@ -87,8 +94,8 @@ public class HystrixStreamAutoConfiguration { } @Bean - public HystrixStreamTask hystrixStreamTask() { - return new HystrixStreamTask(); + public HystrixStreamTask hystrixStreamTask(DiscoveryClient discoveryClient) { + return new HystrixStreamTask(this.outboundChannel, discoveryClient, this.properties); } } 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 282c5631..50fe3048 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 @@ -28,14 +28,28 @@ import lombok.Data; @Data public class HystrixStreamProperties { + /** Flag to indicate that Hystrix Stream is enabled. Default is true. */ private boolean enabled = true; + /** Flag to indicate to prefix metric names with serviceId. Default is true. */ private boolean prefixMetricName = true; + /** Flag to indicate to send the id field in the metrics. Default is true */ private boolean sendId = true; + /** The destination of the stream. Destination as defined by Spring Cloud Stream. Defaults to springCloudHystrixStream */ private String destination = HystrixConstants.HYSTRIX_STREAM_DESTINATION; + /** The content type of the messages. Defaults to application/json */ private String contentType = "application/json"; + /** How often (in ms) to send messages to the stream. Defaults to 500. */ + private long sendRate = 500; + + /** How often to put messages in the queue. This queue drains to the stream. Defaults to 500. */ + private long gatherRate = 500; + + /** The size of the metrics queue. This queue drains to the stream. Defaults to 1000. */ + private int size = 1000; + } 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 66aa2140..2af36637 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 @@ -23,10 +23,8 @@ import java.util.Collection; import java.util.concurrent.LinkedBlockingQueue; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cloud.client.ServiceInstance; import org.springframework.cloud.client.discovery.DiscoveryClient; -import org.springframework.cloud.stream.annotation.Output; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.messaging.MessageChannel; @@ -55,25 +53,26 @@ import lombok.extern.apachecommons.CommonsLog; @CommonsLog public class HystrixStreamTask implements ApplicationContextAware { - @Autowired - @Output(HystrixStreamClient.OUTPUT) private MessageChannel outboundChannel; - @Autowired private DiscoveryClient discoveryClient; + private HystrixStreamProperties properties; + private ApplicationContext context; - @Autowired - private HystrixStreamProperties properties; - // Visible for testing - // TODO: configurable size. - final LinkedBlockingQueue jsonMetrics = new LinkedBlockingQueue<>( - 1000); + final LinkedBlockingQueue jsonMetrics; private final JsonFactory jsonFactory = new JsonFactory(); + public HystrixStreamTask(MessageChannel outboundChannel, DiscoveryClient discoveryClient, HystrixStreamProperties properties) { + this.outboundChannel = outboundChannel; + this.discoveryClient = discoveryClient; + this.properties = properties; + this.jsonMetrics = new LinkedBlockingQueue<>(properties.getSize()); + } + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { @@ -96,7 +95,7 @@ public class HystrixStreamTask implements ApplicationContextAware { // 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") + this.properties.getContentType()) .build()); } catch (Exception ex) { diff --git a/spring-cloud-netflix-hystrix-stream/src/test/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamTaskTests.java b/spring-cloud-netflix-hystrix-stream/src/test/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamTaskTests.java index 9b55d5dd..fc3dfabb 100644 --- a/spring-cloud-netflix-hystrix-stream/src/test/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamTaskTests.java +++ b/spring-cloud-netflix-hystrix-stream/src/test/java/org/springframework/cloud/netflix/hystrix/stream/HystrixStreamTaskTests.java @@ -61,8 +61,9 @@ public class HystrixStreamTaskTests { new HystrixPropertiesCommandDefault(hystrixCommandKey, HystrixCommandProperties.defaultSetter())); given(this.discoveryClient.getLocalServiceInstance()).willReturn(this.serviceInstance); + this.hystrixStreamTask.setApplicationContext(this.context); this.hystrixStreamTask.gatherMetrics(); assertThat(this.hystrixStreamTask.jsonMetrics.isEmpty(), is(false)); } -} \ No newline at end of file +}