fix #646 : Filter fused flows from tracing

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/TraceReactorAutoConfiguration.java

@@ -16,6 +16,8 @@ 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;
@@ -44,8 +46,15 @@ public class TraceReactorAutoConfiguration {
 
 		@PostConstruct
 		public void setupHooks() {
-			Hooks.onNewSubscriber((pub, sub) ->
-					new SpanSubscriber(sub, sub.currentContext(), this.tracer, pub.toString()));
+			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());
+			});
 			Schedulers.setFactory(new Schedulers.Factory() {
 				@Override public ScheduledExecutorService decorateScheduledExecutorService(
 						String schedulerType,

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriberTests.java

@@ -5,11 +5,11 @@ 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;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -21,8 +21,9 @@ 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 reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Hooks;
+import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
 import static org.assertj.core.api.BDDAssertions.then;
@@ -58,7 +59,71 @@ public class SpanSubscriberTests {
 				.subscribe(System.out::println);
 
 		then(this.tracer.getCurrentSpan()).isNull();
-		then(spanInOperation.get().getTraceId()).isEqualTo(span.getTraceId());
+		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<Span> 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<Subscription> spanInOperation = new AtomicReference<>();
+		log.info("Hello");
+
+		Mono.just(1)
+				.subscribe(new BaseSubscriber<Integer>() {
+					@Override
+					protected void hookOnSubscribe(Subscription subscription) {
+						spanInOperation.set(subscription);
+					}
+				});
+
+		then(this.tracer.getCurrentSpan()).isNotNull();
+		then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class);
+
+		Mono.<Integer>error(new Exception())
+				.subscribe(new BaseSubscriber<Integer>() {
+					@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.<Integer>empty()
+				.subscribe(new BaseSubscriber<Integer>() {
+					@Override
+					protected void hookOnSubscribe(Subscription subscription) {
+						spanInOperation.set(subscription);
+					}
+				});
+
+		then(this.tracer.getCurrentSpan()).isNotNull();
+		then(spanInOperation.get()).isNotInstanceOf(SpanSubscriber.class);
 		then(ExceptionUtils.getLastException()).isNull();
 	}
 
@@ -107,16 +172,10 @@ 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().getTraceId()).isEqualTo(foo2.getTraceId());
+		then(spanInOperation.get().getSavedSpan().getParents().get(0)).isEqualTo(foo2.getSpanId());
 		tracer.close(foo2);
 	}
 
-	@AfterClass
-	public static void cleanup() {
-		Hooks.resetOnNewSubscriber();
-		Schedulers.resetFactory();
-	}
-
 	@EnableAutoConfiguration
 	@Configuration
 	static class Config {