diff --git a/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfiguration.java b/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfiguration.java index c5e748e9b..e69f04564 100644 --- a/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfiguration.java +++ b/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfiguration.java @@ -38,20 +38,23 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.Ordered; import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.scheduling.support.PeriodicTrigger; /** - * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration Auto-configuration} - * for sending spans over Spring Cloud Stream. This is for the producer - * (via {@link SleuthSource}). A consumer can enable binding to {@link SleuthSink} and - * receive the messages coming from the source (they have the same channel name so there - * is no additional configuration to do by default). + * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration + * Auto-configuration} for sending spans over Spring Cloud Stream. This is for + * the producer (via {@link SleuthSource}). A consumer can enable binding to + * {@link SleuthSink} and receive the messages coming from the source (they have + * the same channel name so there is no additional configuration to do by + * default). * * @author Dave Syer * @since 1.0.0 */ @Configuration -@EnableConfigurationProperties({SleuthStreamProperties.class, SamplerProperties.class}) +@EnableConfigurationProperties({ SleuthStreamProperties.class, SamplerProperties.class }) @AutoConfigureAfter(TraceMetricsAutoConfiguration.class) @AutoConfigureBefore(ChannelBindingAutoConfiguration.class) @EnableBinding(SleuthSource.class) @@ -77,6 +80,15 @@ public class SleuthStreamAutoConfiguration { return new StreamSpanReporter(endpointLocator, spanMetricReporter); } + @Bean(name = StreamSpanReporter.POLLER) + @ConditionalOnMissingBean(name = StreamSpanReporter.POLLER) + public PollerMetadata defaultStreamSpanReporterPoller(SleuthStreamProperties sleuth) { + PollerMetadata poller = new PollerMetadata(); + poller.setTrigger(new PeriodicTrigger(sleuth.getPoller().getFixedDelay())); + poller.setMaxMessagesPerPoll(sleuth.getPoller().getMaxMessagesPerPoll()); + return poller; + } + @Configuration @ConditionalOnMissingClass("org.springframework.cloud.client.discovery.DiscoveryClient") protected static class DefaultEndpointLocatorConfiguration { diff --git a/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamProperties.java b/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamProperties.java index 415f75dd8..81c7959d9 100644 --- a/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamProperties.java +++ b/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamProperties.java @@ -28,6 +28,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; public class SleuthStreamProperties { private boolean enabled = true; private String group = SleuthSink.INPUT; + private Poller poller = new Poller(); public boolean isEnabled() { return this.enabled; @@ -44,4 +45,36 @@ public class SleuthStreamProperties { public void setGroup(String group) { this.group = group; } + + public Poller getPoller() { + return this.poller; + } + + public static class Poller { + /** + * Fixed delay (ms). Default: 1000 + */ + private long fixedDelay = 1000L; + + /** + * Max messages per poll. Default: -1 (unbounded) + */ + private int maxMessagesPerPoll = -1; + + public long getFixedDelay() { + return this.fixedDelay; + } + + public int getMaxMessagesPerPoll() { + return this.maxMessagesPerPoll; + } + + public void setFixedDelay(long fixedDelay) { + this.fixedDelay = fixedDelay; + } + + public void setMaxMessagesPerPoll(int maxMessagesPerPoll) { + this.maxMessagesPerPoll = maxMessagesPerPoll; + } + } } diff --git a/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/StreamSpanReporter.java b/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/StreamSpanReporter.java index e28491a0f..7ecbd8c9a 100644 --- a/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/StreamSpanReporter.java +++ b/spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/StreamSpanReporter.java @@ -27,7 +27,7 @@ import org.springframework.cloud.sleuth.SpanReporter; import org.springframework.cloud.sleuth.metric.SpanMetricReporter; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.MessageEndpoint; - +import org.springframework.integration.annotation.Poller; /** * A message source for spans. Also handles RPC flavoured annotations. @@ -38,6 +38,13 @@ import org.springframework.integration.annotation.MessageEndpoint; @MessageEndpoint public class StreamSpanReporter implements SpanReporter { + /** + * Bean name for the + * {@link org.springframework.integration.scheduling.PollerMetadata + * PollerMetadata} + */ + public static final String POLLER = "streamSpanReporterPoller"; + private BlockingQueue queue = new LinkedBlockingQueue<>(); private final HostLocator endpointLocator; private final SpanMetricReporter spanMetricReporter; @@ -51,7 +58,7 @@ public class StreamSpanReporter implements SpanReporter { this.queue = queue; } - @InboundChannelAdapter(value = SleuthSource.OUTPUT) + @InboundChannelAdapter(value = SleuthSource.OUTPUT, poller = @Poller(POLLER)) public Spans poll() { List result = new LinkedList<>(); this.queue.drainTo(result); diff --git a/spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfigurationTest.java b/spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfigurationTest.java new file mode 100644 index 000000000..3e0d02bf1 --- /dev/null +++ b/spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfigurationTest.java @@ -0,0 +1,142 @@ +package org.springframework.cloud.sleuth.stream; + +import java.text.DateFormat; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.Date; + +import org.junit.After; +import org.junit.BeforeClass; +import org.junit.Test; +import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; +import org.springframework.boot.test.EnvironmentTestUtils; +import org.springframework.cloud.sleuth.autoconfig.TraceAutoConfiguration; +import org.springframework.cloud.sleuth.log.NoOpSpanLogger; +import org.springframework.cloud.sleuth.log.SpanLogger; +import org.springframework.cloud.sleuth.metric.TraceMetricsAutoConfiguration; +import org.springframework.cloud.stream.config.ChannelBindingAutoConfiguration; +import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; +import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.scheduling.support.SimpleTriggerContext; + +import static org.assertj.core.api.Assertions.assertThat; + +public class SleuthStreamAutoConfigurationTest { + + private AnnotationConfigApplicationContext ctx; + private static DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + private static final String TEST_SCHEDULED = "2016-01-01 12:00:00"; + private static final String TEST_COMPLETION = "2016-01-01 12:00:02"; + private static Date LAST_SCHEDULED_DATE; + private static Date LAST_EXECUTED_DATE; + private static Date LAST_COMPLETED_DATE; + + @BeforeClass + public static void setupTests() throws ParseException { + LAST_SCHEDULED_DATE = dateFormat.parse(TEST_SCHEDULED); + LAST_EXECUTED_DATE = new Date(LAST_SCHEDULED_DATE.getTime()); + LAST_COMPLETED_DATE = dateFormat.parse(TEST_COMPLETION); + } + + @After + public void cleanup() { + if (ctx != null) { + ctx.close(); + } + } + + @Test + public void shouldUseDefaultPollerConfiguration() { + ctx = createContext(); + ctx.refresh(); + + PollerMetadata poller = ctx.getBean(StreamSpanReporter.POLLER, + PollerMetadata.class); + assertThat(poller).isNotNull(); + assertPollerConfigurationUsingConfigurationProperties(poller); + } + + @Test + public void shouldUseCustomPollerConfiguration() { + ctx = createContext(); + EnvironmentTestUtils.addEnvironment(ctx, + "spring.sleuth.stream.poller.fixed-delay=5000", + "spring.sleuth.stream.poller.max-messages-per-poll=100"); + ctx.refresh(); + + PollerMetadata poller = ctx.getBean(StreamSpanReporter.POLLER, + PollerMetadata.class); + assertThat(poller).isNotNull(); + assertPollerConfigurationUsingConfigurationProperties(poller); + } + + @Test + public void shouldUseCustomPollerBean() { + ctx = createContext(CustomPollerConfiguration.class); + ctx.refresh(); + + PollerMetadata poller = ctx.getBean(StreamSpanReporter.POLLER, + PollerMetadata.class); + assertThat(poller).isNotNull(); + assertPollerConfiguration(poller, 500L, 5000L); + } + + private void assertPollerConfigurationUsingConfigurationProperties( + PollerMetadata poller) { + SleuthStreamProperties sleuth = ctx.getBean(SleuthStreamProperties.class); + assertPollerConfiguration(poller, sleuth.getPoller().getMaxMessagesPerPoll(), + sleuth.getPoller().getFixedDelay()); + } + + private void assertPollerConfiguration(PollerMetadata poller, + long expectedMaxMessages, long expectedFixedDelay) { + assertThat(poller.getMaxMessagesPerPoll()).isEqualTo(expectedMaxMessages); + Trigger trigger = poller.getTrigger(); + assertThat(trigger).isInstanceOf(PeriodicTrigger.class); + TriggerContext triggerContext = new SimpleTriggerContext(LAST_SCHEDULED_DATE, + LAST_EXECUTED_DATE, LAST_COMPLETED_DATE); + Date nextExecution = trigger.nextExecutionTime(triggerContext); + assertThat(nextExecution.getTime()) + .isEqualTo(LAST_COMPLETED_DATE.getTime() + expectedFixedDelay); + } + + public AnnotationConfigApplicationContext createContext(Class... classes) { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + if (classes != null && classes.length > 0) { + context.register(classes); + } + context.register(BaseConfiguration.class); + return context; + } + + @Configuration + @Import({ SleuthStreamAutoConfiguration.class, TraceMetricsAutoConfiguration.class, + TestSupportBinderAutoConfiguration.class, + ChannelBindingAutoConfiguration.class, TraceAutoConfiguration.class, + PropertyPlaceholderAutoConfiguration.class }) + public static class BaseConfiguration { + @Bean + SpanLogger spanLogger() { + return new NoOpSpanLogger(); + } + } + + @Configuration + public static class CustomPollerConfiguration { + + @Bean(name = StreamSpanReporter.POLLER) + PollerMetadata customPoller() { + PollerMetadata poller = new PollerMetadata(); + poller.setMaxMessagesPerPoll(500); + poller.setTrigger(new PeriodicTrigger(5000L)); + return poller; + } + } +} diff --git a/spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/StreamSpanListenerTests.java b/spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/StreamSpanListenerTests.java index 1ad85c568..cf41a4c9c 100644 --- a/spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/StreamSpanListenerTests.java +++ b/spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/StreamSpanListenerTests.java @@ -33,7 +33,6 @@ import org.mockito.Mockito; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.actuate.metrics.CounterService; import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; -import org.springframework.boot.test.SpringApplicationConfiguration; import org.springframework.cloud.sleuth.Sampler; import org.springframework.cloud.sleuth.Span; import org.springframework.cloud.sleuth.SpanReporter; @@ -53,22 +52,29 @@ import org.springframework.context.annotation.Import; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.messaging.Message; +import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Dave Syer * */ -@SpringApplicationConfiguration(classes = TestConfiguration.class) +@ContextConfiguration(classes = TestConfiguration.class) @RunWith(SpringJUnit4ClassRunner.class) public class StreamSpanListenerTests { - @Autowired Tracer tracer; - @Autowired ApplicationContext application; - @Autowired ZipkinTestConfiguration test; - @Autowired StreamSpanReporter listener; - @Autowired CounterService counterService; - @Autowired SpanReporter spanReporter; + @Autowired + Tracer tracer; + @Autowired + ApplicationContext application; + @Autowired + ZipkinTestConfiguration test; + @Autowired + StreamSpanReporter listener; + @Autowired + CounterService counterService; + @Autowired + SpanReporter spanReporter; @PostConstruct public void init() { @@ -149,7 +155,8 @@ public class StreamSpanListenerTests { TraceMetricsAutoConfiguration.class, TestSupportBinderAutoConfiguration.class, ChannelBindingAutoConfiguration.class, TraceAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class }) - protected static class TestConfiguration {} + protected static class TestConfiguration { + } @Configuration @MessageEndpoint @@ -160,7 +167,7 @@ public class StreamSpanListenerTests { @Autowired StreamSpanReporter listener; - @ServiceActivator(inputChannel=SleuthSource.OUTPUT) + @ServiceActivator(inputChannel = SleuthSource.OUTPUT) public void handle(Message msg) { } @@ -174,7 +181,8 @@ public class StreamSpanListenerTests { return new AlwaysSampler(); } - @Bean CounterService counterService() { + @Bean + CounterService counterService() { return Mockito.mock(CounterService.class); }