Fix armeria latest dep tests
MODIFY
instrumentation/armeria-1.3/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/armeria/v1_3/AbstractStreamMessageSubscriptionInstrumentation.java
instrumentation/armeria-1.3/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/armeria/v1_3/AbstractStreamMessageSubscriptionInstrumentation.java
@@ -11,6 +11,7 @@ import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import java.util.concurrent.CompletableFuture;
import net.bytebuddy.asm.Advice;
import net.bytebuddy.description.type.TypeDescription;
import net.bytebuddy.matcher.ElementMatcher;
@@ -30,15 +31,32 @@ public class AbstractStreamMessageSubscriptionInstrumentation implements TypeIns
takesArgument(0, named("com.linecorp.armeria.common.stream.AbstractStreamMessage")))
.and(takesArgument(1, named("org.reactivestreams.Subscriber"))),
AbstractStreamMessageSubscriptionInstrumentation.class.getName() + "$WrapSubscriberAdvice");
+ // since 1.9.0
+ transformer.applyAdviceToMethod(
+ isConstructor()
+ .and(
+ takesArgument(0, named("com.linecorp.armeria.common.stream.AbstractStreamMessage")))
+ .and(takesArgument(4, named("java.util.concurrent.CompletableFuture"))),
+ AbstractStreamMessageSubscriptionInstrumentation.class.getName()
+ + "$WrapCompletableFutureAdvice");
}
@SuppressWarnings("unused")
public static class WrapSubscriberAdvice {
@Advice.OnMethodEnter(suppress = Throwable.class)
- public static void attachContext(
- @Advice.Argument(value = 1, readOnly = false) Subscriber subscriber) {
+ public static void wrapSubscriber(
+ @Advice.Argument(value = 1, readOnly = false) Subscriber<?> subscriber) {
subscriber = SubscriberWrapper.wrap(subscriber);
}
}
+
+ public static class WrapCompletableFutureAdvice {
+
+ @Advice.OnMethodEnter(suppress = Throwable.class)
+ public static void wrapCompletableFuture(
+ @Advice.Argument(value = 4, readOnly = false) CompletableFuture<?> future) {
+ future = CompletableFutureWrapper.wrap(future);
+ }
+ }
}

ADD
instrumentation/armeria-1.3/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/armeria/v1_3/CompletableFutureWrapper.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.armeria.v1_3;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import java.util.concurrent.CompletableFuture;
+
+public final class CompletableFutureWrapper {
+
+ private CompletableFutureWrapper() {}
+
+ public static <T> CompletableFuture<T> wrap(CompletableFuture<T> future) {
+ if (future == null) {
+ return null;
+ }
+ Context context = Context.current();
+ if (context != Context.root()) {
+ return wrap(future, context);
+ }
+ return future;
+ }
+
+ private static <T> CompletableFuture<T> wrap(CompletableFuture<T> future, Context context) {
+ CompletableFuture<T> result = new CompletableFuture<>();
+ result.whenComplete(
+ (T value, Throwable throwable) -> {
+ try (Scope ignored = context.makeCurrent()) {
+ if (throwable != null) {
+ future.completeExceptionally(throwable);
+ } else {
+ future.complete(value);
+ }
+ }
+ });
+
+ return result;
+ }
+}

MODIFY
instrumentation/armeria-1.3/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/armeria/v1_3/SubscriberWrapper.java
instrumentation/armeria-1.3/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/armeria/v1_3/SubscriberWrapper.java
@@ -10,19 +10,49 @@ import io.opentelemetry.context.Scope;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
-public class SubscriberWrapper implements Subscriber<Object> {
- private final Subscriber<Object> delegate;
+public class SubscriberWrapper<T> implements Subscriber<T> {
+ private static final Class<?> abortingSubscriberClass = getAbortingSubscriberClass();
+ private static final Class<?> noopSubscriberClass = getNoopSubscriberClass();
+
+ private final Subscriber<T> delegate;
private final Context context;
- private SubscriberWrapper(Subscriber<Object> delegate, Context context) {
+ private static Class<?> getAbortingSubscriberClass() {
+ // AbortingSubscriber is package private
+ try {
+ return Class.forName("com.linecorp.armeria.common.stream.AbortingSubscriber");
+ } catch (ClassNotFoundException exception) {
+ return null;
+ }
+ }
+
+ private static Class<?> getNoopSubscriberClass() {
+ // NoopSubscriber is package private
+ try {
+ return Class.forName("com.linecorp.armeria.common.stream.NoopSubscriber");
+ } catch (ClassNotFoundException exception) {
+ return null;
+ }
+ }
+
+ private SubscriberWrapper(Subscriber<T> delegate, Context context) {
this.delegate = delegate;
this.context = context;
}
- public static Subscriber<Object> wrap(Subscriber<Object> delegate) {
+ private static <T> boolean isIgnored(Subscriber<T> delegate) {
+ return (abortingSubscriberClass != null && abortingSubscriberClass.isInstance(delegate))
+ || (noopSubscriberClass != null && noopSubscriberClass.isInstance(delegate));
+ }
+
+ public static <T> Subscriber<T> wrap(Subscriber<T> delegate) {
+ if (isIgnored(delegate)) {
+ return delegate;
+ }
+
Context context = Context.current();
if (context != Context.root()) {
- return new SubscriberWrapper(delegate, context);
+ return new SubscriberWrapper<>(delegate, context);
}
return delegate;
}
@@ -35,7 +65,7 @@ public class SubscriberWrapper implements Subscriber<Object> {
}
@Override
- public void onNext(Object o) {
+ public void onNext(T o) {
try (Scope ignored = context.makeCurrent()) {
delegate.onNext(o);
}

Fix armeria latest dep tests (#3407)