From 0f06e1e4252d2b1ff00dbe0a59740e11cf1bb0d9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Simon=20Basl=C3=A9?= <sbasle@pivotal.io>
Date: Wed, 30 May 2018 17:23:28 +0200
Subject: [PATCH] fix #994 Make SpanSubscriber fake fuseability (#995)

This is done by implementing NO-OP QueueSubscription on the subscriber
and removing the guard against `Fuseable` publisher sources.

As `Fuseable` is a marker interface there should be no reason for code
to perform casts so for now the fact that the `lift` call in
ReactorSleuth doesn't actually produces a `Fuseable` publisher should
not be a problem.

fixes gh-994
---
 .../instrument/reactor/ReactorSleuth.java     | 13 +++------
 .../instrument/reactor/SpanSubscriber.java    | 28 ++++++++++++++++++-
 2 files changed, 31 insertions(+), 10 deletions(-)

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 b959cd0b6..be63caba3 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
@@ -16,8 +16,6 @@
 
 package org.springframework.cloud.sleuth.instrument.reactor;
 
-import java.util.function.Function;
-
 import brave.Tracing;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -25,13 +23,14 @@ import org.reactivestreams.Publisher;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import reactor.core.CoreSubscriber;
-import reactor.core.Fuseable;
 import reactor.core.Scannable;
 import reactor.core.publisher.ConnectableFlux;
 import reactor.core.publisher.GroupedFlux;
 import reactor.core.publisher.Operators;
 import reactor.util.context.Context;
 
+import java.util.function.Function;
+
 /**
  * Reactive Span pointcuts factories
  *
@@ -60,9 +59,7 @@ public abstract class ReactorSleuth {
 		return sourcePub -> {
 			// TODO: Remove this once Reactor 3.1.8 is released
 			//do the checks directly on actual original Publisher
-			if (sourcePub instanceof Fuseable //Sleuth can't handle that
-					|| sourcePub instanceof Fuseable.ScalarCallable //IIRC Sleuth can't handle that
-					|| sourcePub instanceof ConnectableFlux //Operators.lift can't handle that
+			if (sourcePub instanceof ConnectableFlux //Operators.lift can't handle that
 					|| sourcePub instanceof GroupedFlux  //Operators.lift can't handle that
 					) {
 				return sourcePub;
@@ -116,9 +113,7 @@ public abstract class ReactorSleuth {
 		return sourcePub -> {
 			// TODO: Remove this once Reactor 3.1.8 is released
 			//do the checks directly on actual original Publisher
-			if (sourcePub instanceof Fuseable //Sleuth can't handle that
-					|| sourcePub instanceof Fuseable.ScalarCallable //IIRC Sleuth can't handle that
-					|| sourcePub instanceof ConnectableFlux //Operators.lift can't handle that
+			if (sourcePub instanceof ConnectableFlux //Operators.lift can't handle that
 					|| sourcePub instanceof GroupedFlux  //Operators.lift can't handle that
 					) {
 				return sourcePub;
diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java
index 26ff2ea33..3b6a01f24 100644
--- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java
+++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriber.java
@@ -24,6 +24,7 @@ import brave.Tracing;
 import brave.propagation.TraceContextOrSamplingFlags;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import reactor.core.Fuseable;
 import reactor.util.Logger;
 import reactor.util.Loggers;
 import reactor.util.context.Context;
@@ -35,7 +36,32 @@ import reactor.util.context.Context;
  * @author Marcin Grzejszczak
  * @since 2.0.0
  */
-final class SpanSubscriber<T> extends AtomicBoolean implements SpanSubscription<T> {
+final class SpanSubscriber<T> extends AtomicBoolean implements SpanSubscription<T>, Fuseable.QueueSubscription<T> {
+
+	@Override
+	public T poll() {
+		return null;
+	}
+
+	@Override
+	public int requestFusion(int i) {
+		return Fuseable.NONE; //always negotiate to no fusion
+	}
+
+	@Override
+	public int size() {
+		return 0;
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return true;
+	}
+
+	@Override
+	public void clear() {
+		//NO-OP
+	}
 
 	private static final Logger log = Loggers.getLogger(
 			SpanSubscriber.class);