diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/DefaultSpanCreator.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/DefaultSpanCreator.java index e6cff02ec..e2dd913ba 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/DefaultSpanCreator.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/DefaultSpanCreator.java @@ -35,7 +35,7 @@ class DefaultNewSpanParser implements NewSpanParser { @Override public void parse(MethodInvocation pjp, NewSpan newSpan, SpanCustomizer span) { - String name = StringUtils.isEmpty(newSpan.name()) ? + String name = newSpan == null || StringUtils.isEmpty(newSpan.name()) ? pjp.getMethod().getName() : newSpan.name(); String changedName = SpanNameUtil.toLowerHyphen(name); if (log.isDebugEnabled()) { diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java index 6c58c19a4..a69269bad 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/annotation/SleuthAdvisorConfig.java @@ -16,17 +16,13 @@ package org.springframework.cloud.sleuth.annotation; -import java.lang.annotation.Annotation; -import java.lang.reflect.Method; -import java.util.concurrent.atomic.AtomicBoolean; -import javax.annotation.PostConstruct; - import brave.Span; import brave.Tracer; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.reactivestreams.Publisher; import org.springframework.aop.ClassFilter; import org.springframework.aop.IntroductionInterceptor; import org.springframework.aop.Pointcut; @@ -40,6 +36,15 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import javax.annotation.PostConstruct; +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.function.Supplier; /** * Custom pointcut advisor that picks all classes / interfaces that @@ -143,19 +148,14 @@ class SleuthAdvisorConfig extends AbstractPointcutAdvisor implements BeanFactory public boolean hasAnnotatedMethods(Class clazz) { final AtomicBoolean found = new AtomicBoolean(false); - ReflectionUtils.doWithMethods(clazz, - new ReflectionUtils.MethodCallback() { - @Override - public void doWith(Method method) throws IllegalArgumentException, - IllegalAccessException { - if (found.get()) { - return; - } - Annotation annotation = AnnotationUtils.findAnnotation(method, - SleuthAdvisorConfig.AnnotationMethodsResolver.this.annotationType); - if (annotation != null) { found.set(true); } - } - }); + ReflectionUtils.doWithMethods(clazz, method -> { + if (found.get()) { + return; + } + Annotation annotation = AnnotationUtils.findAnnotation(method, + AnnotationMethodsResolver.this.annotationType); + if (annotation != null) { found.set(true); } + }); return found.get(); } @@ -183,6 +183,7 @@ class SleuthInterceptor implements IntroductionInterceptor, BeanFactoryAware { if (method == null) { return invocation.proceed(); } + Method mostSpecificMethod = AopUtils .getMostSpecificMethod(method, invocation.getThis().getClass()); NewSpan newSpan = SleuthAnnotationUtils.findAnnotation(mostSpecificMethod, NewSpan.class); @@ -190,39 +191,135 @@ class SleuthInterceptor implements IntroductionInterceptor, BeanFactoryAware { if (newSpan == null && continueSpan == null) { return invocation.proceed(); } + + if(isReactorReturnType(method.getReturnType())){ + return proceedUnderReactorSpan(invocation, newSpan, continueSpan); + } else { + return proceedUnderSynchronousSpan(invocation, newSpan, continueSpan); + } + } + + private boolean isReactorReturnType(Class returnType) { + return Flux.class.equals(returnType) || Mono.class.equals(returnType); + } + + private Object proceedUnderSynchronousSpan( + MethodInvocation invocation, NewSpan newSpan, ContinueSpan continueSpan) throws Throwable { Span span = tracer().currentSpan(); - if (newSpan != null || span == null) { - span = tracer().nextSpan().start(); + //in case of @ContinueSpan and no span in tracer we start new span and should close it on completion + boolean startNewSpan = newSpan != null || span == null; + if (startNewSpan) { + span = tracer().nextSpan(); newSpanParser().parse(invocation, newSpan, span); + span.start(); } String log = log(continueSpan); boolean hasLog = StringUtils.hasText(log); try (Tracer.SpanInScope ws = tracer().withSpanInScope(span)) { - if (hasLog) { - logEvent(span, log + ".before"); - } - spanTagAnnotationHandler().addAnnotatedParameters(invocation); - addTags(invocation, span); + before(invocation, span, log, hasLog); return invocation.proceed(); } catch (Exception e) { - if (logger.isDebugEnabled()) { - logger.debug("Exception occurred while trying to continue the pointcut", e); - } - if (hasLog) { - logEvent(span, log + ".afterFailure"); - } - span.error(e); + onFailure(span, log, hasLog, e); throw e; } finally { - if (hasLog) { - logEvent(span, log + ".after"); + after(span, startNewSpan, log, hasLog); + } + } + + private Object proceedUnderReactorSpan( + MethodInvocation invocation, NewSpan newSpan, ContinueSpan continueSpan) throws Throwable{ + Span spanPrevious = tracer().currentSpan(); + //in case of @ContinueSpan and no span in tracer we start new span and should close it on completion + boolean startNewSpan = newSpan != null || spanPrevious == null; + Span span; + if (startNewSpan) { + span = tracer().nextSpan(); + newSpanParser().parse(invocation, newSpan, span); + } else { + span = spanPrevious; + } + + String log = log(continueSpan); + boolean hasLog = StringUtils.hasText(log); + + try(Tracer.SpanInScope ws = tracer().withSpanInScope(span)) { + + Publisher publisher = (Publisher) invocation.proceed(); + + Mono startSpan = Mono.defer(() -> withSpanInScope(span, () -> { + if (startNewSpan) { + span.start(); + } + + before(invocation, span, log, hasLog); + return Mono.just(span); + })); + + if(publisher instanceof Mono){ + return startSpan.flatMap(spanStarted -> ((Mono)publisher) + .doOnError(onFailureReactor(log, hasLog, spanStarted)) + .doOnTerminate(afterReactor(startNewSpan, log, hasLog, spanStarted))); } - if (newSpan != null) { - span.finish(); + else if(publisher instanceof Flux){ + return startSpan.flatMapMany(spanStarted -> ((Flux)publisher) + .doOnError(onFailureReactor(log, hasLog, spanStarted)) + .doOnTerminate(afterReactor(startNewSpan, log, hasLog, spanStarted))); + } + else { + throw new IllegalArgumentException("Unexpected type of publisher: "+publisher.getClass()); } } } + private T withSpanInScope(Span span, Supplier supplier) { + try(Tracer.SpanInScope ws1 = tracer().withSpanInScope(span)) { + return supplier.get(); + } + } + + private Runnable afterReactor(boolean isNewSpan, String log, boolean hasLog, Span span) { + return () -> { + try(Tracer.SpanInScope ws = tracer().withSpanInScope(span)) { + after(span, isNewSpan, log, hasLog); + } + }; + } + + private Consumer onFailureReactor(String log, boolean hasLog, Span span) { + return throwable -> { + try(Tracer.SpanInScope ws = tracer().withSpanInScope(span)) { + onFailure(span, log, hasLog, throwable); + } + }; + } + + private void before(MethodInvocation invocation, Span span, String log, boolean hasLog) { + if (hasLog) { + logEvent(span, log + ".before"); + } + spanTagAnnotationHandler().addAnnotatedParameters(invocation); + addTags(invocation, span); + } + + private void after(Span span, boolean isNewSpan, String log, boolean hasLog) { + if (hasLog) { + logEvent(span, log + ".after"); + } + if (isNewSpan) { + span.finish(); + } + } + + private void onFailure(Span span, String log, boolean hasLog, Throwable e) { + if (logger.isDebugEnabled()) { + logger.debug("Exception occurred while trying to continue the pointcut", e); + } + if (hasLog) { + logEvent(span, log + ".afterFailure"); + } + span.error(e); + } + private void addTags(MethodInvocation invocation, Span span) { span.tag(CLASS_KEY, invocation.getThis().getClass().getSimpleName()); span.tag(METHOD_KEY, invocation.getMethod().getName()); diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectFluxTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectFluxTests.java new file mode 100644 index 000000000..7231e91a0 --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectFluxTests.java @@ -0,0 +1,505 @@ +/* + * Copyright 2013-2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.annotation; + +import brave.Span; +import brave.Tracer; +import brave.sampler.Sampler; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.sleuth.util.ArrayListSpanReporter; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import zipkin2.Annotation; +import zipkin2.reporter.Reporter; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +import static org.assertj.core.api.BDDAssertions.then; +import static org.springframework.cloud.sleuth.annotation.SleuthSpanCreatorAspectFluxTests.TestBean.TEST_STRING1; +import static org.springframework.cloud.sleuth.annotation.SleuthSpanCreatorAspectFluxTests.TestBean.TEST_STRING2; + +@SpringBootTest(classes = SleuthSpanCreatorAspectFluxTests.TestConfiguration.class) +@RunWith(SpringJUnit4ClassRunner.class) +public class SleuthSpanCreatorAspectFluxTests { + + @Autowired TestBeanInterface testBean; + @Autowired Tracer tracer; + @Autowired ArrayListSpanReporter reporter; + + @Before + public void setup() { + this.reporter.clear(); + testBean.reset(); + } + + @Test + public void shouldCreateSpanWhenAnnotationOnInterfaceMethod() { + Flux flux = this.testBean.testMethod(); + + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWhenAnnotationOnClassMethod() { + Flux flux = this.testBean.testMethod2(); + + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method2"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithCustomNameWhenAnnotationOnClassMethod() { + Flux flux = this.testBean.testMethod3(); + + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method3"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithCustomNameWhenAnnotationOnInterfaceMethod() { + Flux flux = this.testBean.testMethod4(); + + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method4"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithTagWhenAnnotationOnInterfaceMethod() { + // tag::execution[] + Flux flux = this.testBean.testMethod5("test"); + + // end::execution[] + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method5"); + then(spans.get(0).tags()).containsEntry("testTag", "test"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithTagWhenAnnotationOnClassMethod() { + Flux flux = this.testBean.testMethod6("test"); + + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method6"); + then(spans.get(0).tags()).containsEntry("testTag6", "test"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithLogWhenAnnotationOnInterfaceMethod() { + Flux flux = this.testBean.testMethod8("test"); + + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method8"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithLogWhenAnnotationOnClassMethod() { + Flux flux = this.testBean.testMethod9("test"); + + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method9"); + then(spans.get(0).tags()) + .containsEntry("class", "TestBean") + .containsEntry("method", "testMethod9"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldContinueSpanWithLogWhenAnnotationOnInterfaceMethod() { + Span span = this.tracer.nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.start())) { + Flux flux = this.testBean.testMethod10("test"); + + verifyNoSpansUntilFluxComplete(flux); + } finally { + span.finish(); + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("foo"); + then(spans.get(0).tags()) + .containsEntry("customTestTag10", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldStartAndCloseSpanOnContinueSpanIfSpanNotSet() { + Flux flux = this.testBean.testMethod10("test"); + verifyNoSpansUntilFluxComplete(flux); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method10"); + then(spans.get(0).tags()) + .containsEntry("customTestTag10", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldContinueSpanWhenKeyIsUsedOnSpanTagWhenAnnotationOnInterfaceMethod() { + Span span = this.tracer.nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.start())) { + Flux flux = this.testBean.testMethod10_v2("test"); + + verifyNoSpansUntilFluxComplete(flux); + } finally { + span.finish(); + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("foo"); + then(spans.get(0).tags()) + .containsEntry("customTestTag10", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldContinueSpanWithLogWhenAnnotationOnClassMethod() { + Span span = this.tracer.nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.start())) { + // tag::continue_span_execution[] + Flux flux = this.testBean.testMethod11("test"); + // end::continue_span_execution[] + verifyNoSpansUntilFluxComplete(flux); + } finally { + span.finish(); + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("foo"); + then(spans.get(0).tags()) + .containsEntry("class", "TestBean") + .containsEntry("method", "testMethod11") + .containsEntry("customTestTag11", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldAddErrorTagWhenExceptionOccurredInNewSpan() { + try { + Flux flux = this.testBean.testMethod12("test"); + + then(this.reporter.getSpans()).isEmpty(); + + flux.toIterable().iterator().next(); + } catch (RuntimeException ignored) { + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method12"); + then(spans.get(0).tags()) + .containsEntry("testTag12", "test") + .containsEntry("error", "test exception 12"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldAddErrorTagWhenExceptionOccurredInContinueSpan() { + Span span = this.tracer.nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.start())) { + // tag::continue_span_execution[] + Flux flux = this.testBean.testMethod13(); + + then(this.reporter.getSpans()).isEmpty(); + + flux.toIterable().iterator().next(); + // end::continue_span_execution[] + } catch (RuntimeException ignored) { + } finally { + span.finish(); + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("foo"); + then(spans.get(0).tags()) + .containsEntry("error", "test exception 13"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("testMethod13.before", "testMethod13.afterFailure", + "testMethod13.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldNotCreateSpanWhenNotAnnotated() { + Flux flux = this.testBean.testMethod7(); + verifyNoSpansUntilFluxComplete(flux); + + List spans = new ArrayList<>(this.reporter.getSpans()); + then(spans).isEmpty(); + } + + private void verifyNoSpansUntilFluxComplete(Flux flux) { + Iterator iterator = flux.toIterable().iterator(); + + then(this.reporter.getSpans()).isEmpty(); + testBean.proceed(); + + String result1 = iterator.next(); + then(result1).isEqualTo(TEST_STRING1); + then(this.reporter.getSpans()).isEmpty(); + + testBean.proceed(); + String result2 = iterator.next(); + + then(result2).isEqualTo(TEST_STRING2); + } + + protected interface TestBeanInterface { + + // tag::annotated_method[] + @NewSpan + Flux testMethod(); + // end::annotated_method[] + + Flux testMethod2(); + + @NewSpan(name = "interfaceCustomNameOnTestMethod3") + Flux testMethod3(); + + // tag::custom_name_on_annotated_method[] + @NewSpan("customNameOnTestMethod4") + Flux testMethod4(); + // end::custom_name_on_annotated_method[] + + // tag::custom_name_and_tag_on_annotated_method[] + @NewSpan(name = "customNameOnTestMethod5") + Flux testMethod5(@SpanTag("testTag") String param); + // end::custom_name_and_tag_on_annotated_method[] + + Flux testMethod6(String test); + + Flux testMethod7(); + + @NewSpan(name = "customNameOnTestMethod8") + Flux testMethod8(String param); + + @NewSpan(name = "testMethod9") + Flux testMethod9(String param); + + @ContinueSpan(log = "customTest") + Flux testMethod10(@SpanTag(value = "testTag10") String param); + + @ContinueSpan(log = "customTest") + Flux testMethod10_v2(@SpanTag(key = "testTag10") String param); + + // tag::continue_span[] + @ContinueSpan(log = "testMethod11") + Flux testMethod11(@SpanTag("testTag11") String param); + // end::continue_span[] + + @NewSpan + Flux testMethod12(@SpanTag("testTag12") String param); + + @ContinueSpan(log = "testMethod13") + Flux testMethod13(); + + @ContinueSpan + Flux testMethod14(String param); + + void proceed(); + + void reset(); + } + + protected static class TestBean implements TestBeanInterface { + + public static final String TEST_STRING1 = "Test String 1"; + public static final String TEST_STRING2 = "Test String 2"; + + private AtomicReference> proceed + = new AtomicReference<>(new CompletableFuture<>()); + private Flux testFlux = Flux.defer(() -> Flux.just(TEST_STRING1, TEST_STRING2)) + .delayUntil(s -> Mono.fromFuture(proceed.get())) + .doOnNext(s -> proceed.set(new CompletableFuture<>())); + + @Override + public void reset(){ + proceed.set(new CompletableFuture<>()); + } + + public void proceed(){ + proceed.get().complete(null); + } + + @Override + public Flux testMethod() { + return testFlux; + } + + @NewSpan + @Override + public Flux testMethod2() { + return testFlux; + } + + // tag::name_on_implementation[] + @NewSpan(name = "customNameOnTestMethod3") + @Override + public Flux testMethod3() { + return testFlux; + } + // end::name_on_implementation[] + + @Override + public Flux testMethod4() { + return testFlux; + } + + @Override + public Flux testMethod5(String test) { + return testFlux; + } + + @NewSpan(name = "customNameOnTestMethod6") + @Override + public Flux testMethod6(@SpanTag("testTag6") String test) { + return testFlux; + } + + @Override + public Flux testMethod7() { + return testFlux; + } + + @Override + public Flux testMethod8(String param) { + return testFlux; + } + + @NewSpan(name = "customNameOnTestMethod9") + @Override + public Flux testMethod9(String param) { + return testFlux; + } + + @Override + public Flux testMethod10(@SpanTag(value = "customTestTag10") String param) { + return testFlux; + } + + @Override + public Flux testMethod10_v2(@SpanTag(key = "customTestTag10") String param) { + return testFlux; + } + + @ContinueSpan(log = "customTest") + @Override + public Flux testMethod11(@SpanTag("customTestTag11") String param) { + return testFlux; + } + + @Override + public Flux testMethod12(String param) { + return Flux.defer(() -> Flux.error(new RuntimeException("test exception 12"))); + } + + @Override + public Flux testMethod13() { + return Flux.defer(() -> Flux.error(new RuntimeException("test exception 13"))); + } + + @Override + public Flux testMethod14(String param) { + return Flux.just(TEST_STRING1, TEST_STRING2); + } + } + + @Configuration + @EnableAutoConfiguration + protected static class TestConfiguration { + + @Bean + public TestBeanInterface testBean() { + return new TestBean(); + } + + @Bean Reporter spanReporter() { + return new ArrayListSpanReporter(); + } + + @Bean Sampler alwaysSampler() { + return Sampler.ALWAYS_SAMPLE; + } + } +} diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectMonoTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectMonoTests.java new file mode 100644 index 000000000..530a8554b --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectMonoTests.java @@ -0,0 +1,480 @@ +/* + * Copyright 2013-2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.annotation; + +import brave.Span; +import brave.Tracer; +import brave.sampler.Sampler; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.sleuth.util.ArrayListSpanReporter; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import reactor.core.publisher.Mono; +import zipkin2.Annotation; +import zipkin2.reporter.Reporter; + +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; + +import static org.assertj.core.api.BDDAssertions.then; +import static org.springframework.cloud.sleuth.annotation.SleuthSpanCreatorAspectMonoTests.TestBean.TEST_STRING; +import static reactor.core.publisher.Mono.just; + +@SpringBootTest(classes = SleuthSpanCreatorAspectMonoTests.TestConfiguration.class) +@RunWith(SpringJUnit4ClassRunner.class) +public class SleuthSpanCreatorAspectMonoTests { + + @Autowired TestBeanInterface testBean; + @Autowired Tracer tracer; + @Autowired ArrayListSpanReporter reporter; + + @Before + public void setup() { + this.reporter.clear(); + } + + @Test + public void shouldCreateSpanWhenAnnotationOnInterfaceMethod() { + Mono mono = this.testBean.testMethod(); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method"); + then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldCreateSpanWhenAnnotationOnClassMethod() { + Mono mono = this.testBean.testMethod2(); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method2"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithCustomNameWhenAnnotationOnClassMethod() { + Mono mono = this.testBean.testMethod3(); + + then(this.reporter.getSpans()).isEmpty(); + + String result = mono.block(); + + then(result).isEqualTo(TEST_STRING); + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method3"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithCustomNameWhenAnnotationOnInterfaceMethod() { + Mono mono = this.testBean.testMethod4(); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method4"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithTagWhenAnnotationOnInterfaceMethod() { + // tag::execution[] + Mono mono = this.testBean.testMethod5("test"); + + // end::execution[] + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method5"); + then(spans.get(0).tags()).containsEntry("testTag", "test"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithTagWhenAnnotationOnClassMethod() { + Mono mono = this.testBean.testMethod6("test"); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method6"); + then(spans.get(0).tags()).containsEntry("testTag6", "test"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithLogWhenAnnotationOnInterfaceMethod() { + Mono mono = this.testBean.testMethod8("test"); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method8"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldCreateSpanWithLogWhenAnnotationOnClassMethod() { + Mono mono = this.testBean.testMethod9("test"); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("custom-name-on-test-method9"); + then(spans.get(0).tags()) + .containsEntry("class", "TestBean") + .containsEntry("method", "testMethod9"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldContinueSpanWithLogWhenAnnotationOnInterfaceMethod() { + Span span = this.tracer.nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.start())) { + Mono mono = this.testBean.testMethod10("test"); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + } finally { + span.finish(); + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("foo"); + then(spans.get(0).tags()) + .containsEntry("customTestTag10", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldStartAndCloseSpanOnContinueSpanIfSpanNotSet() { + this.testBean.testMethod10("test").block(); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method10"); + then(spans.get(0).tags()) + .containsEntry("customTestTag10", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldContinueSpanWhenKeyIsUsedOnSpanTagWhenAnnotationOnInterfaceMethod() { + Span span = this.tracer.nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.start())) { + Mono mono = this.testBean.testMethod10_v2("test"); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + } finally { + span.finish(); + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("foo"); + then(spans.get(0).tags()) + .containsEntry("customTestTag10", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldContinueSpanWithLogWhenAnnotationOnClassMethod() { + Span span = this.tracer.nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.start())) { + // tag::continue_span_execution[] + Mono mono = this.testBean.testMethod11("test"); + // end::continue_span_execution[] + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + } finally { + span.finish(); + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("foo"); + then(spans.get(0).tags()) + .containsEntry("class", "TestBean") + .containsEntry("method", "testMethod11") + .containsEntry("customTestTag11", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldAddErrorTagWhenExceptionOccurredInNewSpan() { + try { + Mono mono = this.testBean.testMethod12("test"); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + } catch (RuntimeException ignored) { + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method12"); + then(spans.get(0).tags()) + .containsEntry("testTag12", "test") + .containsEntry("error", "test exception 12"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldAddErrorTagWhenExceptionOccurredInContinueSpan() { + Span span = this.tracer.nextSpan().name("foo"); + + try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.start())) { + // tag::continue_span_execution[] + Mono mono = this.testBean.testMethod13(); + + then(this.reporter.getSpans()).isEmpty(); + + mono.block(); + // end::continue_span_execution[] + } catch (RuntimeException ignored) { + } finally { + span.finish(); + } + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("foo"); + then(spans.get(0).tags()) + .containsEntry("error", "test exception 13"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("testMethod13.before", "testMethod13.afterFailure", + "testMethod13.after"); + then(spans.get(0).duration()).isNotZero(); + } + + @Test + public void shouldNotCreateSpanWhenNotAnnotated() { + Mono mono = this.testBean.testMethod7(); + mono.block(); + + List spans = new ArrayList<>(this.reporter.getSpans()); + then(spans).isEmpty(); + } + + protected interface TestBeanInterface { + + // tag::annotated_method[] + @NewSpan + Mono testMethod(); + // end::annotated_method[] + + Mono testMethod2(); + + @NewSpan(name = "interfaceCustomNameOnTestMethod3") + Mono testMethod3(); + + // tag::custom_name_on_annotated_method[] + @NewSpan("customNameOnTestMethod4") + Mono testMethod4(); + // end::custom_name_on_annotated_method[] + + // tag::custom_name_and_tag_on_annotated_method[] + @NewSpan(name = "customNameOnTestMethod5") + Mono testMethod5(@SpanTag("testTag") String param); + // end::custom_name_and_tag_on_annotated_method[] + + Mono testMethod6(String test); + + Mono testMethod7(); + + @NewSpan(name = "customNameOnTestMethod8") + Mono testMethod8(String param); + + @NewSpan(name = "testMethod9") + Mono testMethod9(String param); + + @ContinueSpan(log = "customTest") + Mono testMethod10(@SpanTag(value = "testTag10") String param); + + @ContinueSpan(log = "customTest") + Mono testMethod10_v2(@SpanTag(key = "testTag10") String param); + + // tag::continue_span[] + @ContinueSpan(log = "testMethod11") + Mono testMethod11(@SpanTag("testTag11") String param); + // end::continue_span[] + + @NewSpan + Mono testMethod12(@SpanTag("testTag12") String param); + + @ContinueSpan(log = "testMethod13") + Mono testMethod13(); + } + + protected static class TestBean implements TestBeanInterface { + + public static final String TEST_STRING = "Test String"; + public static final Mono TEST_MONO = Mono.defer(() -> just(TEST_STRING)); + + @Override + public Mono testMethod() { + return TEST_MONO; + } + + @NewSpan + @Override + public Mono testMethod2() { + return TEST_MONO; + } + + // tag::name_on_implementation[] + @NewSpan(name = "customNameOnTestMethod3") + @Override + public Mono testMethod3() { + return TEST_MONO; + } + // end::name_on_implementation[] + + @Override + public Mono testMethod4() { + return TEST_MONO; + } + + @Override + public Mono testMethod5(String test) { + return TEST_MONO; + } + + @NewSpan(name = "customNameOnTestMethod6") + @Override + public Mono testMethod6(@SpanTag("testTag6") String test) { + return TEST_MONO; + } + + @Override + public Mono testMethod7() { + return TEST_MONO; + } + + @Override + public Mono testMethod8(String param) { + return TEST_MONO; + } + + @NewSpan(name = "customNameOnTestMethod9") + @Override + public Mono testMethod9(String param) { + return TEST_MONO; + } + + @Override + public Mono testMethod10(@SpanTag(value = "customTestTag10") String param) { + return TEST_MONO; + } + + @Override + public Mono testMethod10_v2(@SpanTag(key = "customTestTag10") String param) { + return TEST_MONO; + } + + @ContinueSpan(log = "customTest") + @Override + public Mono testMethod11(@SpanTag("customTestTag11") String param) { + return TEST_MONO; + } + + @Override + public Mono testMethod12(String param) { + return Mono.defer(() -> Mono.error(new RuntimeException("test exception 12"))); + } + + @Override + public Mono testMethod13() { + return Mono.defer(() -> Mono.error(new RuntimeException("test exception 13"))); + } + } + + @Configuration + @EnableAutoConfiguration + protected static class TestConfiguration { + + @Bean + public TestBeanInterface testBean() { + return new TestBean(); + } + + @Bean Reporter spanReporter() { + return new ArrayListSpanReporter(); + } + + @Bean Sampler alwaysSampler() { + return Sampler.ALWAYS_SAMPLE; + } + } +} diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectTests.java index 2bf256e2e..991880bcd 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/annotation/SleuthSpanCreatorAspectTests.java @@ -60,6 +60,7 @@ public class SleuthSpanCreatorAspectTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("test-method"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -70,6 +71,7 @@ public class SleuthSpanCreatorAspectTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("test-method2"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -80,6 +82,7 @@ public class SleuthSpanCreatorAspectTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method3"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -90,6 +93,7 @@ public class SleuthSpanCreatorAspectTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method4"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -103,6 +107,7 @@ public class SleuthSpanCreatorAspectTests { then(spans.get(0).name()).isEqualTo("custom-name-on-test-method5"); then(spans.get(0).tags()).containsEntry("testTag", "test"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -114,6 +119,7 @@ public class SleuthSpanCreatorAspectTests { then(spans.get(0).name()).isEqualTo("custom-name-on-test-method6"); then(spans.get(0).tags()).containsEntry("testTag6", "test"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -124,6 +130,7 @@ public class SleuthSpanCreatorAspectTests { then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("custom-name-on-test-method8"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -137,6 +144,7 @@ public class SleuthSpanCreatorAspectTests { .containsEntry("class", "TestBean") .containsEntry("method", "testMethod9"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -149,7 +157,7 @@ public class SleuthSpanCreatorAspectTests { span.finish(); } - List spans = new ArrayList<>(this.reporter.getSpans()); + List spans = this.reporter.getSpans(); then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("foo"); then(spans.get(0).tags()) @@ -158,6 +166,23 @@ public class SleuthSpanCreatorAspectTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); + } + + @Test + public void shouldStartAndCloseSpanOnContinueSpanIfSpanNotSet() { + this.testBean.testMethod10("test"); + + List spans = this.reporter.getSpans(); + then(spans).hasSize(1); + then(spans.get(0).name()).isEqualTo("test-method10"); + then(spans.get(0).tags()) + .containsEntry("customTestTag10", "test"); + then(spans.get(0).annotations() + .stream().map(Annotation::value).collect(Collectors.toList())) + .contains("customTest.before", "customTest.after"); + then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -170,7 +195,7 @@ public class SleuthSpanCreatorAspectTests { span.finish(); } - List spans = new ArrayList<>(this.reporter.getSpans()); + List spans = this.reporter.getSpans(); then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("foo"); then(spans.get(0).tags()) @@ -179,6 +204,7 @@ public class SleuthSpanCreatorAspectTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -193,7 +219,7 @@ public class SleuthSpanCreatorAspectTests { span.finish(); } - List spans = new ArrayList<>(this.reporter.getSpans()); + List spans = this.reporter.getSpans(); then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("foo"); then(spans.get(0).tags()) @@ -204,6 +230,7 @@ public class SleuthSpanCreatorAspectTests { .stream().map(Annotation::value).collect(Collectors.toList())) .contains("customTest.before", "customTest.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -213,13 +240,14 @@ public class SleuthSpanCreatorAspectTests { } catch (RuntimeException ignored) { } - List spans = new ArrayList<>(this.reporter.getSpans()); + List spans = this.reporter.getSpans(); then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("test-method12"); then(spans.get(0).tags()) .containsEntry("testTag12", "test") .containsEntry("error", "test exception 12"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test @@ -235,7 +263,7 @@ public class SleuthSpanCreatorAspectTests { span.finish(); } - List spans = new ArrayList<>(this.reporter.getSpans()); + List spans = this.reporter.getSpans(); then(spans).hasSize(1); then(spans.get(0).name()).isEqualTo("foo"); then(spans.get(0).tags()) @@ -245,14 +273,16 @@ public class SleuthSpanCreatorAspectTests { .contains("testMethod13.before", "testMethod13.afterFailure", "testMethod13.after"); then(spans.get(0).duration()).isNotZero(); + then(this.tracer.currentSpan()).isNull(); } @Test public void shouldNotCreateSpanWhenNotAnnotated() { this.testBean.testMethod7(); - List spans = new ArrayList<>(this.reporter.getSpans()); + List spans = this.reporter.getSpans(); then(spans).isEmpty(); + then(this.tracer.currentSpan()).isNull(); } protected interface TestBeanInterface { diff --git a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-feign/pom.xml b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-feign/pom.xml index 4dc515e57..481dfa076 100644 --- a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-feign/pom.xml +++ b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-feign/pom.xml @@ -90,6 +90,16 @@ spring-boot-starter-test test + + io.projectreactor + reactor-core + test + + + org.reactivestreams + reactive-streams + test + diff --git a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-test-core/pom.xml b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-test-core/pom.xml index b53221fc7..91d952916 100644 --- a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-test-core/pom.xml +++ b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample-test-core/pom.xml @@ -109,6 +109,14 @@ io.zipkin.zipkin2 zipkin + + io.projectreactor + reactor-core + + + org.reactivestreams + reactive-streams + diff --git a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample/pom.xml b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample/pom.xml index ea89767e0..d6efe3e81 100644 --- a/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample/pom.xml +++ b/spring-cloud-sleuth-samples/spring-cloud-sleuth-sample/pom.xml @@ -77,6 +77,16 @@ spring-boot-starter-test test + + io.projectreactor + reactor-core + test + + + org.reactivestreams + reactive-streams + test + diff --git a/spring-cloud-sleuth-zipkin/pom.xml b/spring-cloud-sleuth-zipkin/pom.xml index b34430f07..2d4729b05 100644 --- a/spring-cloud-sleuth-zipkin/pom.xml +++ b/spring-cloud-sleuth-zipkin/pom.xml @@ -156,6 +156,16 @@ aspectjweaver test + + io.projectreactor + reactor-core + test + + + org.reactivestreams + reactive-streams + test +