Fix latest dep RxJava2 tests
And remove latestDepTest version bounds
MODIFY
instrumentation/rxjava-2.0/javaagent/rxjava-2.0-javaagent.gradle
instrumentation/rxjava-2.0/javaagent/rxjava-2.0-javaagent.gradle
@@ -15,5 +15,4 @@ dependencies {
implementation project(":instrumentation:rxjava-2.0:library")
testImplementation project(':instrumentation:rxjava-2.0:testing')
- latestDepTestLibrary group: 'io.reactivex.rxjava2', name: 'rxjava', version: "(,2.2.0)"
}

MODIFY
instrumentation/rxjava-2.0/library/rxjava-2.0-library.gradle
instrumentation/rxjava-2.0/library/rxjava-2.0-library.gradle
@@ -4,5 +4,4 @@ dependencies {
library group: 'io.reactivex.rxjava2', name: 'rxjava', version: "2.1.3"
testImplementation project(':instrumentation:rxjava-2.0:testing')
- latestDepTestLibrary group: 'io.reactivex.rxjava2', name: 'rxjava', version: "(,2.2.0)"
}

MODIFY
instrumentation/rxjava-2.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava2/TracingConditionalSubscriber.java
instrumentation/rxjava-2.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava2/TracingConditionalSubscriber.java
@@ -30,39 +30,43 @@ import io.reactivex.internal.subscribers.BasicFuseableConditionalSubscriber;
class TracingConditionalSubscriber<T> extends BasicFuseableConditionalSubscriber<T, T> {
+ // BasicFuseableConditionalSubscriber#actual has been renamed to downstream in newer versions, we
+ // can't use it in this class
+ private final ConditionalSubscriber<? super T> wrappedSubscriber;
private final Context context;
TracingConditionalSubscriber(
final ConditionalSubscriber<? super T> actual, final Context context) {
super(actual);
+ this.wrappedSubscriber = actual;
this.context = context;
}
@Override
public boolean tryOnNext(T t) {
try (Scope ignored = context.makeCurrent()) {
- return actual.tryOnNext(t);
+ return wrappedSubscriber.tryOnNext(t);
}
}
@Override
public void onNext(T t) {
try (Scope ignored = context.makeCurrent()) {
- actual.onNext(t);
+ wrappedSubscriber.onNext(t);
}
}
@Override
public void onError(Throwable t) {
try (Scope ignored = context.makeCurrent()) {
- actual.onError(t);
+ wrappedSubscriber.onError(t);
}
}
@Override
public void onComplete() {
try (Scope ignored = context.makeCurrent()) {
- actual.onComplete();
+ wrappedSubscriber.onComplete();
}
}

MODIFY
instrumentation/rxjava-2.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava2/TracingObserver.java
instrumentation/rxjava-2.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava2/TracingObserver.java
@@ -28,33 +28,37 @@ import io.reactivex.Observer;
import io.reactivex.internal.fuseable.QueueDisposable;
import io.reactivex.internal.observers.BasicFuseableObserver;
-public class TracingObserver<T> extends BasicFuseableObserver<T, T> {
+class TracingObserver<T> extends BasicFuseableObserver<T, T> {
+ // BasicFuseableObserver#actual has been renamed to downstream in newer versions, we can't use it
+ // in this class
+ private final Observer<? super T> wrappedObserver;
private final Context context;
TracingObserver(final Observer<? super T> actual, final Context context) {
super(actual);
+ this.wrappedObserver = actual;
this.context = context;
}
@Override
public void onNext(T t) {
try (Scope ignored = context.makeCurrent()) {
- actual.onNext(t);
+ wrappedObserver.onNext(t);
}
}
@Override
public void onError(Throwable t) {
try (Scope ignored = context.makeCurrent()) {
- actual.onError(t);
+ wrappedObserver.onError(t);
}
}
@Override
public void onComplete() {
try (Scope ignored = context.makeCurrent()) {
- actual.onComplete();
+ wrappedObserver.onComplete();
}
}

MODIFY
instrumentation/rxjava-2.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava2/TracingSubscriber.java
instrumentation/rxjava-2.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava2/TracingSubscriber.java
@@ -30,31 +30,35 @@ import org.reactivestreams.Subscriber;
class TracingSubscriber<T> extends BasicFuseableSubscriber<T, T> {
+ // BasicFuseableSubscriber#actual has been renamed to downstream in newer versions, we can't use
+ // it in this class
+ private final Subscriber<? super T> wrappedSubscriber;
private final Context context;
TracingSubscriber(final Subscriber<? super T> actual, final Context context) {
super(actual);
+ this.wrappedSubscriber = actual;
this.context = context;
}
@Override
public void onNext(T t) {
try (Scope ignored = context.makeCurrent()) {
- actual.onNext(t);
+ wrappedSubscriber.onNext(t);
}
}
@Override
public void onError(Throwable t) {
try (Scope ignored = context.makeCurrent()) {
- actual.onError(t);
+ wrappedSubscriber.onError(t);
}
}
@Override
public void onComplete() {
try (Scope ignored = context.makeCurrent()) {
- actual.onComplete();
+ wrappedSubscriber.onComplete();
}
}

Fix latest dep RxJava2 tests (#2700)

And remove latestDepTest version bounds