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
This commit is contained in:
Simon Baslé
2018-05-30 17:23:28 +02:00
committed by Marcin Grzejszczak
parent 31b3b1e9d6
commit 0f06e1e425
2 changed files with 31 additions and 10 deletions

View File

@@ -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;

View File

@@ -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);