diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java deleted file mode 100644 index 667080201..000000000 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/LazySpanSubscriber.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Supplier; - -import org.reactivestreams.Subscription; -import reactor.util.context.Context; - -/** - * A lazy representation of the {@link SpanSubscription}. - * - * @param of what subscription returns - * @author Marcin Grzejszczak - * @since 2.0.0 - */ -// TODO: why are we extending AtomicBoolean and not actually using its methods? -final class LazySpanSubscriber extends AtomicBoolean implements SpanSubscription { - - private final Supplier> supplier; - - LazySpanSubscriber(Supplier> supplier) { - this.supplier = supplier; - } - - @Override - public void onSubscribe(Subscription subscription) { - this.supplier.get().onSubscribe(subscription); - } - - @Override - public void request(long n) { - this.supplier.get().request(n); - } - - @Override - public void cancel() { - this.supplier.get().cancel(); - } - - @Override - public void onNext(T o) { - this.supplier.get().onNext(o); - } - - @Override - public void onError(Throwable throwable) { - this.supplier.get().onError(throwable); - } - - @Override - public void onComplete() { - this.supplier.get().onComplete(); - } - - @Override - public Context currentContext() { - return this.supplier.get().currentContext(); - } - -} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java index a1746e0ce..73a212a16 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java @@ -32,7 +32,6 @@ import reactor.core.Scannable; import reactor.core.publisher.Operators; import reactor.util.context.Context; -import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ConfigurableApplicationContext; /** @@ -55,15 +54,20 @@ public abstract class ReactorSleuth { * {@link reactor.core.publisher.Hooks#onLastOperator(Function)} or * {@link reactor.core.publisher.Hooks#onLastOperator(Function)}. The Span operator * pointcut will pass the Scope of the Span without ever creating any new spans. - * @param beanFactory - {@link BeanFactory} + * @param springContext the Spring context. * @param an arbitrary type that is left unchanged by the span operator * @return a new lazy span operator pointcut */ - @SuppressWarnings("unchecked") + // Much of Boot assumes that the Spring context will be a + // ConfigurableApplicationContext, rooted in SpringApplication's + // requirement for it to be so. Previous versions of Reactor + // instrumentation injected both BeanFactory and also + // ConfigurableApplicationContext. This chooses the more narrow + // signature as it is simpler than explaining instanceof checks. public static Function, ? extends Publisher> scopePassingSpanOperator( - BeanFactory beanFactory) { + ConfigurableApplicationContext springContext) { if (log.isTraceEnabled()) { - log.trace("Scope passing operator [" + beanFactory + "]"); + log.trace("Scope passing operator [" + springContext + "]"); } return Operators.liftPublisher((p, sub) -> { @@ -74,24 +78,23 @@ public abstract class ReactorSleuth { return sub; } - if (beanFactory instanceof ConfigurableApplicationContext - && ((ConfigurableApplicationContext) beanFactory).isActive()) { + if (!springContext.isActive()) { if (log.isTraceEnabled()) { - log.trace("Spring Context [" + beanFactory - + "] already refreshed. Creating a scope " - + "passing span subscriber with Reactor Context " + "[" - + sub.currentContext() + "] and name [" + name(sub) + "]"); + log.trace("Spring Context [" + springContext + + "] is not yet refreshed. This is unexpected. Reactor Context is [" + + sub.currentContext() + "] and name is [" + name(sub) + "]"); } + assert false; // should never happen, but don't break. + return sub; + } - return scopePassingSpanSubscription(beanFactory, sub); - } if (log.isTraceEnabled()) { - log.trace("Spring Context [" + beanFactory - + "] is not yet refreshed, falling back to lazy span subscriber. Reactor Context is [" - + sub.currentContext() + "] and name is [" + name(sub) + "]"); + log.trace("Spring Context [" + springContext + + "] Creating a scope passing span subscriber with Reactor Context " + + "[" + sub.currentContext() + "] and name [" + name(sub) + "]"); } - return new LazySpanSubscriber<>( - new SpanSubscriptionProvider<>(beanFactory, sub)); + + return scopePassingSpanSubscription(springContext, sub); }); } @@ -99,12 +102,12 @@ public abstract class ReactorSleuth { return Scannable.from(sub).name(); } - private static Map CACHE = new ConcurrentHashMap<>(); + private static Map CACHE = new ConcurrentHashMap<>(); static CoreSubscriber scopePassingSpanSubscription( - BeanFactory beanFactory, CoreSubscriber sub) { - CurrentTraceContext currentTraceContext = CACHE.computeIfAbsent(beanFactory, - beanFactory1 -> beanFactory1.getBean(CurrentTraceContext.class)); + ConfigurableApplicationContext springContext, CoreSubscriber sub) { + CurrentTraceContext currentTraceContext = CACHE.computeIfAbsent(springContext, + springContext1 -> springContext1.getBean(CurrentTraceContext.class)); Context context = sub.currentContext(); TraceContext parent = context.getOrDefault(TraceContext.class, null); diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java index 488e26632..aa76125e4 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java @@ -120,4 +120,10 @@ final class ScopePassingSpanSubscriber implements SpanSubscription, Scanna } } + @Override + public String toString() { + return "ScopePassingSpanSubscriber{" + "subscriber=" + this.subscriber + + ", parent=" + this.parent + "}"; + } + } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java deleted file mode 100644 index 2c79e5975..000000000 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java +++ /dev/null @@ -1,91 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import java.util.function.Supplier; - -import brave.propagation.CurrentTraceContext; -import brave.propagation.TraceContext; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import reactor.core.CoreSubscriber; -import reactor.util.context.Context; - -import org.springframework.beans.factory.BeanFactory; - -import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.name; - -/** - * Supplier to lazily start a {@link SpanSubscription}. - * - * @param type of returned subscription - * @author Marcin Grzejszczak - */ -final class SpanSubscriptionProvider implements Supplier> { - - private static final Log log = LogFactory.getLog(SpanSubscriptionProvider.class); - - final BeanFactory beanFactory; - - final CoreSubscriber subscriber; - - final Context context; - - private volatile CurrentTraceContext currentTraceContext; - - SpanSubscriptionProvider(BeanFactory beanFactory, - CoreSubscriber subscriber) { - this.beanFactory = beanFactory; - this.subscriber = subscriber; - this.context = subscriber.currentContext(); - if (log.isTraceEnabled()) { - log.trace("Spring context [" + beanFactory + "], Reactor context [" + context - + "], name [" + name(subscriber) + "]"); - } - } - - @Override - public SpanSubscription get() { - return newCoreSubscriber(currentTraceContext()); - } - - SpanSubscription newCoreSubscriber(CurrentTraceContext currentTraceContext) { - TraceContext root = this.context.hasKey(TraceContext.class) - ? this.context.get(TraceContext.class) : currentTraceContext.get(); - return new ScopePassingSpanSubscriber<>(this.subscriber, this.context, - currentTraceContext, root); - } - - private CurrentTraceContext currentTraceContext() { - if (this.currentTraceContext == null) { - try { - this.currentTraceContext = this.beanFactory - .getBean(CurrentTraceContext.class); - } - catch (Exception ex) { - if (log.isDebugEnabled()) { - log.debug( - "Exception occurred while trying to get the currentTraceContext bean. Will return a default instance", - ex); - } - return CurrentTraceContext.Default.create(); - } - } - return this.currentTraceContext; - } - -} diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java index c0c6c6ed4..f28ddba1c 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java @@ -25,8 +25,6 @@ import reactor.core.publisher.Hooks; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionRegistry; @@ -76,14 +74,14 @@ public class TraceReactorAutoConfiguration { private static final Log log = LogFactory.getLog(TraceReactorConfiguration.class); @Autowired - BeanFactory beanFactory; + ConfigurableApplicationContext springContext; @PreDestroy public void cleanupHooks() { if (log.isTraceEnabled()) { log.trace("Cleaning up hooks"); } - SleuthReactorProperties reactorProperties = this.beanFactory + SleuthReactorProperties reactorProperties = this.springContext .getBean(SleuthReactorProperties.class); if (reactorProperties.isDecorateOnEach()) { if (log.isTraceEnabled()) { @@ -175,26 +173,24 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor private static final Log log = LogFactory .getLog(HookRegisteringBeanDefinitionRegistryPostProcessor.class); - private final ConfigurableApplicationContext context; + final ConfigurableApplicationContext springContext; HookRegisteringBeanDefinitionRegistryPostProcessor( - ConfigurableApplicationContext context) { - this.context = context; + ConfigurableApplicationContext springContext) { + this.springContext = springContext; } @Override - public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) - throws BeansException { + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) { } @Override - public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) - throws BeansException { - setupHooks(beanFactory); + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) { + setupHooks(this.springContext); } - void setupHooks(BeanFactory beanFactory) { - ConfigurableEnvironment environment = this.context.getEnvironment(); + void setupHooks(ConfigurableApplicationContext springContext) { + ConfigurableEnvironment environment = springContext.getEnvironment(); boolean decorateOnEach = environment.getProperty( "spring.sleuth.reactor.decorate-on-each", Boolean.class, true); if (decorateOnEach) { @@ -202,20 +198,20 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor log.trace("Decorating onEach operator instrumentation"); } Hooks.onEachOperator(SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(this.context)); + ReactorSleuth.scopePassingSpanOperator(springContext)); } else { if (log.isTraceEnabled()) { log.trace("Decorating onLast operator instrumentation"); } Hooks.onLastOperator(SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(this.context)); + ReactorSleuth.scopePassingSpanOperator(springContext)); } Schedulers.setExecutorServiceDecorator( TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY, (scheduler, scheduledExecutorService) -> new TraceableScheduledExecutorService( - beanFactory, scheduledExecutorService)); + springContext, scheduledExecutorService)); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java index c78114d84..ef596dddf 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientAutoConfiguration.java @@ -43,6 +43,7 @@ import org.springframework.boot.web.client.RestTemplateCustomizer; import org.springframework.cloud.commons.httpclient.HttpClientConfiguration; import org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter; import org.springframework.cloud.sleuth.instrument.web.TraceWebServletAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.annotation.Order; @@ -133,8 +134,8 @@ public class TraceWebClientAutoConfiguration { @Bean static TraceWebClientBeanPostProcessor traceWebClientBeanPostProcessor( - BeanFactory beanFactory) { - return new TraceWebClientBeanPostProcessor(beanFactory); + ConfigurableApplicationContext springContext) { + return new TraceWebClientBeanPostProcessor(springContext); } } diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java index 8cafff570..558f384cf 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java @@ -38,9 +38,8 @@ import reactor.util.annotation.Nullable; import reactor.util.context.Context; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.BeanPostProcessor; -import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.web.client.RestClientException; import org.springframework.web.reactive.function.client.ClientRequest; @@ -49,6 +48,8 @@ import org.springframework.web.reactive.function.client.ExchangeFilterFunction; import org.springframework.web.reactive.function.client.ExchangeFunction; import org.springframework.web.reactive.function.client.WebClient; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; + /** * {@link BeanPostProcessor} to wrap a {@link WebClient} instance into its trace * representation. @@ -58,10 +59,10 @@ import org.springframework.web.reactive.function.client.WebClient; */ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { - private final BeanFactory beanFactory; + private final ConfigurableApplicationContext springContext; - TraceWebClientBeanPostProcessor(BeanFactory beanFactory) { - this.beanFactory = beanFactory; + TraceWebClientBeanPostProcessor(ConfigurableApplicationContext springContext) { + this.springContext = springContext; } @Override @@ -92,7 +93,7 @@ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { return functions -> { boolean noneMatch = noneMatchTraceExchangeFunction(functions); if (noneMatch) { - functions.add(new TraceExchangeFilterFunction(this.beanFactory)); + functions.add(new TraceExchangeFilterFunction(this.springContext)); } }; } @@ -134,7 +135,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { private static final String CANCELLED_SUBSCRIPTION_ERROR = "CANCELLED"; - final BeanFactory beanFactory; + final ConfigurableApplicationContext springContext; final Function, ? extends Publisher> scopePassingTransformer; @@ -146,14 +147,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { TraceContext.Injector injector; - TraceExchangeFilterFunction(BeanFactory beanFactory) { - this.beanFactory = beanFactory; - this.scopePassingTransformer = ReactorSleuth - .scopePassingSpanOperator(beanFactory); + TraceExchangeFilterFunction(ConfigurableApplicationContext springContext) { + this.springContext = springContext; + this.scopePassingTransformer = scopePassingSpanOperator(springContext); } - public static ExchangeFilterFunction create(BeanFactory beanFactory) { - return new TraceExchangeFilterFunction(beanFactory); + public static ExchangeFilterFunction create( + ConfigurableApplicationContext springContext) { + return new TraceExchangeFilterFunction(springContext); } @Override @@ -177,7 +178,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { HttpClientHandler handler() { if (this.handler == null) { this.handler = HttpClientHandler - .create(this.beanFactory.getBean(HttpTracing.class)); + .create(this.springContext.getBean(HttpTracing.class)); } return this.handler; } @@ -191,14 +192,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { HttpTracing httpTracing() { if (this.httpTracing == null) { - this.httpTracing = this.beanFactory.getBean(HttpTracing.class); + this.httpTracing = this.springContext.getBean(HttpTracing.class); } return this.httpTracing; } TraceContext.Injector injector() { if (this.injector == null) { - this.injector = this.beanFactory.getBean(HttpTracing.class).tracing() + this.injector = this.springContext.getBean(HttpTracing.class).tracing() .propagation().injector(SETTER); } return this.injector; diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java deleted file mode 100644 index 2146941f5..000000000 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProviderTests.java +++ /dev/null @@ -1,47 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import org.assertj.core.api.BDDAssertions; -import org.junit.Test; -import org.mockito.BDDMockito; -import reactor.core.CoreSubscriber; -import reactor.util.context.Context; - -import org.springframework.beans.factory.BeanFactory; - -public class SpanSubscriptionProviderTests { - - @Test - public void should_return_default_tracing_instance_when_exception_thrown_upon_bean_retrieval() { - CoreSubscriber subscriber = BDDMockito.mock(CoreSubscriber.class); - BDDMockito.when(subscriber.currentContext()).thenReturn(Context.empty()); - - BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); - - BDDMockito.when(beanFactory.getBean(BDDMockito.any(Class.class))) - .thenThrow(new IllegalStateException()); - - SpanSubscriptionProvider provider = new SpanSubscriptionProvider<>( - beanFactory, subscriber); - - SpanSubscription spanSubscription = provider.get(); - - BDDAssertions.then(spanSubscription).isNotNull(); - } - -} diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java index db74c2a13..4bf00805c 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java @@ -22,7 +22,7 @@ import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; -import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.web.reactive.function.client.WebClient; /** @@ -32,12 +32,12 @@ import org.springframework.web.reactive.function.client.WebClient; public class TraceWebClientBeanPostProcessorTest { @Mock - BeanFactory beanFactory; + ConfigurableApplicationContext springContext; @Test public void should_add_filter_only_once_to_web_client() { TraceWebClientBeanPostProcessor processor = new TraceWebClientBeanPostProcessor( - this.beanFactory); + this.springContext); WebClient client = WebClient.create(); client = (WebClient) processor.postProcessAfterInitialization(client, "foo"); @@ -53,7 +53,7 @@ public class TraceWebClientBeanPostProcessorTest { @Test public void should_add_filter_only_once_to_web_client_via_builder() { TraceWebClientBeanPostProcessor processor = new TraceWebClientBeanPostProcessor( - this.beanFactory); + this.springContext); WebClient.Builder builder = WebClient.builder(); builder = (WebClient.Builder) processor.postProcessAfterInitialization(builder, diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java new file mode 100644 index 000000000..ef0a6bbd8 --- /dev/null +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberSpringBootTests.java @@ -0,0 +1,204 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.reactor; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; + +import brave.propagation.CurrentTraceContext; +import brave.propagation.CurrentTraceContext.Scope; +import brave.propagation.TraceContext; +import brave.sampler.Sampler; +import org.awaitility.Awaitility; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.BDDAssertions.then; + +/** + * Like {@link ScopePassingSpanSubscriberTests}, except this tests wiring with spring boot + * config. + */ +@RunWith(SpringRunner.class) +@SpringBootTest(classes = ScopePassingSpanSubscriberSpringBootTests.Config.class, + webEnvironment = SpringBootTest.WebEnvironment.NONE) +public class ScopePassingSpanSubscriberSpringBootTests { + + @Autowired + CurrentTraceContext currentTraceContext; + + TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true) + .build(); + + TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true) + .build(); + + @Test + public void should_pass_tracing_info_when_using_reactor() { + final AtomicReference spanInOperation = new AtomicReference<>(); + Publisher traced = Flux.just(1, 2, 3); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.from(traced).map(d -> d + 1).map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).subscribe(d -> { + }); + } + + then(this.currentTraceContext.get()).isNull(); + then(spanInOperation.get()).isEqualTo(context); + } + + @Test + public void should_support_reactor_fusion_optimization() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.just(1).flatMap(d -> Flux.just(d + 1).collectList().map(p -> p.get(0))) + .map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).subscribe(d -> { + }); + } + + then(this.currentTraceContext.get()).isNull(); + then(spanInOperation.get()).isEqualTo(context); + } + + @Test + public void should_pass_tracing_info_when_using_reactor_async() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.1") + .map(d -> d + 1).map(d -> d + 1) + .publishOn(Schedulers.newSingle("secondThread")).log("reactor.2") + .map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).blockLast(); + + Awaitility.await() + .untilAsserted(() -> then(spanInOperation.get()).isEqualTo(context)); + then(this.currentTraceContext.get()).isEqualTo(context); + } + + then(this.currentTraceContext.get()).isNull(); + + try (Scope ws = this.currentTraceContext.newScope(context2)) { + Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.") + .map(d -> d + 1).map(d -> d + 1).map((d) -> { + spanInOperation.set(this.currentTraceContext.get()); + return d + 1; + }).map(d -> d + 1).blockLast(); + + then(this.currentTraceContext.get()).isEqualTo(context2); + then(spanInOperation.get()).isEqualTo(context2); + } + + then(this.currentTraceContext.get()).isNull(); + } + + @Test + public void onlyConsidersContextDuringSubscribe() { + Mono fromMono = Mono.fromCallable(this.currentTraceContext::get); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + then(fromMono.map(context -> context).block()).isNotNull(); + } + } + + @Test + public void checkTraceIdDuringZipOperation() { + final AtomicReference spanInOperation = new AtomicReference<>(); + final AtomicReference spanInZipOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.fromCallable(this.currentTraceContext::get).map(span -> span) + .doOnNext(spanInOperation::set) + .zipWith(Mono.fromCallable(this.currentTraceContext::get) + .map(span -> span).doOnNext(spanInZipOperation::set)) + .block(); + } + + then(spanInZipOperation).hasValue(context); + then(spanInOperation).hasValue(context); + } + + // #646 + @Test + public void should_work_for_mono_just_with_flat_map() { + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.just("value1") + .flatMap(request -> Mono.just("value2").then(Mono.just("foo"))) + .map(a -> "qwe").block(); + } + } + + // #1030 + @Test + public void checkTraceIdFromSubscriberContext() { + final AtomicReference spanInSubscriberContext = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Mono.subscriberContext().map(context -> this.currentTraceContext.get()) + .doOnNext(spanInSubscriberContext::set).block(); + } + + then(spanInSubscriberContext).hasValue(context); // ok here + } + + @Test + public void should_pass_tracing_info_into_inner_publishers() { + final AtomicReference spanInOperation = new AtomicReference<>(); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Flux.range(0, 5) + .flatMap(it -> Mono.delay(Duration.ofMillis(1)) + .map(context -> this.currentTraceContext.get()) + .doOnNext(spanInOperation::set)) + .blockFirst(); + } + + then(spanInOperation.get()).isEqualTo(context); + } + + @EnableAutoConfiguration + @Configuration + static class Config { + + @Bean + Sampler sampler() { + return Sampler.ALWAYS_SAMPLE; + } + + } + +} diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java index 6c0bf89c6..b809a743e 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java @@ -16,32 +16,59 @@ package org.springframework.cloud.sleuth.instrument.reactor; +import java.util.Objects; +import java.util.function.Function; + import brave.propagation.CurrentTraceContext; import brave.propagation.CurrentTraceContext.Scope; import brave.propagation.TraceContext; +import org.assertj.core.presentation.StandardRepresentation; +import org.awaitility.Awaitility; +import org.junit.After; import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.BDDMockito; -import org.mockito.junit.MockitoJUnitRunner; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Mono; import reactor.util.context.Context; -import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.BDDAssertions.then; +import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator; /** * @author Marcin Grzejszczak */ -@RunWith(MockitoJUnitRunner.class) public class ScopePassingSpanSubscriberTests { - CurrentTraceContext currentTraceContext = CurrentTraceContext.Default.create(); + static { + // AssertJ will recognise QueueSubscription implements queue and try to invoke + // iterator. That's not allowed, and will cause an exception + // Fuseable$QueueSubscription.NOT_SUPPORTED_MESSAGE. + // This ensures AssertJ uses normal toString. + StandardRepresentation.registerFormatterForType(ScopePassingSpanSubscriber.class, + Objects::toString); + } + + final CurrentTraceContext currentTraceContext = CurrentTraceContext.Default.create(); TraceContext context = TraceContext.newBuilder().traceId(1).spanId(1).sampled(true) .build(); + TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true) + .build(); + + AnnotationConfigApplicationContext springContext = new AnnotationConfigApplicationContext(); + + @After + public void close() { + springContext.close(); + } + @Test public void should_propagate_current_context() { ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, @@ -53,28 +80,96 @@ public class ScopePassingSpanSubscriberTests { @Test public void should_set_empty_context_when_context_is_null() { ScopePassingSpanSubscriber subscriber = new ScopePassingSpanSubscriber<>(null, - null, this.currentTraceContext, null); + Context.empty(), this.currentTraceContext, null); then(subscriber.currentContext().isEmpty()).isTrue(); } @Test public void should_put_current_span_to_context() { - try (Scope ws = this.currentTraceContext.newScope(context)) { - CoreSubscriber subscriber = ReactorSleuth.scopePassingSpanSubscription( - beanFactory(), new BaseSubscriber() { - }); + try (Scope ws = this.currentTraceContext.newScope(context2)) { + CoreSubscriber subscriber = new ScopePassingSpanSubscriber<>( + new BaseSubscriber() { + }, Context.empty(), currentTraceContext, context); then(subscriber.currentContext().get(TraceContext.class)).isEqualTo(context); } - } - private BeanFactory beanFactory() { - BeanFactory beanFactory = BDDMockito.mock(BeanFactory.class); - BDDMockito.given(beanFactory.getBean(CurrentTraceContext.class)) - .willReturn(this.currentTraceContext); - return beanFactory; + @Test + public void should_not_trace_scalar_flows() { + springContext.registerBean(CurrentTraceContext.class, () -> currentTraceContext); + springContext.refresh(); + + Function, ? extends Publisher> transformer = scopePassingSpanOperator( + this.springContext); + + try (Scope ws = this.currentTraceContext.newScope(context)) { + Subscriber assertNoSpanSubscriber = new CoreSubscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class); + } + + @Override + public void onNext(Object o) { + + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + }; + + Subscriber assertSpanSubscriber = new CoreSubscriber() { + @Override + public void onSubscribe(Subscription s) { + s.request(Long.MAX_VALUE); + assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class); + } + + @Override + public void onNext(Object o) { + + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + }; + transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber); + + transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber); + + transformer.apply(Mono.error(new Exception()).hide()) + .subscribe(assertSpanSubscriber); + + transformer.apply(Mono.error(new Exception())) + .subscribe(assertNoSpanSubscriber); + + transformer.apply(Mono.empty().hide()) + .subscribe(assertSpanSubscriber); + + transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber); + + } + + Awaitility.await().untilAsserted(() -> { + then(this.currentTraceContext.get()).isNull(); + }); } } diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java deleted file mode 100644 index 9a8278f6c..000000000 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java +++ /dev/null @@ -1,333 +0,0 @@ -/* - * Copyright 2013-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.sleuth.instrument.reactor; - -import java.time.Duration; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Function; - -import brave.Span; -import brave.Tracer; -import brave.sampler.Sampler; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.awaitility.Awaitility; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; -import reactor.core.CoreSubscriber; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.test.context.junit4.SpringRunner; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.BDDAssertions.then; - -@RunWith(SpringRunner.class) -@SpringBootTest(classes = SpanSubscriberTests.Config.class, - webEnvironment = SpringBootTest.WebEnvironment.NONE) -public class SpanSubscriberTests { - - private static final Log log = LogFactory.getLog(SpanSubscriberTests.class); - - @Autowired - Tracer tracer; - - @Autowired - ConfigurableApplicationContext factory; - - @Test - public void should_pass_tracing_info_when_using_reactor() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - Publisher traced = Flux.just(1, 2, 3); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.from(traced).map(d -> d + 1).map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).subscribe(System.out::println); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @Test - public void should_support_reactor_fusion_optimization() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Mono.just(1).flatMap(d -> Flux.just(d + 1).collectList().map(p -> p.get(0))) - .map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).subscribe(System.out::println); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @Test - public void should_not_trace_scalar_flows() { - Span span = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - - // Disable global hooks for local hook testing - TraceReactorAutoConfigurationAccessorConfiguration.close(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - - Function, ? extends Publisher> transformer = ReactorSleuth - .scopePassingSpanOperator(this.factory); - - Subscriber assertNoSpanSubscriber = new CoreSubscriber() { - @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); - assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class); - } - - @Override - public void onNext(Object o) { - - } - - @Override - public void onError(Throwable t) { - - } - - @Override - public void onComplete() { - - } - }; - - Subscriber assertSpanSubscriber = new CoreSubscriber() { - @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); - assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class); - } - - @Override - public void onNext(Object o) { - - } - - @Override - public void onError(Throwable t) { - - } - - @Override - public void onComplete() { - - } - }; - transformer.apply(Mono.just(1).hide()).subscribe(assertSpanSubscriber); - - transformer.apply(Mono.just(1)).subscribe(assertNoSpanSubscriber); - - transformer.apply(Mono.error(new Exception()).hide()) - .subscribe(assertSpanSubscriber); - - transformer.apply(Mono.error(new Exception())) - .subscribe(assertNoSpanSubscriber); - - transformer.apply(Mono.empty().hide()) - .subscribe(assertSpanSubscriber); - - transformer.apply(Mono.empty()).subscribe(assertNoSpanSubscriber); - - } - finally { - span.finish(); - } - - Awaitility.await().untilAsserted(() -> { - then(this.tracer.currentSpan()).isNull(); - }); - - TraceReactorAutoConfigurationAccessorConfiguration.setup(this.factory); - } - - @Test - public void should_pass_tracing_info_when_using_reactor_async() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.1") - .map(d -> d + 1).map(d -> d + 1) - .publishOn(Schedulers.newSingle("secondThread")).log("reactor.2") - .map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).blockLast(); - - Awaitility.await().untilAsserted(() -> { - then(spanInOperation.get().context().traceId()) - .isEqualTo(span.context().traceId()); - }); - then(this.tracer.currentSpan()).isEqualTo(span); - } - finally { - span.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - Span foo2 = this.tracer.nextSpan().name("foo").start(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(foo2)) { - Flux.just(1, 2, 3).publishOn(Schedulers.single()).log("reactor.") - .map(d -> d + 1).map(d -> d + 1).map((d) -> { - spanInOperation.set(this.tracer.currentSpan()); - return d + 1; - }).map(d -> d + 1).blockLast(); - - then(this.tracer.currentSpan()).isEqualTo(foo2); - // parent cause there's an async span in the meantime - then(spanInOperation.get().context().traceId()) - .isEqualTo(foo2.context().traceId()); - } - finally { - foo2.finish(); - } - - then(this.tracer.currentSpan()).isNull(); - } - - @Test - public void checkSequenceOfOperations() { - Span parentSpan = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(parentSpan)) { - final Long spanId = Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).block(); - then(spanId).isNotNull(); - - final Long secondSpanId = Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).block(); - then(secondSpanId).isEqualTo(spanId); // different trace ids here - } - } - - @Test - public void checkTraceIdDuringZipOperation() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - final AtomicReference spanInZipOperation = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()).doOnNext(spanInOperation::set) - .zipWith(Mono.fromCallable(this.tracer::currentSpan) - .map(span -> span.context().spanId()) - .doOnNext(spanInZipOperation::set)) - .block(); - } - - then(spanInZipOperation).hasValue(initSpan.context().spanId()); // ok here - then(spanInOperation).hasValue(initSpan.context().spanId()); // Expecting - // - // to have value: - // <1L> but did - // not. - } - - // #646 - @Test - public void should_work_for_mono_just_with_flat_map() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - log.info("Hello"); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.just("value1") - .flatMap(request -> Mono.just("value2").then(Mono.just("foo"))) - .map(a -> "qwe").block(); - } - } - - // #1030 - @Test - public void checkTraceIdFromSubscriberContext() { - Span initSpan = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInSubscriberContext = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(initSpan)) { - Mono.subscriberContext() - .map(context -> this.tracer.currentSpan().context().spanId()) - .doOnNext(spanInSubscriberContext::set).block(); - } - - then(spanInSubscriberContext).hasValue(initSpan.context().spanId()); // ok here - } - - @Test - public void should_pass_tracing_info_into_inner_publishers() { - Span span = this.tracer.nextSpan().name("foo").start(); - final AtomicReference spanInOperation = new AtomicReference<>(); - - try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) { - Flux.range(0, 5) - .flatMap(it -> Mono.delay(Duration.ofMillis(1)) - .map(context -> this.tracer.currentSpan()) - .doOnNext(spanInOperation::set)) - .blockFirst(); - } - finally { - span.finish(); - } - - then(spanInOperation.get().context().spanId()).isEqualTo(span.context().spanId()); - } - - @EnableAutoConfiguration - @Configuration - static class Config { - - @Bean - Sampler sampler() { - return Sampler.ALWAYS_SAMPLE; - } - - } - -}