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 af6489705..dc05aa4d1 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 @@ -24,6 +24,9 @@ import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.NoSuchBeanDefinitionException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; + import reactor.core.CoreSubscriber; import reactor.core.Scannable; import reactor.core.publisher.ConnectableFlux; @@ -43,65 +46,6 @@ public abstract class ReactorSleuth { private ReactorSleuth() { } - - /** - * Return a span operator pointcut given a {@link BeanFactory}. This can be used in - * reactor via {@link reactor.core.publisher.Flux#transform(Function)}, - * {@link reactor.core.publisher.Mono#transform(Function)}, - * {@link reactor.core.publisher.Hooks#onEachOperator(Function)} or - * {@link reactor.core.publisher.Hooks#onLastOperator(Function)}. - * @deprecated use {@link ReactorSleuth#scopePassingSpanOperator} instead - * @param beanFactory - {@link BeanFactory} - * @param an arbitrary type that is left unchanged by the span operator - * @return a new lazy span operator pointcut - */ - @SuppressWarnings("unchecked") - @Deprecated - public static Function, ? extends Publisher> spanOperator( - BeanFactory beanFactory) { - - if (log.isWarnEnabled()) { - log.warn("spanOperator method will be deleted in the next major release. " - + "Use scopePassingSpanOperator() method instead"); - } - - return (sourcePub -> { - // TODO: Remove this once Reactor 3.1.8 is released - // do the checks directly on actual original Publisher - if (sourcePub instanceof ConnectableFlux // Operators.lift can't handle that - || sourcePub instanceof GroupedFlux // Operators.lift can't handle - // that - ) { - return sourcePub; - } - // no more POINTCUT_FILTER since mechanism is broken - Function, ? extends Publisher> lift = Operators - .lift((scannable, sub) -> { - if (contextRefreshed(beanFactory)) { - if (log.isTraceEnabled()) { - log.trace( - "Spring Context already refreshed. Creating a Sleuth span subscriber with Reactor Context " - + "[" + sub.currentContext() - + "] and name [" + scannable.name() - + "]"); - } - return spanSubscriptionProvider(beanFactory, scannable, sub) - .get(); - } - if (log.isTraceEnabled()) { - log.trace( - "Spring Context is not yet refreshed, falling back to lazy span subscriber. Reactor Context is [" - + sub.currentContext() + "] and name is [" - + scannable.name() + "]"); - } - // rest of the logic unchanged... - return new LazySpanSubscriber( - spanSubscriptionProvider(beanFactory, scannable, sub)); - }); - return lift.apply(sourcePub); - }); - } - private static SpanSubscriptionProvider spanSubscriptionProvider( BeanFactory beanFactory, Scannable scannable, CoreSubscriber sub) { return new SpanSubscriptionProvider(beanFactory, sub, sub.currentContext(), @@ -121,7 +65,7 @@ public abstract class ReactorSleuth { */ @SuppressWarnings("unchecked") public static Function, ? extends Publisher> scopePassingSpanOperator( - BeanFactory beanFactory) { + ConfigurableApplicationContext beanFactory) { return (sourcePub -> { // TODO: Remove this once Reactor 3.1.8 is released // do the checks directly on actual original Publisher @@ -135,7 +79,7 @@ public abstract class ReactorSleuth { Function, ? extends Publisher> lift = Operators .lift((scannable, sub) -> { // rest of the logic unchanged... - if (contextRefreshed(beanFactory)) { + if (beanFactory.isActive()) { if (log.isTraceEnabled()) { log.trace( "Spring Context already refreshed. Creating a scope " @@ -161,16 +105,6 @@ public abstract class ReactorSleuth { }); } - private static boolean contextRefreshed(BeanFactory beanFactory) { - try { - return beanFactory.getBean(ApplicationContextRefreshedListener.class) - .isRefreshed(); - } - catch (NoSuchBeanDefinitionException ex) { - return false; - } - } - private static SpanSubscriptionProvider scopePassingSpanSubscription( BeanFactory beanFactory, Scannable scannable, CoreSubscriber sub) { return new SpanSubscriptionProvider(beanFactory, sub, sub.currentContext(), 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 fa58837be..055dda68a 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 @@ -35,7 +35,9 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.cloud.sleuth.instrument.async.TraceableScheduledExecutorService; import org.springframework.cloud.sleuth.instrument.web.TraceWebFluxAutoConfiguration; +import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationListener; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.event.ContextRefreshedEvent; @@ -67,8 +69,9 @@ public class TraceReactorAutoConfiguration { @Bean // for tests @ConditionalOnMissingBean - static HookRegisteringBeanDefinitionRegistryPostProcessor traceHookRegisteringBeanDefinitionRegistryPostProcessor() { - return new HookRegisteringBeanDefinitionRegistryPostProcessor(); + static HookRegisteringBeanDefinitionRegistryPostProcessor traceHookRegisteringBeanDefinitionRegistryPostProcessor( + ConfigurableApplicationContext context) { + return new HookRegisteringBeanDefinitionRegistryPostProcessor(context); } @PreDestroy @@ -77,11 +80,6 @@ public class TraceReactorAutoConfiguration { Schedulers.resetFactory(); } - @Bean - ApplicationContextRefreshedListener traceApplicationContextRefreshedListener() { - return new ApplicationContextRefreshedListener(); - } - } } @@ -89,6 +87,12 @@ public class TraceReactorAutoConfiguration { class HookRegisteringBeanDefinitionRegistryPostProcessor implements BeanDefinitionRegistryPostProcessor { + private final ConfigurableApplicationContext context; + + HookRegisteringBeanDefinitionRegistryPostProcessor(ConfigurableApplicationContext context) { + this.context = context; + } + @Override public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException { @@ -103,7 +107,7 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor void setupHooks(BeanFactory beanFactory) { Hooks.onEachOperator( TraceReactorAutoConfiguration.TraceReactorConfiguration.SLEUTH_TRACE_REACTOR_KEY, - ReactorSleuth.scopePassingSpanOperator(beanFactory)); + ReactorSleuth.scopePassingSpanOperator(this.context)); Schedulers.setFactory(factoryInstance(beanFactory)); } @@ -112,25 +116,10 @@ class HookRegisteringBeanDefinitionRegistryPostProcessor @Override public ScheduledExecutorService decorateExecutorService(String schedulerType, Supplier actual) { - return new TraceableScheduledExecutorService(beanFactory, actual.get()); + return new TraceableScheduledExecutorService( + HookRegisteringBeanDefinitionRegistryPostProcessor.this.context, actual.get()); } }; } -} - -class ApplicationContextRefreshedListener - implements ApplicationListener { - - AtomicBoolean refreshed = new AtomicBoolean(); - - @Override - public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) { - this.refreshed.set(true); - } - - boolean isRefreshed() { - return this.refreshed.get(); - } - -} +} \ No newline at end of file diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/Issue866Configuration.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/Issue866Configuration.java index 845d09720..5987362d0 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/Issue866Configuration.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/Issue866Configuration.java @@ -18,6 +18,7 @@ package org.springframework.cloud.sleuth.instrument.reactor; import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -34,8 +35,8 @@ public class Issue866Configuration { public static TestHook hook; @Bean - HookRegisteringBeanDefinitionRegistryPostProcessor overridingProcessorForTests() { - TestHook hook = new TestHook(); + HookRegisteringBeanDefinitionRegistryPostProcessor overridingProcessorForTests(ConfigurableApplicationContext context) { + TestHook hook = new TestHook(context); Issue866Configuration.hook = hook; return hook; } @@ -45,6 +46,10 @@ public class Issue866Configuration { public boolean executed = false; + public TestHook(ConfigurableApplicationContext context) { + super(context); + } + @Override public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {