Improve wrapping jetty response listener (#3831)
This commit is contained in:
parent
539f104869
commit
392fc5a6c1
|
@ -9,10 +9,22 @@ import static java.util.stream.Collectors.toList;
|
||||||
|
|
||||||
import io.opentelemetry.context.Context;
|
import io.opentelemetry.context.Context;
|
||||||
import io.opentelemetry.context.Scope;
|
import io.opentelemetry.context.Scope;
|
||||||
|
import java.lang.reflect.Proxy;
|
||||||
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import org.eclipse.jetty.client.api.Response;
|
import org.eclipse.jetty.client.api.Response;
|
||||||
|
|
||||||
public final class JettyClientWrapUtil {
|
public final class JettyClientWrapUtil {
|
||||||
|
private static final Class<?>[] listenerInterfaces = {
|
||||||
|
Response.CompleteListener.class,
|
||||||
|
Response.FailureListener.class,
|
||||||
|
Response.SuccessListener.class,
|
||||||
|
Response.AsyncContentListener.class,
|
||||||
|
Response.ContentListener.class,
|
||||||
|
Response.HeadersListener.class,
|
||||||
|
Response.HeaderListener.class,
|
||||||
|
Response.BeginListener.class
|
||||||
|
};
|
||||||
|
|
||||||
private JettyClientWrapUtil() {}
|
private JettyClientWrapUtil() {}
|
||||||
|
|
||||||
|
@ -33,17 +45,29 @@ public final class JettyClientWrapUtil {
|
||||||
|
|
||||||
private static Response.ResponseListener wrapTheListener(
|
private static Response.ResponseListener wrapTheListener(
|
||||||
Response.ResponseListener listener, Context context) {
|
Response.ResponseListener listener, Context context) {
|
||||||
Response.ResponseListener wrappedListener = listener;
|
if (listener == null || listener instanceof JettyHttpClient9TracingInterceptor) {
|
||||||
if (listener instanceof Response.CompleteListener
|
return listener;
|
||||||
&& !(listener instanceof JettyHttpClient9TracingInterceptor)) {
|
|
||||||
wrappedListener =
|
|
||||||
(Response.CompleteListener)
|
|
||||||
result -> {
|
|
||||||
try (Scope ignored = context.makeCurrent()) {
|
|
||||||
((Response.CompleteListener) listener).onComplete(result);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
}
|
}
|
||||||
return wrappedListener;
|
|
||||||
|
Class<?> listenerClass = listener.getClass();
|
||||||
|
List<Class<?>> interfaces = new ArrayList<>();
|
||||||
|
for (Class<?> type : listenerInterfaces) {
|
||||||
|
if (type.isInstance(listener)) {
|
||||||
|
interfaces.add(type);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (interfaces.isEmpty()) {
|
||||||
|
return listener;
|
||||||
|
}
|
||||||
|
|
||||||
|
return (Response.ResponseListener)
|
||||||
|
Proxy.newProxyInstance(
|
||||||
|
listenerClass.getClassLoader(),
|
||||||
|
interfaces.toArray(new Class[0]),
|
||||||
|
(proxy, method, args) -> {
|
||||||
|
try (Scope ignored = context.makeCurrent()) {
|
||||||
|
return method.invoke(listener, args);
|
||||||
|
}
|
||||||
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -18,6 +18,7 @@ import org.eclipse.jetty.client.api.Result
|
||||||
import org.eclipse.jetty.http.HttpMethod
|
import org.eclipse.jetty.http.HttpMethod
|
||||||
import org.eclipse.jetty.util.ssl.SslContextFactory
|
import org.eclipse.jetty.util.ssl.SslContextFactory
|
||||||
import spock.lang.Shared
|
import spock.lang.Shared
|
||||||
|
import spock.lang.Unroll
|
||||||
|
|
||||||
abstract class AbstractJettyClient9Test extends HttpClientTest<Request> {
|
abstract class AbstractJettyClient9Test extends HttpClientTest<Request> {
|
||||||
|
|
||||||
|
@ -34,7 +35,6 @@ abstract class AbstractJettyClient9Test extends HttpClientTest<Request> {
|
||||||
Request jettyRequest = null
|
Request jettyRequest = null
|
||||||
|
|
||||||
def setupSpec() {
|
def setupSpec() {
|
||||||
|
|
||||||
//Start the main Jetty HttpClient and a https client
|
//Start the main Jetty HttpClient and a https client
|
||||||
client.start()
|
client.start()
|
||||||
|
|
||||||
|
@ -46,7 +46,6 @@ abstract class AbstractJettyClient9Test extends HttpClientTest<Request> {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
Request buildRequest(String method, URI uri, Map<String, String> headers) {
|
Request buildRequest(String method, URI uri, Map<String, String> headers) {
|
||||||
|
|
||||||
HttpClient theClient = uri.scheme == 'https' ? httpsClient : client
|
HttpClient theClient = uri.scheme == 'https' ? httpsClient : client
|
||||||
|
|
||||||
Request request = theClient.newRequest(uri)
|
Request request = theClient.newRequest(uri)
|
||||||
|
@ -78,25 +77,21 @@ abstract class AbstractJettyClient9Test extends HttpClientTest<Request> {
|
||||||
}
|
}
|
||||||
|
|
||||||
private static class JettyClientListener implements Request.FailureListener, Response.FailureListener {
|
private static class JettyClientListener implements Request.FailureListener, Response.FailureListener {
|
||||||
|
|
||||||
volatile Throwable failure
|
volatile Throwable failure
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
void onFailure(Request requestF, Throwable failure) {
|
void onFailure(Request requestF, Throwable failure) {
|
||||||
this.failure = failure
|
this.failure = failure
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
void onFailure(Response responseF, Throwable failure) {
|
void onFailure(Response responseF, Throwable failure) {
|
||||||
this.failure = failure
|
this.failure = failure
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
void sendRequestWithCallback(Request request, String method, URI uri, Map<String, String> headers, AbstractHttpClientTest.RequestResult requestResult) {
|
void sendRequestWithCallback(Request request, String method, URI uri, Map<String, String> headers, AbstractHttpClientTest.RequestResult requestResult) {
|
||||||
|
|
||||||
JettyClientListener jcl = new JettyClientListener()
|
JettyClientListener jcl = new JettyClientListener()
|
||||||
|
|
||||||
request.onRequestFailure(jcl)
|
request.onRequestFailure(jcl)
|
||||||
|
@ -108,7 +103,6 @@ abstract class AbstractJettyClient9Test extends HttpClientTest<Request> {
|
||||||
request.send(new Response.CompleteListener() {
|
request.send(new Response.CompleteListener() {
|
||||||
@Override
|
@Override
|
||||||
void onComplete(Result result) {
|
void onComplete(Result result) {
|
||||||
|
|
||||||
if (jcl.failure != null) {
|
if (jcl.failure != null) {
|
||||||
requestResult.complete(jcl.failure)
|
requestResult.complete(jcl.failure)
|
||||||
return
|
return
|
||||||
|
@ -119,7 +113,6 @@ abstract class AbstractJettyClient9Test extends HttpClientTest<Request> {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
boolean testRedirects() {
|
boolean testRedirects() {
|
||||||
false
|
false
|
||||||
|
@ -135,4 +128,27 @@ abstract class AbstractJettyClient9Test extends HttpClientTest<Request> {
|
||||||
super.httpAttributes(uri) + extra
|
super.httpAttributes(uri) + extra
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Unroll
|
||||||
|
def "test content of #method request #url"() {
|
||||||
|
when:
|
||||||
|
def request = buildRequest(method, url, [:])
|
||||||
|
def response = request.send()
|
||||||
|
|
||||||
|
then:
|
||||||
|
response.status == 200
|
||||||
|
response.getContentAsString() == "Hello."
|
||||||
|
|
||||||
|
assertTraces(1) {
|
||||||
|
trace(0, 2) {
|
||||||
|
clientSpan(it, 0, null, method, url)
|
||||||
|
serverSpan(it, 1, span(0))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
where:
|
||||||
|
path << ["/success"]
|
||||||
|
|
||||||
|
method = "GET"
|
||||||
|
url = resolveAddress(path)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue