From 2ebba99c4ef87e207fc611e1a3a485e374d23e57 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Thu, 20 Jul 2017 11:58:10 +0200 Subject: [PATCH] Reverting the change --- .../TraceReactorAutoConfiguration.java | 13 +-- .../reactor/SpanSubscriberTests.java | 81 +++---------------- 2 files changed, 12 insertions(+), 82 deletions(-) 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 123f769ac..5890d2bd3 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 @@ -16,8 +16,6 @@ import org.springframework.cloud.sleuth.Tracer; import org.springframework.cloud.sleuth.instrument.async.TraceableScheduledExecutorService; import org.springframework.cloud.sleuth.instrument.web.TraceWebFluxAutoConfiguration; import org.springframework.context.annotation.Configuration; - -import reactor.core.Fuseable; import reactor.core.publisher.Hooks; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; @@ -46,15 +44,8 @@ public class TraceReactorAutoConfiguration { @PostConstruct public void setupHooks() { - Hooks.onNewSubscriber((pub, sub) -> { - //do not trace fused flows or simple just/error/empty - if(pub instanceof Fuseable && sub instanceof Fuseable.QueueSubscription - || pub instanceof Fuseable.ScalarCallable){ - return sub; - } - return new SpanSubscriber(sub, sub.currentContext(), this.tracer, pub - .toString()); - }); + Hooks.onNewSubscriber((pub, sub) -> + new SpanSubscriber(sub, sub.currentContext(), this.tracer, pub.toString())); Schedulers.setFactory(new Schedulers.Factory() { @Override public ScheduledExecutorService decorateScheduledExecutorService( String schedulerType, diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java index f3e2146d3..eac363666 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java @@ -5,6 +5,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.awaitility.Awaitility; +import org.junit.AfterClass; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -20,12 +21,8 @@ import org.springframework.cloud.sleuth.util.ExceptionUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.test.context.junit4.SpringRunner; - -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; -import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; +import reactor.core.publisher.Hooks; import reactor.core.scheduler.Schedulers; import static org.assertj.core.api.BDDAssertions.then; @@ -61,71 +58,7 @@ public class SpanSubscriberTests { .subscribe(System.out::println); then(this.tracer.getCurrentSpan()).isNull(); - then(spanInOperation.get().getParents().get(0)).isEqualTo(span.getSpanId()); - then(ExceptionUtils.getLastException()).isNull(); - } - - @Test public void should_support_reactor_fusion_optimization() { - Span span = this.tracer.createSpan("foo"); - final AtomicReference spanInOperation = new AtomicReference<>(); - log.info("Hello"); - - Mono.just(1) - .flatMap( d -> Flux.just(d + 1).collectList().map(p -> p.get(0))) - .map( d -> d + 1) - .map( (d) -> { - spanInOperation.set(SpanSubscriberTests.this.tracer.getCurrentSpan()); - return d + 1; - }) - .map( d -> d + 1) - .subscribe(System.out::println); - - then(this.tracer.getCurrentSpan()).isNull(); - then(spanInOperation.get().getParents().get(0)).isEqualTo(span.getSpanId()); - then(ExceptionUtils.getLastException()).isNull(); - } - - @Test public void should_not_trace_scalar_flows() { - Span span = this.tracer.createSpan("foo"); - final AtomicReference spanInOperation = new AtomicReference<>(); - log.info("Hello"); - - Mono.just(1) - .subscribe(new BaseSubscriber() { - @Override - protected void hookOnSubscribe(Subscription subscription) { - spanInOperation.set(subscription); - } - }); - - then(this.tracer.getCurrentSpan()).isNotNull(); - then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); - - Mono.error(new Exception()) - .subscribe(new BaseSubscriber() { - @Override - protected void hookOnSubscribe(Subscription subscription) { - spanInOperation.set(subscription); - } - - @Override - protected void hookOnError(Throwable throwable) { - } - }); - - then(this.tracer.getCurrentSpan()).isNotNull(); - then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); - - Mono.empty() - .subscribe(new BaseSubscriber() { - @Override - protected void hookOnSubscribe(Subscription subscription) { - spanInOperation.set(subscription); - } - }); - - then(this.tracer.getCurrentSpan()).isNotNull(); - then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class); + then(spanInOperation.get().getTraceId()).isEqualTo(span.getTraceId()); then(ExceptionUtils.getLastException()).isNull(); } @@ -174,10 +107,16 @@ public class SpanSubscriberTests { then(this.tracer.getCurrentSpan()).isEqualTo(foo2); then(ExceptionUtils.getLastException()).isNull(); // parent cause there's an async span in the meantime - then(spanInOperation.get().getSavedSpan().getParents().get(0)).isEqualTo(foo2.getSpanId()); + then(spanInOperation.get().getTraceId()).isEqualTo(foo2.getTraceId()); tracer.close(foo2); } + @AfterClass + public static void cleanup() { + Hooks.resetOnNewSubscriber(); + Schedulers.resetFactory(); + } + @EnableAutoConfiguration @Configuration static class Config {