refactor: reduce the complexity of messaging-wrappers

Author: Cirilla-zmh

messaging-wrappers/README.md

@@ -150,25 +150,33 @@ Below is an example of how to initialize a messaging wrapper.
 ```java
 public class Demo {
 
-  public static MessagingProcessWrapper<MyMessagingProcessRequest> createWrapper(
-      OpenTelemetry openTelemetry,
-      MyTextMapGetter textMapGetter,
-      List<AttributesExtractor<MyMessagingProcessRequest, Void>> additionalExtractor) {
+  public static MessagingProcessWrapper<MessagingProcessRequest> createWrapper(OpenTelemetry openTelemetry) {
 
-    return MessagingProcessWrapper.<MyMessagingProcessRequest>defaultBuilder()
+    return MessagingProcessWrapper.<MessagingProcessRequest>defaultBuilder()
         .openTelemetry(openTelemetry)
-        .textMapGetter(textMapGetter)
-        .addAttributesExtractors(additionalExtractor)
+        .textMapGetter(DefaultMessageTextMapGetter.create())
+        .attributesExtractors(
+            Collections.singletonList(
+                MessagingAttributesExtractor.create(
+                    DefaultMessagingAttributesGetter.create(), MessageOperation.PROCESS)))
         .build();
   }
 }
 
 public class MyMessagingProcessRequest implements MessagingProcessRequest {
   // your implementation here
-}
 
-public class MyTextMapGetter implements TextMapGetter<MyMessagingProcessRequest> {
-  // your implementation here
+  @Override
+  public List<String> getMessageHeader(String name) {
+    // impl: how to get specific header from your message
+    return Collections.singletonList(message.getHeaders().get(name));
+  }
+
+  @Override
+  public Collection<String> getAllMessageHeadersKey() {
+    // impl: how to get all headers key set from your message
+    return message.getHeaders().keySet();
+  }
 }
 ```
 

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsHelper.java

@@ -9,14 +9,12 @@
 import com.aliyun.mns.model.MessagePropertyValue;
 import com.aliyun.mns.model.MessageSystemPropertyName;
 import com.aliyun.mns.model.MessageSystemPropertyValue;
