Merge remote-tracking branch 'origin/main' into 3.1.x
This commit is contained in:
Marcin Grzejszczak
@@ -18,24 +18,11 @@ package org.springframework.cloud.sleuth.instrument.reactor;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.function.BiFunction;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Predicate;
|
||||
|
||||
import org.reactivestreams.Processor;
|
||||
import org.reactivestreams.Publisher;
|
||||
import reactor.core.CoreSubscriber;
|
||||
import reactor.core.Disposable;
|
||||
import reactor.core.Fuseable;
|
||||
import reactor.core.Scannable;
|
||||
import reactor.core.publisher.ConnectableFlux;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.FluxOperator;
|
||||
import reactor.core.publisher.GroupedFlux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.publisher.MonoOperator;
|
||||
import reactor.core.publisher.Operators;
|
||||
import reactor.core.publisher.ParallelFlux;
|
||||
import reactor.util.annotation.Nullable;
|
||||
|
||||
import org.springframework.util.Assert;
|
||||
@@ -87,9 +74,13 @@ import org.springframework.util.Assert;
|
||||
* })
|
||||
* .subscribe();
|
||||
*}</pre>
|
||||
*
|
||||
* @author Roman Matiushchenko
|
||||
*/
|
||||
final class ReactorHooksHelper {
|
||||
|
||||
static final String LIFTER_NAME = "org.springframework.cloud.sleuth.instrument.reactor.ReactorHooksHelper.ScopePassingLifter";
|
||||
|
||||
// need a way to determine SYNC sources to not add redundant scope passing decorator
|
||||
// most of reactor-core SYNC sources are marked with SourceProducer interface
|
||||
static final Class<?> sourceProducerClass;
|
||||
@@ -132,6 +123,10 @@ final class ReactorHooksHelper {
|
||||
}
|
||||
|
||||
if (!isSync(current)) {
|
||||
boolean isLifter = getLifterName(current) != null;
|
||||
if (isLifter) {
|
||||
return shouldDecorateLifter(current);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -143,8 +138,41 @@ final class ReactorHooksHelper {
|
||||
}
|
||||
}
|
||||
|
||||
static boolean isTraceContextPropagator(Publisher<?> current) {
|
||||
return current instanceof TraceContextPropagator;
|
||||
/**
|
||||
* xxxLift Publishers get their RunStyle from source Publisher. So need to check
|
||||
* whether current chain was decorated with scope passing operator.
|
||||
* @param p first not sync lifter Publisher in the chain.
|
||||
* @return {@code true} if Publisher chain does not contain lifter with
|
||||
* {@link #LIFTER_NAME} name.
|
||||
*/
|
||||
private static boolean shouldDecorateLifter(Publisher<?> p) {
|
||||
Publisher<?> current = getParent(p);
|
||||
while (true) {
|
||||
if (current == null) {
|
||||
// is start of the chain, Publisher without source or foreign Publisher
|
||||
return true;
|
||||
}
|
||||
String lifterName = getLifterName(current);
|
||||
if (isScopePassingLifter(lifterName)) {
|
||||
return false;
|
||||
}
|
||||
if (lifterName == null) {
|
||||
return true;
|
||||
}
|
||||
current = getParent(current);
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isScopePassingLifter(String lifterName) {
|
||||
return lifterName == LIFTER_NAME;
|
||||
}
|
||||
|
||||
private static String getLifterName(Publisher<?> current) {
|
||||
return Scannable.from(current).scan(Scannable.Attr.LIFTER);
|
||||
}
|
||||
|
||||
public static boolean isTraceContextPropagator(Publisher<?> current) {
|
||||
return current instanceof TraceContextPropagator || isScopePassingLifter(getLifterName(current));
|
||||
}
|
||||
|
||||
private static boolean isSourceProducer(Publisher<?> p) {
|
||||
@@ -166,299 +194,39 @@ final class ReactorHooksHelper {
|
||||
}
|
||||
|
||||
/**
|
||||
* Decorates {@link Publisher} with {@link TraceContextPropagator} {@code Publisher}.
|
||||
* Mostly it is a copy of reactor-core logic from
|
||||
* {@code reactor.core.publisher.Operators#liftPublisher()}.
|
||||
* @param filter the Predicate that the raw Publisher must pass for the transformation
|
||||
* to occur
|
||||
* @param lifter the {@link BiFunction} taking the raw {@link Publisher} and
|
||||
* {@link CoreSubscriber}. The function must return a receiving {@link CoreSubscriber}
|
||||
* that will immediately subscribe to the {@link Publisher}.
|
||||
* @param <O> the input and output type.
|
||||
* @return a new {@link Function}.
|
||||
* @param name function name.
|
||||
* @param delegate delegate function.
|
||||
* @param <T> the type of the first argument to the function
|
||||
* @param <U> the type of the second argument to the function
|
||||
* @param <R> the type of the result of the function
|
||||
* @return function that {@link Object#toString()} returns provided name which is used
|
||||
* as a value of {@link Scannable.Attr#LIFTER} attribute.
|
||||
*/
|
||||
public static <O> Function<? super Publisher<O>, ? extends Publisher<O>> liftPublisher(Predicate<Publisher> filter,
|
||||
BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super O>> lifter) {
|
||||
Assert.notNull(lifter, "lifter is null");
|
||||
return publisher -> {
|
||||
if (filter != null && !filter.test(publisher)) {
|
||||
return (Publisher<O>) publisher;
|
||||
}
|
||||
static <T, U, R> BiFunction<T, U, R> named(String name, BiFunction<T, U, R> delegate) {
|
||||
return new NamedLifter<>(name, delegate);
|
||||
}
|
||||
|
||||
static class NamedLifter<T, U, R> implements BiFunction<T, U, R> {
|
||||
|
||||
private final BiFunction<T, U, R> delegate;
|
||||
|
||||
private final String name;
|
||||
|
||||
NamedLifter(String name, BiFunction<T, U, R> delegate) {
|
||||
this.name = Objects.requireNonNull(name, "name");
|
||||
this.delegate = Objects.requireNonNull(delegate, "delegate");
|
||||
}
|
||||
|
||||
@Override
|
||||
public R apply(T t, U u) {
|
||||
return delegate.apply(t, u);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return name;
|
||||
}
|
||||
|
||||
if (publisher instanceof Mono) {
|
||||
return new SleuthMonoLift<>(publisher, lifter);
|
||||
}
|
||||
if (publisher instanceof ParallelFlux) {
|
||||
return new SleuthParallelLift<>((ParallelFlux<O>) publisher, lifter);
|
||||
}
|
||||
if (publisher instanceof ConnectableFlux) {
|
||||
return new SleuthConnectableLift<>((ConnectableFlux<O>) publisher, lifter);
|
||||
}
|
||||
if (publisher instanceof GroupedFlux) {
|
||||
return new SleuthGroupedLift<>((GroupedFlux<?, O>) publisher, lifter);
|
||||
}
|
||||
return new SleuthFluxLift<>(publisher, lifter);
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Copy of {@code reactor.core.publisher.MonoLift} which implements
|
||||
* {@link TraceContextPropagator}.
|
||||
*/
|
||||
final class SleuthMonoLift<I, O> extends MonoOperator<I, O> implements TraceContextPropagator {
|
||||
|
||||
final BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter;
|
||||
|
||||
SleuthMonoLift(Publisher<I> p,
|
||||
BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter) {
|
||||
super(Mono.from(p));
|
||||
this.lifter = lifter;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subscribe(CoreSubscriber<? super O> actual) {
|
||||
CoreSubscriber input = actual;
|
||||
try {
|
||||
input = Objects.requireNonNull(lifter.apply(source, actual), "Lifted subscriber MUST NOT be null");
|
||||
|
||||
source.subscribe(input);
|
||||
}
|
||||
catch (Throwable e) {
|
||||
Operators.reportThrowInSubscribe(input, e);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object scanUnsafe(Attr key) {
|
||||
if (key == Attr.RUN_STYLE) {
|
||||
return Attr.RunStyle.SYNC;
|
||||
}
|
||||
return super.scanUnsafe(key);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Copy of {@code reactor.core.publisher.FluxLift} which implements
|
||||
* {@link TraceContextPropagator}.
|
||||
*/
|
||||
final class SleuthFluxLift<I, O> extends FluxOperator<I, O> implements TraceContextPropagator {
|
||||
|
||||
final BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter;
|
||||
|
||||
SleuthFluxLift(Publisher<I> p,
|
||||
BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter) {
|
||||
super(Flux.from(p));
|
||||
this.lifter = lifter;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subscribe(CoreSubscriber<? super O> actual) {
|
||||
CoreSubscriber input = actual;
|
||||
try {
|
||||
input = Objects.requireNonNull(lifter.apply(source, actual), "Lifted subscriber MUST NOT be null");
|
||||
|
||||
source.subscribe(input);
|
||||
}
|
||||
catch (Throwable e) {
|
||||
Operators.reportThrowInSubscribe(input, e);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object scanUnsafe(Attr key) {
|
||||
if (key == Attr.RUN_STYLE) {
|
||||
return Attr.RunStyle.SYNC;
|
||||
}
|
||||
return super.scanUnsafe(key);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Copy of {@code reactor.core.publisher.ConnectableLift} which implements
|
||||
* {@link TraceContextPropagator}.
|
||||
*/
|
||||
final class SleuthConnectableLift<I, O> extends ConnectableFlux<O> implements Scannable, TraceContextPropagator {
|
||||
|
||||
final ConnectableFlux<I> source;
|
||||
|
||||
final BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter;
|
||||
|
||||
SleuthConnectableLift(ConnectableFlux<I> p,
|
||||
BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter) {
|
||||
this.source = Objects.requireNonNull(p, "source");
|
||||
this.lifter = lifter;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPrefetch() {
|
||||
return source.getPrefetch();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void connect(Consumer<? super Disposable> cancelSupport) {
|
||||
this.source.connect(cancelSupport);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Nullable
|
||||
public Object scanUnsafe(Attr key) {
|
||||
if (key == Attr.PREFETCH) {
|
||||
return source.getPrefetch();
|
||||
}
|
||||
if (key == Attr.PARENT) {
|
||||
return source;
|
||||
}
|
||||
if (key == Attr.RUN_STYLE) {
|
||||
return Attr.RunStyle.SYNC;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subscribe(CoreSubscriber<? super O> actual) {
|
||||
CoreSubscriber input = actual;
|
||||
try {
|
||||
input = Objects.requireNonNull(lifter.apply(source, actual), "Lifted subscriber MUST NOT be null");
|
||||
|
||||
source.subscribe(input);
|
||||
}
|
||||
catch (Throwable e) {
|
||||
Operators.reportThrowInSubscribe(input, e);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Copy of {@code reactor.core.publisher.GroupedLift} which implements
|
||||
* {@link TraceContextPropagator}.
|
||||
*/
|
||||
final class SleuthGroupedLift<K, I, O> extends GroupedFlux<K, O> implements Scannable, TraceContextPropagator {
|
||||
|
||||
final BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter;
|
||||
|
||||
final GroupedFlux<K, I> source;
|
||||
|
||||
SleuthGroupedLift(GroupedFlux<K, I> p,
|
||||
BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter) {
|
||||
this.source = Objects.requireNonNull(p, "source");
|
||||
this.lifter = lifter;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPrefetch() {
|
||||
return source.getPrefetch();
|
||||
}
|
||||
|
||||
@Override
|
||||
public K key() {
|
||||
return source.key();
|
||||
}
|
||||
|
||||
@Override
|
||||
@Nullable
|
||||
public Object scanUnsafe(Attr key) {
|
||||
if (key == Attr.PARENT) {
|
||||
return source;
|
||||
}
|
||||
if (key == Attr.PREFETCH) {
|
||||
return getPrefetch();
|
||||
}
|
||||
if (key == Attr.RUN_STYLE) {
|
||||
return Attr.RunStyle.SYNC;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String stepName() {
|
||||
if (source instanceof Scannable) {
|
||||
return Scannable.from(source).stepName();
|
||||
}
|
||||
return Scannable.super.stepName();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subscribe(CoreSubscriber<? super O> actual) {
|
||||
CoreSubscriber<? super I> input = lifter.apply(source, actual);
|
||||
|
||||
Objects.requireNonNull(input, "Lifted subscriber MUST NOT be null");
|
||||
|
||||
source.subscribe(input);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Copy of {@code reactor.core.publisher.ParallelLift} which implements
|
||||
* {@link TraceContextPropagator}.
|
||||
*/
|
||||
final class SleuthParallelLift<I, O> extends ParallelFlux<O> implements Scannable, TraceContextPropagator {
|
||||
|
||||
final BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter;
|
||||
|
||||
final ParallelFlux<I> source;
|
||||
|
||||
SleuthParallelLift(ParallelFlux<I> p,
|
||||
BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super I>> lifter) {
|
||||
this.source = Objects.requireNonNull(p, "source");
|
||||
this.lifter = lifter;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPrefetch() {
|
||||
return source.getPrefetch();
|
||||
}
|
||||
|
||||
@Override
|
||||
public int parallelism() {
|
||||
return source.parallelism();
|
||||
}
|
||||
|
||||
@Override
|
||||
@Nullable
|
||||
public Object scanUnsafe(Attr key) {
|
||||
if (key == Attr.PARENT) {
|
||||
return source;
|
||||
}
|
||||
if (key == Attr.PREFETCH) {
|
||||
return getPrefetch();
|
||||
}
|
||||
if (key == Attr.RUN_STYLE) {
|
||||
return Attr.RunStyle.SYNC;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String stepName() {
|
||||
if (source instanceof Scannable) {
|
||||
return Scannable.from(source).stepName();
|
||||
}
|
||||
return Scannable.super.stepName();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subscribe(CoreSubscriber<? super O>[] s) {
|
||||
@SuppressWarnings("unchecked")
|
||||
CoreSubscriber<? super I>[] subscribers = new CoreSubscriber[parallelism()];
|
||||
|
||||
int i = 0;
|
||||
while (i < subscribers.length) {
|
||||
subscribers[i] = Objects.requireNonNull(lifter.apply(source, s[i]), "Lifted subscriber MUST NOT be null");
|
||||
i++;
|
||||
}
|
||||
|
||||
source.subscribe(subscribers);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -45,10 +45,13 @@ import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.lang.NonNull;
|
||||
import org.springframework.lang.Nullable;
|
||||
|
||||
import static org.springframework.cloud.sleuth.instrument.reactor.ReactorHooksHelper.named;
|
||||
|
||||
/**
|
||||
* Reactive Span pointcuts factories.
|
||||
*
|
||||
* @author Stephane Maldini
|
||||
* @author Roman Matiushchenko
|
||||
* @since 2.0.0
|
||||
*/
|
||||
// TODO: this is public as it is used out of package, but unlikely intended to be
|
||||
@@ -121,7 +124,7 @@ public abstract class ReactorSleuth {
|
||||
BiFunction<Publisher, ? super CoreSubscriber<? super T>, ? extends CoreSubscriber<? super T>> lifter = liftFunction(
|
||||
springContext, lazyCurrentTraceContext, lazyTracer);
|
||||
|
||||
return ReactorHooksHelper.liftPublisher(shouldDecorate, lifter);
|
||||
return Operators.liftPublisher(shouldDecorate, named(ReactorHooksHelper.LIFTER_NAME, lifter));
|
||||
}
|
||||
|
||||
static <O> BiFunction<Publisher, ? super CoreSubscriber<? super O>, ? extends CoreSubscriber<? super O>> liftFunction(
|
||||
@@ -253,7 +256,7 @@ public abstract class ReactorSleuth {
|
||||
return scopePassingSpanSubscriber.apply(pub, sub);
|
||||
};
|
||||
|
||||
return ReactorHooksHelper.liftPublisher(p -> {
|
||||
return Operators.liftPublisher(p -> {
|
||||
/*
|
||||
* this prevent double decoration when last operator in the chain is not SYNC
|
||||
* like {@code Mono.fromSuppler(() -> ...).subscribeOn(Schedulers.parallel())}
|
||||
@@ -269,7 +272,7 @@ public abstract class ReactorSleuth {
|
||||
}
|
||||
}
|
||||
return addContext;
|
||||
}, skipIfNoTraceCtx);
|
||||
}, named(ReactorHooksHelper.LIFTER_NAME, skipIfNoTraceCtx));
|
||||
}
|
||||
|
||||
private static <T> Context context(CoreSubscriber<? super T> sub) {
|
||||
|
||||
@@ -17,36 +17,24 @@
|
||||
package org.springframework.cloud.sleuth.instrument.reactor;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.BiFunction;
|
||||
import java.util.function.Function;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.Mockito;
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.reactivestreams.Subscription;
|
||||
import reactor.core.CoreSubscriber;
|
||||
import reactor.core.Disposable;
|
||||
import reactor.core.Scannable;
|
||||
import reactor.core.publisher.BaseSubscriber;
|
||||
import reactor.core.publisher.ConnectableFlux;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.GroupedFlux;
|
||||
import reactor.core.publisher.Hooks;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.publisher.MonoOperator;
|
||||
import reactor.core.publisher.Operators;
|
||||
import reactor.core.publisher.ParallelFlux;
|
||||
import reactor.core.publisher.Sinks;
|
||||
import reactor.core.scheduler.Schedulers;
|
||||
import reactor.test.StepVerifier;
|
||||
import reactor.util.annotation.Nullable;
|
||||
import reactor.util.context.Context;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatCode;
|
||||
import static org.springframework.cloud.sleuth.instrument.reactor.ReactorHooksHelper.named;
|
||||
|
||||
class ReactorHooksHelperTests {
|
||||
|
||||
@@ -114,8 +102,16 @@ class ReactorHooksHelperTests {
|
||||
|
||||
@Test
|
||||
public void shouldDecorateWhenAsyncSourceAndTraceContextPropagatorThanShouldNotDecorate() {
|
||||
Function<? super Publisher<Long>, ? extends Publisher<Long>> traceContextPropagator = ReactorHooksHelper
|
||||
.liftPublisher(publisher -> true, (publisher, coreSubscriber) -> coreSubscriber);
|
||||
Mono<?> syncSourceWithLifter = Mono.delay(Duration.ofMillis(10)).as(TraceContextPropagatorOperator::new);
|
||||
boolean actual = ReactorHooksHelper.shouldDecorate(syncSourceWithLifter.map(Function.identity()));
|
||||
assertThat(actual).isFalse();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldDecorateWhenAsyncSourceAndScopePassingLifterThanShouldNotDecorate() {
|
||||
Function<? super Publisher<Long>, ? extends Publisher<Long>> traceContextPropagator = Operators.liftPublisher(
|
||||
publisher -> true,
|
||||
named(ReactorHooksHelper.LIFTER_NAME, (publisher, coreSubscriber) -> coreSubscriber));
|
||||
Mono<?> syncSourceWithLifter = Mono.delay(Duration.ofMillis(10)).transform(traceContextPropagator);
|
||||
boolean actual = ReactorHooksHelper.shouldDecorate(syncSourceWithLifter.map(Function.identity()));
|
||||
assertThat(actual).isFalse();
|
||||
@@ -134,182 +130,13 @@ class ReactorHooksHelperTests {
|
||||
assertThat(actual).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void liftPublisherWhenFilterDiscardsThenReturnSource() {
|
||||
Mono<Integer> source = Mono.just(1);
|
||||
final Function<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = ReactorHooksHelper
|
||||
.liftPublisher(p -> false, (p, s) -> s);
|
||||
Mono<Integer> transformed = source.transform(transformer);
|
||||
assertThat(source).isSameAs(transformed);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void liftPublisherWhenFilterIsNullThenDecorate() {
|
||||
Mono<Integer> source = Mono.just(1);
|
||||
final Function<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = ReactorHooksHelper
|
||||
.liftPublisher(null, (p, s) -> s);
|
||||
Mono<Integer> transformed = source.transform(transformer);
|
||||
assertThat(source).isNotSameAs(transformed);
|
||||
assertThat(transformed).isInstanceOf(SleuthMonoLift.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void liftPublisherWhenSourceMonoThenDecorateWithMono() {
|
||||
Mono<Integer> source = Mono.just(1);
|
||||
final Function<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = ReactorHooksHelper
|
||||
.liftPublisher(p -> true, (p, s) -> new PlusOneSubscriber(s));
|
||||
Mono<Integer> transformed = source.transform(transformer);
|
||||
assertThat(source).isNotSameAs(transformed);
|
||||
assertThat(transformed).isInstanceOf(SleuthMonoLift.class).isInstanceOf(TraceContextPropagator.class);
|
||||
|
||||
StepVerifier.create(transformed).expectSubscription().expectNext(2).expectComplete()
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void liftPublisherWhenSourceFluxThenDecorateWithFlux() {
|
||||
Flux<Integer> source = Flux.just(1, 2, 3);
|
||||
final Function<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = ReactorHooksHelper
|
||||
.liftPublisher(p -> true, (p, s) -> new PlusOneSubscriber(s));
|
||||
Flux<Integer> transformed = source.transform(transformer);
|
||||
assertThat(source).isNotSameAs(transformed);
|
||||
assertThat(transformed).isInstanceOf(SleuthFluxLift.class).isInstanceOf(TraceContextPropagator.class);
|
||||
|
||||
StepVerifier.create(transformed).expectSubscription().expectNext(2).expectNext(3).expectNext(4).expectComplete()
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void liftPublisherWhenSourceParallelFluxThenDecorateWithParallelFlux() {
|
||||
ParallelFlux<Integer> source = Flux.just(1, 2, 3).parallel();
|
||||
|
||||
Function function = ReactorHooksHelper.liftPublisher(p -> true,
|
||||
(p, s) -> (CoreSubscriber) new PlusOneSubscriber(s));
|
||||
|
||||
ParallelFlux<Integer> transformed = source.transform(function);
|
||||
assertThat(source).isNotSameAs(transformed);
|
||||
assertThat(transformed).isInstanceOf(SleuthParallelLift.class).isInstanceOf(TraceContextPropagator.class);
|
||||
Flux<Integer> sorted = transformed.map(it -> it).sequential().sort();
|
||||
StepVerifier.create(sorted).expectSubscription().expectNext(2).expectNext(3).expectNext(4).expectComplete()
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void liftPublisherWhenSourceConnectableFluxThenDecorateWithConnectableFlux() {
|
||||
final AtomicBoolean sourceCanceled = new AtomicBoolean();
|
||||
ConnectableFlux<Integer> source = Flux.just(1, 2, 3).doOnCancel(() -> sourceCanceled.set(true)).replay();
|
||||
|
||||
Function function = ReactorHooksHelper.liftPublisher(p -> true,
|
||||
(p, s) -> (CoreSubscriber) new PlusOneSubscriber(s));
|
||||
|
||||
Flux<Integer> transformed = source.transform(function);
|
||||
assertThat(source).isNotSameAs(transformed);
|
||||
assertThat(transformed).isInstanceOf(SleuthConnectableLift.class).isInstanceOf(TraceContextPropagator.class);
|
||||
final AtomicReference<Disposable> cancelSource = new AtomicReference<>();
|
||||
Flux<Integer> autoConnect = ((ConnectableFlux<Integer>) transformed).autoConnect(1, cancelSource::set);
|
||||
|
||||
StepVerifier.create(autoConnect).expectSubscription().expectNext(2).expectNext(3).expectNext(4).thenCancel()
|
||||
.verify(Duration.ofSeconds(5));
|
||||
|
||||
assertThat(cancelSource.get()).isNotNull();
|
||||
cancelSource.get().dispose();
|
||||
assertThat(sourceCanceled).isTrue();
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void liftPublisherWhenSourceGroupedFluxThenDecorateWithGroupedFlux() {
|
||||
Flux<GroupedFlux<Integer, Integer>> source = Flux.just(1, 2, 3).groupBy(i -> i % 2);
|
||||
|
||||
Function function = ReactorHooksHelper.liftPublisher(p -> true,
|
||||
(p, s) -> (CoreSubscriber) new PlusOneSubscriber(s));
|
||||
|
||||
GroupedFlux<Integer, Integer> grouped = source.filter(it -> it.key() == 1).blockFirst();
|
||||
Flux<Integer> transformed = grouped.transform(function);
|
||||
assertThat(grouped).isNotSameAs(transformed);
|
||||
assertThat(transformed).isInstanceOf(SleuthGroupedLift.class).isInstanceOf(TraceContextPropagator.class);
|
||||
|
||||
assertThat(((GroupedFlux) transformed).key()).isEqualTo(1);
|
||||
|
||||
StepVerifier.create(transformed).expectSubscription().expectNext(2).expectNext(4).expectComplete()
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void sleuthMonoLiftWhenScanRunStyleThenSync() {
|
||||
final Mono source = Mockito.mock(Mono.class);
|
||||
|
||||
SleuthMonoLift lifter = new SleuthMonoLift(source, (p, s) -> s);
|
||||
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.RUN_STYLE)).isEqualTo(Scannable.Attr.RunStyle.SYNC);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PARENT)).isEqualTo(source);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PREFETCH)).isEqualTo(Integer.MAX_VALUE);
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void testSleuthFluxLiftScanUnsafe() {
|
||||
final Flux source = Mockito.mock(Flux.class);
|
||||
|
||||
SleuthFluxLift lifter = new SleuthFluxLift(source, (p, s) -> s);
|
||||
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.RUN_STYLE)).isEqualTo(Scannable.Attr.RunStyle.SYNC);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PARENT)).isEqualTo(source);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PREFETCH)).isEqualTo(lifter.getPrefetch());
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void testSleuthConnectableLiftScanUnsafe() {
|
||||
final ConnectableFlux source = Mockito.mock(ConnectableFlux.class);
|
||||
int sourcePrefetch = 1;
|
||||
Mockito.when(source.getPrefetch()).thenReturn(sourcePrefetch);
|
||||
|
||||
SleuthConnectableLift lifter = new SleuthConnectableLift(source, (p, s) -> s);
|
||||
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.RUN_STYLE)).isEqualTo(Scannable.Attr.RunStyle.SYNC);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PARENT)).isEqualTo(source);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PREFETCH)).isEqualTo(sourcePrefetch);
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void testSleuthGroupedLiftFluxScanUnsafe() {
|
||||
final GroupedFlux source = Mockito.mock(GroupedFlux.class);
|
||||
int sourcePrefetch = 1;
|
||||
Mockito.when(source.getPrefetch()).thenReturn(sourcePrefetch);
|
||||
|
||||
SleuthGroupedLift lifter = new SleuthGroupedLift(source, (p, s) -> s);
|
||||
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.RUN_STYLE)).isEqualTo(Scannable.Attr.RunStyle.SYNC);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PARENT)).isEqualTo(source);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PREFETCH)).isEqualTo(sourcePrefetch);
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
public void testSleuthParallelLiftScanUnsafe() {
|
||||
final ParallelFlux source = Mockito.mock(ParallelFlux.class);
|
||||
int sourcePrefetch = 1;
|
||||
Mockito.when(source.getPrefetch()).thenReturn(sourcePrefetch);
|
||||
|
||||
SleuthParallelLift lifter = new SleuthParallelLift(source, (p, s) -> s);
|
||||
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.RUN_STYLE)).isEqualTo(Scannable.Attr.RunStyle.SYNC);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PARENT)).isEqualTo(source);
|
||||
assertThat(lifter.scanUnsafe(Scannable.Attr.PREFETCH)).isEqualTo(sourcePrefetch);
|
||||
}
|
||||
|
||||
static class CustomMonoWithoutRunStyleOperator<O> extends MonoOperator<O, O> {
|
||||
|
||||
/**
|
||||
* Build a {@link MonoOperator} wrapper around the passed parent {@link Publisher}
|
||||
* @param source the {@link Publisher} to decorate
|
||||
*/
|
||||
protected CustomMonoWithoutRunStyleOperator(Mono<? extends O> source) {
|
||||
CustomMonoWithoutRunStyleOperator(Mono<? extends O> source) {
|
||||
super(source);
|
||||
}
|
||||
|
||||
@@ -329,37 +156,29 @@ class ReactorHooksHelperTests {
|
||||
|
||||
}
|
||||
|
||||
static class PlusOneSubscriber extends BaseSubscriber<Integer> {
|
||||
static class TraceContextPropagatorOperator<O> extends MonoOperator<O, O> implements TraceContextPropagator {
|
||||
|
||||
CoreSubscriber<? super Integer> actual;
|
||||
|
||||
PlusOneSubscriber(CoreSubscriber<? super Integer> actual) {
|
||||
this.actual = actual;
|
||||
/**
|
||||
* Build a {@link MonoOperator} wrapper around the passed parent {@link Publisher}
|
||||
* @param source the {@link Publisher} to decorate
|
||||
*/
|
||||
TraceContextPropagatorOperator(Mono<? extends O> source) {
|
||||
super(source);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Context currentContext() {
|
||||
return actual.currentContext();
|
||||
public void subscribe(CoreSubscriber<? super O> actual) {
|
||||
source.subscribe(actual);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
@Override
|
||||
protected void hookOnSubscribe(Subscription subscription) {
|
||||
actual.onSubscribe(subscription);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void hookOnNext(Integer value) {
|
||||
actual.onNext(value + 1);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void hookOnComplete() {
|
||||
actual.onComplete();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void hookOnError(Throwable throwable) {
|
||||
actual.onError(throwable);
|
||||
public Object scanUnsafe(Attr key) {
|
||||
// just to check that it pass by instanceof
|
||||
if (Attr.RUN_STYLE == key) {
|
||||
return null;
|
||||
}
|
||||
return super.scanUnsafe(key);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -57,6 +57,11 @@ public abstract class FlowsScopePassingSpanSubscriberTests {
|
||||
|
||||
static final String HOOK_KEY = "org.springframework.cloud.sleuth.autoconfig.instrument.reactor.TraceReactorAutoConfiguration.TraceReactorConfiguration";
|
||||
|
||||
// Cannot access
|
||||
// org.springframework.cloud.sleuth.instrument.reactor.ReactorHooksHelper.LIFTER_NAME
|
||||
// directly from this test so copy its value
|
||||
static final String LIFTER_NAME = "org.springframework.cloud.sleuth.instrument.reactor.ReactorHooksHelper.ScopePassingLifter";
|
||||
|
||||
static {
|
||||
// AssertJ will recognise QueueSubscription implements queue and try to invoke
|
||||
// iterator. That's not allowed, and will cause an exception
|
||||
@@ -171,9 +176,9 @@ public abstract class FlowsScopePassingSpanSubscriberTests {
|
||||
Hooks.onLastOperator("test", p -> {
|
||||
// check only first onLast Hook
|
||||
if (once.compareAndSet(false, true)) {
|
||||
assertThat(p).isInstanceOf(TraceContextPropagator.class);
|
||||
assertThat(Scannable.from(p).scanUnsafe(Scannable.Attr.LIFTER)).isEqualTo(LIFTER_NAME);
|
||||
Object parent = Scannable.from(p).scanUnsafe(Scannable.Attr.PARENT);
|
||||
assertThat(parent).isNotInstanceOf(TraceContextPropagator.class);
|
||||
assertThat(Scannable.from(parent).scanUnsafe(Scannable.Attr.LIFTER)).isNotEqualTo(LIFTER_NAME);
|
||||
}
|
||||
return p;
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user