baseline on 3-25-24

rxjava2-2.0/build.gradle

@@ -13,7 +13,11 @@ dependencies {
    implementation 'com.newrelic.agent.java:newrelic-api:6.4.0'
    implementation fileTree(include: ['*.jar'], dir: '../libs')
 
-    testImplementation 'junit:junit:4.12' 
+   testImplementation 'junit:junit:4.12' 
+   testImplementation 'com.newrelic.agent.java:newrelic-agent:6.4.0'
+   testImplementation 'com.newrelic.agent.java:newrelic-api:6.4.0'
+   testImplementation fileTree(include: ['*.jar'], dir: '../libs')
+   testImplementation fileTree(include: ['*.jar'], dir: '../test-lib')
 
 }
 

rxjava2-2.0/src/main/java/com/newrelic/instrumentation/rxjava2/NRSingleObserver2.java

@@ -35,6 +35,9 @@ public void onSubscribe(Disposable d) {
 		downstream.onSubscribe(d);
 	}
 
+	public String getName() {
+		return name;
+	}
 
 	@Override
 	public void onError(Throwable e) {

rxjava2-2.0/src/main/java/com/newrelic/instrumentation/rxjava2/NRSingleSubWrapper.java

@@ -17,10 +17,20 @@ public NRSingleSubWrapper(BiFunction<Single, SingleObserver, SingleObserver> d)
 	@Override
 	public SingleObserver apply(Single t1, SingleObserver t2) throws Exception {
 		if(delegate == null) {
-			return t2 instanceof NRSingleObserver2 ? t2 : new NRSingleObserver2(t2, t1.singleName != null ? t1.singleName : t1.getClass().getSimpleName());
+			NRSingleObserver2 result = t2 instanceof NRSingleObserver2 ? (NRSingleObserver2) t2 : new NRSingleObserver2(t2, t1.singleName != null ? t1.singleName : t1.getClass().getSimpleName());
+			String name = result.getName();
+			if(name.equals("FlowableSingleSingle")) {
+				result.ignore = true;
+			}
+			return result;
 		}
 		SingleObserver actual = delegate.apply(t1, t2);
-		return t2 instanceof NRSingleObserver2 ? actual : new NRSingleObserver2(actual, t1.singleName != null ? t1.singleName : t1.getClass().getSimpleName());
+		NRSingleObserver2 result = t2 instanceof NRSingleObserver2 ? (NRSingleObserver2) actual : new NRSingleObserver2(actual, t1.singleName != null ? t1.singleName : t1.getClass().getSimpleName());
+		String name = result.getName();
+		if(name.equals("FlowableSingleSingle")) {
+			result.ignore = true;
+		}
+		return result;
 	}
 
 	public BiFunction<Single, SingleObserver, SingleObserver> getDelegate() {

rxjava2-2.0/src/main/java/io/reactivex/Observable.java

@@ -1,13 +1,28 @@
 package io.reactivex;
 
+import java.util.concurrent.Callable;
+
 import com.newrelic.api.agent.weaver.MatchType;
 import com.newrelic.api.agent.weaver.NewField;
 import com.newrelic.api.agent.weaver.Weave;
+import com.newrelic.api.agent.weaver.Weaver;
+
+import io.reactivex.disposables.Disposable;
+import io.reactivex.functions.Action;
+import io.reactivex.functions.Consumer;
 
 @Weave(type=MatchType.BaseClass)
 public abstract class Observable<T> {
-	
+
 	@NewField
 	public String observableName = null;
 
+	// Included in order to run tests
+	public static <T> Observable<T> fromCallable(Callable<? extends T> supplier) {
+		return Weaver.callOriginal();
+	}
+
+
+	public abstract Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
+            Action onComplete);
 }

rxjava2-2.0/src/main/java/io/reactivex/internal/operators/flowable/FlowableUtils.java

@@ -10,6 +10,10 @@ public static <T> boolean ignore(Flowable<T> flowable) {
 			return true;
 		}
 
+		if(flowable instanceof FlowableZip) {
+			return true;
+		}
+
 		return false;
 	}
 }

rxjava2-2.0/src/test/java/com/newrelic/instrumentation/labs/test/rxjava2/MyCallable.java

@@ -0,0 +1,17 @@
+package com.newrelic.instrumentation.labs.test.rxjava2;
+
+import java.util.concurrent.Callable;
+
+import com.newrelic.api.agent.Trace;
+
+public class MyCallable implements Callable<String> {
+
+	private String retValue = "Blue";
+
+	@Override
+	@Trace
+	public String call() throws Exception {
+		return retValue;
+	}
+
+}