-import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MnsProcessRequest;
 import javax.annotation.Nullable;
 
 public final class MnsHelper {
 
-  public static <REQUEST extends MnsProcessRequest>
-      MnsProcessWrapperBuilder<REQUEST> processWrapperBuilder() {
-    return new MnsProcessWrapperBuilder<>();
+  public static MnsProcessWrapperBuilder processWrapperBuilder() {
+    return new MnsProcessWrapperBuilder();
   }
 
   @Nullable

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsProcessWrapperBuilder.java

@@ -5,14 +5,25 @@
 
 package io.opentelemetry.contrib.messaging.wrappers.mns;
 
-import io.opentelemetry.contrib.messaging.wrappers.DefaultMessagingProcessWrapperBuilder;
+import io.opentelemetry.contrib.messaging.wrappers.MessagingProcessWrapperBuilder;
+import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MnsConsumerAttributesGetter;
 import io.opentelemetry.contrib.messaging.wrappers.mns.semconv.MnsProcessRequest;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
+import java.util.ArrayList;
 
-public class MnsProcessWrapperBuilder<REQUEST extends MnsProcessRequest>
-    extends DefaultMessagingProcessWrapperBuilder<REQUEST> {
+public class MnsProcessWrapperBuilder extends MessagingProcessWrapperBuilder<MnsProcessRequest> {
 
   MnsProcessWrapperBuilder() {
     super();
     super.textMapGetter = MnsTextMapGetter.create();
+    super.spanNameExtractor =
+        MessagingSpanNameExtractor.create(
+            MnsConsumerAttributesGetter.INSTANCE, MessageOperation.PROCESS);
+    super.attributesExtractors = new ArrayList<>();
+    super.attributesExtractors.add(
+        MessagingAttributesExtractor.create(
+            MnsConsumerAttributesGetter.INSTANCE, MessageOperation.PROCESS));
   }
 }

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/MnsTextMapGetter.java

@@ -19,14 +19,14 @@
 import java.util.Map;
 import javax.annotation.Nullable;
 
-public class MnsTextMapGetter<REQUEST extends MnsProcessRequest> implements TextMapGetter<REQUEST> {
+public class MnsTextMapGetter implements TextMapGetter<MnsProcessRequest> {
 
-  public static <REQUEST extends MnsProcessRequest> TextMapGetter<REQUEST> create() {
-    return new MnsTextMapGetter<>();
+  public static TextMapGetter<MnsProcessRequest> create() {
+    return new MnsTextMapGetter();
   }
 
   @Override
-  public Iterable<String> keys(@Nullable REQUEST carrier) {
+  public Iterable<String> keys(@Nullable MnsProcessRequest carrier) {
     if (carrier == null || carrier.getMessage() == null) {
       return emptyList();
     }
@@ -48,7 +48,7 @@ public Iterable<String> keys(@Nullable REQUEST carrier) {
 
   @Nullable
   @Override
-  public String get(@Nullable REQUEST carrier, String key) {
+  public String get(@Nullable MnsProcessRequest carrier, String key) {
     if (carrier == null || carrier.getMessage() == null) {
       return null;
     }

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/semconv/MnsConsumerAttributesGetter.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.messaging.wrappers.mns.semconv;
+
+import io.opentelemetry.contrib.messaging.wrappers.mns.MnsHelper;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+
+public enum MnsConsumerAttributesGetter
+    implements MessagingAttributesGetter<MnsProcessRequest, Void> {
+  INSTANCE;
+
+  @Override
+  public String getSystem(MnsProcessRequest request) {
+    return "smq";
+  }
+
+  @Nullable
+  @Override
+  public String getDestination(MnsProcessRequest request) {
+    return request.getDestination();
+  }
+
+  @Nullable
+  @Override
+  public String getDestinationTemplate(MnsProcessRequest request) {
+    return null;
+  }
+
+  @Override
+  public boolean isTemporaryDestination(MnsProcessRequest request) {
+    return false;
+  }
+
+  @Override
+  public boolean isAnonymousDestination(MnsProcessRequest request) {
+    return false;
+  }
+
+  @Override
+  @Nullable
+  public String getConversationId(MnsProcessRequest request) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public Long getMessageBodySize(MnsProcessRequest request) {
+    return (long) request.getMessage().getMessageBodyAsBytes().length;
+  }
+
+  @Nullable
+  @Override
+  public Long getMessageEnvelopeSize(MnsProcessRequest request) {
+    return (long) request.getMessage().getMessageBodyAsRawBytes().length;
+  }
+
+  @Override
+  @Nullable
+  public String getMessageId(MnsProcessRequest request, @Nullable Void unused) {
+    return request.getMessage().getMessageId();
+  }
+
+  @Nullable
+  @Override
+  public String getClientId(MnsProcessRequest request) {
+    return null;
+  }
+
+  @Nullable
+  @Override
+  public Long getBatchMessageCount(MnsProcessRequest request, @Nullable Void unused) {
+    return null;
+  }
+
+  @Override
+  public List<String> getMessageHeader(MnsProcessRequest request, String name) {
+    String header = MnsHelper.getMessageHeader(request.getMessage(), name);
+    if (header == null) {
+      return Collections.emptyList();
+    }
+    return Collections.singletonList(header);
+  }
+}

messaging-wrappers/aliyun-mns-sdk/src/main/java/io/opentelemetry/contrib/messaging/wrappers/mns/semconv/MnsProcessRequest.java

@@ -6,13 +6,9 @@
 package io.opentelemetry.contrib.messaging.wrappers.mns.semconv;
 
 import com.aliyun.mns.model.Message;
-import io.opentelemetry.contrib.messaging.wrappers.mns.MnsHelper;
-import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessRequest;
-import java.util.Collections;
-import java.util.List;
 import javax.annotation.Nullable;
 
-public class MnsProcessRequest implements MessagingProcessRequest {
+public class MnsProcessRequest {
 
   private final Message message;
 
@@ -26,70 +22,15 @@ public static MnsProcessRequest of(Message message, @Nullable String destination
     return new MnsProcessRequest(message, destination);
   }
 
-  @Override
-  public String getSystem() {
-    return "smq";
+  public Message getMessage() {
+    return message;
   }
 
   @Nullable
-  @Override
   public String getDestination() {
     return this.destination;
   }
 
-  @Nullable
-  @Override
-  public String getDestinationTemplate() {
-    return null;
-  }
-
-  @Override
-  public boolean isTemporaryDestination() {
-    return false;
-  }
-
-  @Override
-  public boolean isAnonymousDestination() {
-    return false;
-  }
-
-  @Nullable
-  @Override
-  public String getConversationId() {
-    return null;
-  }
-
-  @Nullable
-  @Override
-  public Long getMessageBodySize() {
-    return (long) message.getMessageBodyAsBytes().length;
-  }
-
-  @Nullable
-  @Override
-  public Long getMessageEnvelopeSize() {
-    return (long) message.getMessageBodyAsRawBytes().length;
-  }
-
-  @Nullable
-  @Override
-  public String getMessageId() {
-    return message.getMessageId();
-  }
-
-  @Override
-  public List<String> getMessageHeader(String name) {
-    String header = MnsHelper.getMessageHeader(message, name);
-    if (header == null) {
-      return Collections.emptyList();
-    }
-    return Collections.singletonList(header);
-  }
-
-  public Message getMessage() {
-    return message;
-  }
-
   private MnsProcessRequest(Message message, @Nullable String destination) {
     this.message = message;
     this.destination = destination;

messaging-wrappers/aliyun-mns-sdk/src/test/java/io/opentelemetry/contrib/messaging/wrappers/mns/AliyunMnsSdkTest.java

@@ -6,9 +6,11 @@
 package io.opentelemetry.contrib.messaging.wrappers.mns;
 
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_BODY_SIZE;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ENVELOPE_SIZE;
+import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -138,7 +140,7 @@ public void assertTraces() {
                     // instrumentation library.
                     span.hasName("publish " + QUEUE).hasKind(SpanKind.PRODUCER).hasNoParent(),
                 span ->
-                    span.hasName("process " + QUEUE)
+                    span.hasName(QUEUE + " process")
                         .hasKind(SpanKind.CONSUMER)
                         .hasParent(trace.getSpan(0))
                         .hasAttributesSatisfyingExactly(
@@ -151,7 +153,10 @@ public void assertTraces() {
                                 MESSAGING_MESSAGE_ENVELOPE_SIZE,
                                 Base64.encodeBase64(MESSAGE_BODY.getBytes(StandardCharsets.UTF_8))
                                     .length),
-                            equalTo(MESSAGING_OPERATION, "process")),
+                            equalTo(MESSAGING_OPERATION, "process"),
+                            satisfies(
+                                MESSAGING_MESSAGE_ID,
+                                org.assertj.core.api.AbstractStringAssert::isNotEmpty)),
                 span ->
                     span.hasName("process child")
                         .hasKind(SpanKind.INTERNAL)