Add connection reuse tests for Reactor Netty libraries
MODIFY
instrumentation/reactor-3.1/library/src/main/java/io/opentelemetry/instrumentation/reactor/TracingSubscriber.java
instrumentation/reactor-3.1/library/src/main/java/io/opentelemetry/instrumentation/reactor/TracingSubscriber.java
@@ -44,8 +44,8 @@ public class TracingSubscriber<T> implements CoreSubscriber<T> {
Context ctx,
io.opentelemetry.context.Context contextToPropagate) {
this.subscriber = subscriber;
- this.traceContext = contextToPropagate;
this.context = ctx;
+ this.traceContext = contextToPropagate;
}
@Override

MODIFY
instrumentation/reactor-netty-0.9/javaagent/reactor-netty-0.9-javaagent.gradle
instrumentation/reactor-netty-0.9/javaagent/reactor-netty-0.9-javaagent.gradle
@@ -19,4 +19,5 @@ dependencies {
testInstrumentation project(':instrumentation:reactor-netty-1.0:javaagent')
testInstrumentation project(':instrumentation:netty:netty-4.1:javaagent')
+ testInstrumentation project(':instrumentation:reactor-3.1:javaagent')
}

MODIFY
instrumentation/reactor-netty-0.9/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/reactornetty/v0_9/ReactorNettyInstrumentationModule.java
instrumentation/reactor-netty-0.9/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/reactornetty/v0_9/ReactorNettyInstrumentationModule.java
@@ -64,7 +64,7 @@ public class ReactorNettyInstrumentationModule extends InstrumentationModule {
@Override
public Map<? extends ElementMatcher<? super MethodDescription>, String> transformers() {
return singletonMap(
- isStatic().and(named("create")),
+ isStatic().and(named("create").or(named("newConnection"))),
ReactorNettyInstrumentationModule.class.getName() + "$CreateAdvice");
}
}

MODIFY
instrumentation/reactor-netty-0.9/javaagent/src/test/groovy/ReactorNettyHttpClientTest.groovy
instrumentation/reactor-netty-0.9/javaagent/src/test/groovy/ReactorNettyHttpClientTest.groovy
import io.opentelemetry.instrumentation.test.AgentTestTrait
import io.opentelemetry.instrumentation.test.base.HttpClientTest
+import io.opentelemetry.instrumentation.test.base.SingleConnection
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeoutException
import reactor.netty.http.client.HttpClient
import reactor.netty.http.client.HttpClientResponse
@@ -45,4 +48,33 @@ class ReactorNettyHttpClientTest extends HttpClientTest implements AgentTestTrai
}
return resp.status().code()
}
+
+ @Override
+ SingleConnection createSingleConnection(String host, int port) {
+ String url
+ try {
+ url = new URL("http", host, port, "").toString()
+ } catch (MalformedURLException e) {
+ throw new ExecutionException(e)
+ }
+
+ def httpClient = HttpClient
+ .newConnection()
+ .baseUrl(url)
+
+ return new SingleConnection() {
+
+ @Override
+ int doRequest(String path, Map<String, String> headers) throws ExecutionException, InterruptedException, TimeoutException {
+ return httpClient
+ .headers({ h -> headers.each { k, v -> h.add(k, v) } })
+ .get()
+ .uri(path)
+ .response()
+ .block()
+ .status().code()
+ }
+ }
+ }
+
}

MODIFY
instrumentation/reactor-netty-1.0/javaagent/reactor-netty-1.0-javaagent.gradle
instrumentation/reactor-netty-1.0/javaagent/reactor-netty-1.0-javaagent.gradle
@@ -20,4 +20,5 @@ dependencies {
testInstrumentation project(':instrumentation:reactor-netty-0.9:javaagent')
testInstrumentation project(':instrumentation:netty:netty-4.1:javaagent')
+ testInstrumentation project(':instrumentation:reactor-3.1:javaagent')
}

MODIFY
instrumentation/reactor-netty-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/reactornetty/v1_0/ReactorNettyInstrumentationModule.java
instrumentation/reactor-netty-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/reactornetty/v1_0/ReactorNettyInstrumentationModule.java
@@ -63,7 +63,7 @@ public class ReactorNettyInstrumentationModule extends InstrumentationModule {
@Override
public Map<? extends ElementMatcher<? super MethodDescription>, String> transformers() {
return singletonMap(
- isStatic().and(named("create")),
+ isStatic().and(named("create").or(named("newConnection"))),
ReactorNettyInstrumentationModule.class.getName() + "$CreateAdvice");
}
}

MODIFY
instrumentation/reactor-netty-1.0/javaagent/src/test/groovy/ReactorNettyHttpClientTest.groovy
instrumentation/reactor-netty-1.0/javaagent/src/test/groovy/ReactorNettyHttpClientTest.groovy
import io.opentelemetry.instrumentation.test.AgentTestTrait
import io.opentelemetry.instrumentation.test.base.HttpClientTest
+import io.opentelemetry.instrumentation.test.base.SingleConnection
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeoutException
import reactor.netty.http.client.HttpClient
import reactor.netty.http.client.HttpClientResponse
@@ -45,4 +48,26 @@ class ReactorNettyHttpClientTest extends HttpClientTest implements AgentTestTrai
}
return resp.status().code()
}
+
+ @Override
+ SingleConnection createSingleConnection(String host, int port) {
+ def httpClient = HttpClient
+ .newConnection()
+ .host(host)
+ .port(port)
+
+ return new SingleConnection() {
+
+ @Override
+ int doRequest(String path, Map<String, String> headers) throws ExecutionException, InterruptedException, TimeoutException {
+ return httpClient
+ .headers({ h -> headers.each { k, v -> h.add(k, v) } })
+ .get()
+ .uri(path)
+ .response()
+ .block()
+ .status().code()
+ }
+ }
+ }
}

Add connection reuse tests for Reactor Netty libraries (#2647)