rxjava2-2.0/src/test/java/com/newrelic/instrumentation/labs/test/rxjava2/MyCompletedAction.java

@@ -0,0 +1,36 @@
+package com.newrelic.instrumentation.labs.test.rxjava2;
+
+import java.util.Random;
+
+import com.newrelic.api.agent.Trace;
+
+import io.reactivex.functions.Action;
+
+public class MyCompletedAction implements Action {
+	private static Random random = new Random();
+	private static int MAXUNITS = 15;
+
+	@Override
+	@Trace(dispatcher=true)
+	public void run() throws Exception {
+		pauseRandomUnits();
+		System.out.println("Object has completed");
+	}
+
+	private void pauseRandomUnits() {
+		int n = random.nextInt(MAXUNITS);
+		pause(n*100L);
+	}
+
+	private void pause(long ms) {
+		if(ms > 0) {
+			try {
+				Thread.sleep(ms);
+			} catch (InterruptedException e) {
+				//e.printStackTrace();
+			}
+		}
+	}
+
+
+}

rxjava2-2.0/src/test/java/com/newrelic/instrumentation/labs/test/rxjava2/MyErrorConsumer.java

@@ -0,0 +1,36 @@
+package com.newrelic.instrumentation.labs.test.rxjava2;
+
+import java.util.Random;
+
+import com.newrelic.api.agent.Trace;
+
+import io.reactivex.functions.Consumer;
+
+public class MyErrorConsumer implements Consumer<Throwable> {
+
+	private static Random random = new Random();
+	private static int MAXUNITS = 15;
+
+	@Override
+	@Trace(dispatcher=true)
+	public void accept(Throwable t) throws Exception {
+		System.out.println("MyErrorConsumer received error: "+t);
+		t.printStackTrace();
+		pauseRandomUnits();
+	}
+
+	private void pauseRandomUnits() {
+		int n = random.nextInt(MAXUNITS);
+		pause(n*100L);
+	}
+
+	private void pause(long ms) {
+		if(ms > 0) {
+			try {
+				Thread.sleep(ms);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+		}
+	}
+}

rxjava2-2.0/src/test/java/com/newrelic/instrumentation/labs/test/rxjava2/MyStringConsumer.java

@@ -0,0 +1,35 @@
+package com.newrelic.instrumentation.labs.test.rxjava2;
+
+import java.util.Random;
+
+import com.newrelic.api.agent.Trace;
+
+import io.reactivex.functions.Consumer;
+
+public class MyStringConsumer implements Consumer<String> {
+
+	private static Random random = new Random();
+	private static int MAXUNITS = 15;
+
+	@Override
+	@Trace(dispatcher=true)
+	public void accept(String t) throws Exception {
+		System.out.println("MyStringConsumer received: "+t);
+		pauseRandomUnits();
+	}
+
+	private void pauseRandomUnits() {
+		int n = random.nextInt(MAXUNITS);
+		pause(n*100L);
+	}
+
+	private void pause(long ms) {
+		if(ms > 0) {
+			try {
+				Thread.sleep(ms);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+		}
+	}
+}

rxjava2-2.0/src/test/java/com/newrelic/instrumentation/labs/test/rxjava2/RxJava2Tests.java

@@ -0,0 +1,31 @@
+package com.newrelic.instrumentation.labs.test.rxjava2;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import com.newrelic.agent.introspec.InstrumentationTestConfig;
+import com.newrelic.agent.introspec.InstrumentationTestRunner;
+import com.newrelic.agent.introspec.Introspector;
+
+import io.reactivex.Observable;
+
+@RunWith(InstrumentationTestRunner.class)
+//Tell the test harness which classes are part of the instrumentation module
+@InstrumentationTestConfig(includePrefixes = { "io.reactivex" })
+public class RxJava2Tests {
+
+	@Test
+	public void testObservable() {
+		System.out.println("Call to testObservable");
+		Introspector introspector = InstrumentationTestRunner.getIntrospector();
+		Observable<String> observable = Observable.fromCallable(new MyCallable());
+		observable.subscribe(new MyStringConsumer(), new MyErrorConsumer(), new MyCompletedAction());
+
+
+
+	}
+
+	public void doObservable() {
+
+	}
+}