@Test public void testSpanIsFinishedWhenMonoHasCompleted() throws InterruptedException { Span span = tracer.buildSpan("a span").start(); AtomicReference<SignalType> finalSignalType = new AtomicReference<>(); CountDownLatch finalSignalCountDownLatch = new CountDownLatch(1); Mono<Void> source = Mono.just("5") .then() .doFinally(signalType -> { finalSignalType.set(signalType); finalSignalCountDownLatch.countDown(); }) .subscribeOn(Schedulers.single()); try (Scope scope = tracer.activateSpan(span)) { new TracingOperator(source, exchange, tracer, Arrays.asList(spanDecorator)) .subscribe(); finalSignalCountDownLatch.await(); } assertEquals(1, tracer.finishedSpans().size()); assertEquals(SignalType.ON_COMPLETE, finalSignalType.get()); Span finishedSpan = tracer.finishedSpans().get(0); verify(spanDecorator).onRequest(exchange, finishedSpan); verify(spanDecorator).onResponse(exchange, finishedSpan); verify(spanDecorator, never()).onError(any(ServerWebExchange.class), any(Throwable.class), any(Span.class)); }
@Override public void onError(final Throwable throwable) { spanDecorators.forEach(spanDecorator -> safelyCall(() -> spanDecorator.onError(exchange, throwable, span))); span.finish(); exchange.getAttributes().remove(TracingWebFilter.SERVER_SPAN_CONTEXT); subscriber.onError(throwable); }
TracingSubscriber implements CoreSubscriber<Void> { @Override public void onError(final Throwable throwable) { spanDecorators.forEach(spanDecorator -> safelyCall(() -> spanDecorator.onError(exchange, throwable, span))); span.finish(); exchange.getAttributes().remove(TracingWebFilter.SERVER_SPAN_CONTEXT); subscriber.onError(throwable); } }
TracingSubscriber implements CoreSubscriber<Void> { @Override public void onError(final Throwable throwable) { spanDecorators.forEach(spanDecorator -> safelyCall(() -> spanDecorator.onError(exchange, throwable, span))); span.finish(); exchange.getAttributes().remove(TracingWebFilter.SERVER_SPAN_CONTEXT); subscriber.onError(throwable); } TracingSubscriber(
            final CoreSubscriber<? super Void> subscriber,
            final ServerWebExchange exchange,
            final Context context,
            final Span span,
            final List<WebFluxSpanDecorator> spanDecorators
    ); }
TracingSubscriber implements CoreSubscriber<Void> { @Override public void onError(final Throwable throwable) { spanDecorators.forEach(spanDecorator -> safelyCall(() -> spanDecorator.onError(exchange, throwable, span))); span.finish(); exchange.getAttributes().remove(TracingWebFilter.SERVER_SPAN_CONTEXT); subscriber.onError(throwable); } TracingSubscriber(
            final CoreSubscriber<? super Void> subscriber,
            final ServerWebExchange exchange,
            final Context context,
            final Span span,
            final List<WebFluxSpanDecorator> spanDecorators
    ); @Override void onSubscribe(final Subscription subscription); @Override void onNext(final Void aVoid); @Override void onError(final Throwable throwable); @Override void onComplete(); @Override Context currentContext(); }
TracingSubscriber implements CoreSubscriber<Void> { @Override public void onError(final Throwable throwable) { spanDecorators.forEach(spanDecorator -> safelyCall(() -> spanDecorator.onError(exchange, throwable, span))); span.finish(); exchange.getAttributes().remove(TracingWebFilter.SERVER_SPAN_CONTEXT); subscriber.onError(throwable); } TracingSubscriber(
            final CoreSubscriber<? super Void> subscriber,
            final ServerWebExchange exchange,
            final Context context,
            final Span span,
            final List<WebFluxSpanDecorator> spanDecorators
    ); @Override void onSubscribe(final Subscription subscription); @Override void onNext(final Void aVoid); @Override void onError(final Throwable throwable); @Override void onComplete(); @Override Context currentContext();  }