update test

Author: jaydeluca

instrumentation/rxjava/rxjava-2.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava/v2_0/TracingAssembly.java

@@ -67,7 +67,6 @@ public final class TracingAssembly {
   @Nullable
   static volatile Function<? super Observable, ? extends Observable> oldOnObservableAssembly;
 
-
   @SuppressWarnings("rawtypes")
   @GuardedBy("TracingAssembly.class")
   @Nullable

instrumentation/rxjava/rxjava-2.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava/v2_0/TracingObservable.java

@@ -17,23 +17,15 @@ final class TracingObservable<T> extends Observable<T> {
   TracingObservable(Observable<T> source, Context context) {
     this.source = source;
     this.context = context;
-    System.out.println("TracingObservable created with context: " + context);
   }
 
   @Override
   protected void subscribeActual(Observer<? super T> observer) {
-    System.out.println("TracingObservable.subscribeActual called with context: " + context);
-    System.out.println("Current context before making current: " + Context.current());
-
     try (Scope ignored = context.makeCurrent()) {
-      System.out.println("Current context after making current: " + Context.current());
-
-      // Don't double-wrap if the observer is already a TracingObserver
+      // Don't double-wrap if already a TracingObserver
       if (observer instanceof TracingObserver) {
-        System.out.println("Observer is already TracingObserver, not wrapping");
         source.subscribe(observer);
       } else {
-        System.out.println("Wrapping observer in TracingObserver");
         source.subscribe(new TracingObserver<>(observer, context));
       }
     }

instrumentation/rxjava/rxjava-2.0/testing/src/main/java/io/opentelemetry/instrumentation/rxjava/v2_0/AbstractRxJava2Test.java

@@ -10,8 +10,8 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
-import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
 import io.opentelemetry.instrumentation.testing.util.ThrowingRunnable;
@@ -30,7 +30,9 @@
 import io.reactivex.schedulers.Schedulers;
 import java.util.Comparator;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.Test;
@@ -340,29 +342,44 @@ public void basicObservable() {
                             .hasParent(trace.getSpan(0))));
   }
 
-
   @Test
-  @SuppressWarnings("CanIgnoreReturnValue")
-  public void basicObservableFromCallable() {
-    Observable<String> test = createParentSpan(() -> Observable.fromCallable(() -> "success"));
-    assertThat(test).isNotNull();
-
-    // Actually subscribe to trigger the Observable execution
-    String result = test.blockingFirst();
-    assertThat(result).isEqualTo("success");
-
+  public void basicObservableFromCallable() throws InterruptedException {
+    CountDownLatch countDownLatch = new CountDownLatch(1);
+    AtomicReference<Throwable> errorRef = new AtomicReference<>();
     testing()
-        .waitAndAssertTraces(
-            trace ->
-                trace.hasSpansSatisfyingExactly(
-                    span -> span.hasName(PARENT).hasKind(SpanKind.INTERNAL).hasNoParent(),
-                    span ->
-                        span.hasName(ADD_ONE)
-                            .hasKind(SpanKind.INTERNAL)
-                            .hasParent(trace.getSpan(0))));
+        .runWithSpan(
+            "parent",
+            () -> {
+              String traceId = Span.current().getSpanContext().getTraceId();
+
+              Disposable result =
+                  Observable.fromCallable(
+                          () -> {
+                            assertThat(traceId)
+                                .isEqualTo(Span.current().getSpanContext().getTraceId());
+                            return "success";
+                          })
+                      .subscribeOn(Schedulers.io())
+                      .observeOn(Schedulers.single())
+                      .doOnNext(
+                          data ->
+                              assertThat(traceId)
+                                  .isEqualTo(Span.current().getSpanContext().getTraceId()))
+                      .subscribe(
+                          data -> countDownLatch.countDown(),
+                          error -> {
+                            errorRef.set(error);
+                            countDownLatch.countDown();
+                          });
+              assertThat(result).isNotNull();
+            });
+    countDownLatch.await();
+    if (errorRef.get() != null) {
+      throw new AssertionError("Assertion failed in observable thread", errorRef.get());
+    }
+    testing().waitForTraces(1);
   }
 
-
   @Test
   public void connectableFlowable() {
     List<Integer> result =
@@ -804,27 +821,6 @@ public void observableChainHasSubscriptionContext() {
                             .hasParent(trace.getSpan(0))));
   }
 
-  @Test
-  public void basicObservableFromCallableTest() {
-    Disposable disposable = Observable.fromCallable(() -> "success")
-        .onErrorReturnItem("")
-        .subscribeOn(Schedulers.io())
-        .observeOn(Schedulers.single())
-        .subscribe(data -> System.out.print("done"));
-
-    assertThat(disposable).isNotNull();
-
-    testing()
-        .waitAndAssertTraces(
-            trace ->
-                trace.hasSpansSatisfyingExactly(
-                    span -> span.hasName(PARENT).hasKind(SpanKind.INTERNAL).hasNoParent(),
-                    span ->
-                        span.hasName(ADD_ONE)
-                            .hasKind(SpanKind.INTERNAL)
-                            .hasParent(trace.getSpan(0))));
-  }
-
   @MethodSource("schedulers")
   @ParameterizedTest
   public void flowableMultiResults(Scheduler scheduler) {