Update rxjava-1 to new agent api
MODIFY
dd-java-agent/instrumentation/rxjava-1/src/main/java/datadog/trace/instrumentation/rxjava/SpanFinishingSubscription.java
dd-java-agent/instrumentation/rxjava-1/src/main/java/datadog/trace/instrumentation/rxjava/SpanFinishingSubscription.java
package datadog.trace.instrumentation.rxjava;
import datadog.trace.agent.decorator.BaseDecorator;
-import io.opentracing.Span;
+import datadog.trace.instrumentation.api.AgentSpan;
import java.util.concurrent.atomic.AtomicReference;
import rx.Subscription;
public class SpanFinishingSubscription implements Subscription {
private final BaseDecorator decorator;
- private final AtomicReference<Span> spanRef;
+ private final AtomicReference<AgentSpan> spanRef;
public SpanFinishingSubscription(
- final BaseDecorator decorator, final AtomicReference<Span> spanRef) {
+ final BaseDecorator decorator, final AtomicReference<AgentSpan> spanRef) {
this.decorator = decorator;
this.spanRef = spanRef;
}
@Override
public void unsubscribe() {
- final Span span = spanRef.getAndSet(null);
+ final AgentSpan span = spanRef.getAndSet(null);
if (span != null) {
decorator.beforeFinish(span);
span.finish();

MODIFY
dd-java-agent/instrumentation/rxjava-1/src/main/java/datadog/trace/instrumentation/rxjava/TracedOnSubscribe.java
dd-java-agent/instrumentation/rxjava-1/src/main/java/datadog/trace/instrumentation/rxjava/TracedOnSubscribe.java
package datadog.trace.instrumentation.rxjava;
+import static datadog.trace.instrumentation.api.AgentTracer.activateSpan;
+import static datadog.trace.instrumentation.api.AgentTracer.propagate;
+import static datadog.trace.instrumentation.api.AgentTracer.startSpan;
+
import datadog.trace.agent.decorator.BaseDecorator;
import datadog.trace.context.TraceScope;
-import io.opentracing.Scope;
-import io.opentracing.Span;
-import io.opentracing.Tracer;
-import io.opentracing.util.GlobalTracer;
+import datadog.trace.instrumentation.api.AgentScope;
+import datadog.trace.instrumentation.api.AgentSpan;
import rx.DDTracingUtil;
import rx.Observable;
import rx.Subscriber;
@@ -21,38 +23,33 @@ public class TracedOnSubscribe<T> implements Observable.OnSubscribe<T> {
final Observable originalObservable,
final String operationName,
final BaseDecorator decorator) {
- this.delegate = DDTracingUtil.extractOnSubscribe(originalObservable);
+ delegate = DDTracingUtil.extractOnSubscribe(originalObservable);
this.operationName = operationName;
this.decorator = decorator;
- final Scope parentScope = GlobalTracer.get().scopeManager().active();
-
- continuation = parentScope instanceof TraceScope ? ((TraceScope) parentScope).capture() : null;
+ continuation = propagate().capture();
}
@Override
public void call(final Subscriber<? super T> subscriber) {
- final Tracer tracer = GlobalTracer.get();
- final Span span; // span finished by TracedSubscriber
+ final AgentSpan span; // span finished by TracedSubscriber
if (continuation != null) {
try (final TraceScope scope = continuation.activate()) {
- span = tracer.buildSpan(operationName).start();
+ span = startSpan(operationName);
}
} else {
- span = tracer.buildSpan(operationName).start();
+ span = startSpan(operationName);
}
afterStart(span);
- try (final Scope scope = tracer.scopeManager().activate(span, false)) {
- if (!((TraceScope) scope).isAsyncPropagating()) {
- ((TraceScope) scope).setAsyncPropagation(true);
- }
+ try (final AgentScope scope = activateSpan(span, false)) {
+ scope.setAsyncPropagation(true);
delegate.call(new TracedSubscriber(span, subscriber, decorator));
}
}
- protected void afterStart(final Span span) {
+ protected void afterStart(final AgentSpan span) {
decorator.afterStart(span);
}
}

MODIFY
dd-java-agent/instrumentation/rxjava-1/src/main/java/datadog/trace/instrumentation/rxjava/TracedSubscriber.java
dd-java-agent/instrumentation/rxjava-1/src/main/java/datadog/trace/instrumentation/rxjava/TracedSubscriber.java
package datadog.trace.instrumentation.rxjava;
+import static datadog.trace.instrumentation.api.AgentTracer.activateSpan;
+
import datadog.trace.agent.decorator.BaseDecorator;
-import datadog.trace.context.TraceScope;
-import io.opentracing.Scope;
-import io.opentracing.ScopeManager;
-import io.opentracing.Span;
-import io.opentracing.util.GlobalTracer;
+import datadog.trace.instrumentation.api.AgentScope;
+import datadog.trace.instrumentation.api.AgentSpan;
import java.util.concurrent.atomic.AtomicReference;
import rx.Subscriber;
public class TracedSubscriber<T> extends Subscriber<T> {
- private final ScopeManager scopeManager = GlobalTracer.get().scopeManager();
- private final AtomicReference<Span> spanRef;
+ private final AtomicReference<AgentSpan> spanRef;
private final Subscriber<T> delegate;
private final BaseDecorator decorator;
public TracedSubscriber(
- final Span span, final Subscriber<T> delegate, final BaseDecorator decorator) {
+ final AgentSpan span, final Subscriber<T> delegate, final BaseDecorator decorator) {
spanRef = new AtomicReference<>(span);
this.delegate = delegate;
this.decorator = decorator;
@@ -28,12 +26,10 @@ public class TracedSubscriber<T> extends Subscriber<T> {
@Override
public void onStart() {
- final Span span = spanRef.get();
+ final AgentSpan span = spanRef.get();
if (span != null) {
- try (final Scope scope = scopeManager.activate(span, false)) {
- if (scope instanceof TraceScope) {
- ((TraceScope) scope).setAsyncPropagation(true);
- }
+ try (final AgentScope scope = activateSpan(span, false)) {
+ scope.setAsyncPropagation(true);
delegate.onStart();
}
} else {
@@ -43,12 +39,10 @@ public class TracedSubscriber<T> extends Subscriber<T> {
@Override
public void onNext(final T value) {
- final Span span = spanRef.get();
+ final AgentSpan span = spanRef.get();
if (span != null) {
- try (final Scope scope = scopeManager.activate(span, false)) {
- if (scope instanceof TraceScope) {
- ((TraceScope) scope).setAsyncPropagation(true);
- }
+ try (final AgentScope scope = activateSpan(span, false)) {
+ scope.setAsyncPropagation(true);
delegate.onNext(value);
} catch (final Throwable e) {
onError(e);
@@ -60,13 +54,11 @@ public class TracedSubscriber<T> extends Subscriber<T> {
@Override
public void onCompleted() {
- final Span span = spanRef.getAndSet(null);
+ final AgentSpan span = spanRef.getAndSet(null);
if (span != null) {
boolean errored = false;
- try (final Scope scope = scopeManager.activate(span, false)) {
- if (scope instanceof TraceScope) {
- ((TraceScope) scope).setAsyncPropagation(true);
- }
+ try (final AgentScope scope = activateSpan(span, false)) {
+ scope.setAsyncPropagation(true);
delegate.onCompleted();
} catch (final Throwable e) {
// Repopulate the spanRef for onError
@@ -87,12 +79,10 @@ public class TracedSubscriber<T> extends Subscriber<T> {
@Override
public void onError(final Throwable e) {
- final Span span = spanRef.getAndSet(null);
+ final AgentSpan span = spanRef.getAndSet(null);
if (span != null) {
- try (final Scope scope = scopeManager.activate(span, false)) {
- if (scope instanceof TraceScope) {
- ((TraceScope) scope).setAsyncPropagation(true);
- }
+ try (final AgentScope scope = activateSpan(span, false)) {
+ scope.setAsyncPropagation(true);
decorator.onError(span, e);
delegate.onError(e);
} catch (final Throwable e2) {

Update rxjava-1 to new agent api