SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Reverting the change

Browse Source
This commit is contained in:
Marcin Grzejszczak
2017-07-20 11:58:10 +02:00
parent cee2de9b81
commit 2ebba99c4e
2 changed files with 12 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -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<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(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 {