Review fixes for rxjava-1.0:library

Automated code review of instrumentation/rxjava/rxjava-1.0/library.

Author: otelbot[bot]

instrumentation/rxjava/rxjava-1.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava/v1_0/TracedOnSubscribe.java

@@ -8,6 +8,7 @@
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import java.util.concurrent.atomic.AtomicReference;
 import rx.Observable;
 import rx.OpenTelemetryTracingUtil;
 import rx.Subscriber;
@@ -41,8 +42,15 @@ public void call(Subscriber<? super T> subscriber) {
      */
 
     Context context = instrumenter.start(parentContext, request);
+    AtomicReference<Context> contextRef = new AtomicReference<>(context);
     try (Scope ignored = context.makeCurrent()) {
-      delegate.call(new TracedSubscriber<>(subscriber, instrumenter, context, request));
+      delegate.call(new TracedSubscriber<>(subscriber, instrumenter, contextRef, request));
+    } catch (Throwable t) {
+      Context spanContext = contextRef.getAndSet(null);
+      if (spanContext != null) {
+        instrumenter.end(spanContext, request, null, t);
+      }
+      throw t;
     }
   }
 }

instrumentation/rxjava/rxjava-1.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava/v1_0/TracedSubscriber.java

@@ -21,11 +21,11 @@ final class TracedSubscriber<T, REQUEST> extends Subscriber<T> {
   TracedSubscriber(
       Subscriber<? super T> delegate,
       Instrumenter<REQUEST, ?> instrumenter,
-      Context context,
+      AtomicReference<Context> contextRef,
       REQUEST request) {
     this.delegate = delegate;
     this.instrumenter = instrumenter;
-    this.contextRef = new AtomicReference<>(context);
+    this.contextRef = contextRef;
     this.request = request;
 
     delegate.add(new SpanFinishingSubscription<>(instrumenter, contextRef, request));