Instrument basic.get

Author: acogoluegnes

pom.xml

@@ -54,6 +54,7 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
+    <spotless.check.skip>true</spotless.check.skip>
     <slf4j.version>1.7.36</slf4j.version>
     <metrics.version>4.2.19</metrics.version>
     <micrometer.version>1.11.1</micrometer.version>
@@ -88,7 +89,8 @@
     <nexus-staging-maven-plugin.version>1.6.13</nexus-staging-maven-plugin.version>
     <checksum.maven.plugin.version>1.11</checksum.maven.plugin.version>
     <jshell-maven-plugin.version>1.3</jshell-maven-plugin.version>
-
+    <spotless.version>2.35.0</spotless.version>
+    <google-java-format.version>1.17.0</google-java-format.version>
     <!--
     These groovy scripts are used later in this POM file to generate
     source files and resources for the library itself and for the
@@ -1104,6 +1106,42 @@
         </configuration>
       </plugin>
 
+      <plugin>
+        <groupId>com.diffplug.spotless</groupId>
+        <artifactId>spotless-maven-plugin</artifactId>
+        <version>${spotless.version}</version>
+        <configuration>
+          <java>
+            <includes>
+              <include>src/main/java/com/rabbitmq/client/observation/**/*.java</include>
+              <include>src/test/java/com/rabbitmq/client/test/functional/MicrometerObservationCollectorMetrics.java</include>
+            </includes>
+            <googleJavaFormat>
+              <version>${google-java-format.version}</version>
+              <style>GOOGLE</style>
+            </googleJavaFormat>
+          </java>
+          <!--          <ratchetFrom>origin/main</ratchetFrom>-->
+          <licenseHeader> <!-- specify either content or file, but not both -->
+            <content>// Copyright (c) $YEAR VMware, Inc. or its affiliates.  All rights reserved.
+              //
+              // This software, the RabbitMQ Java client library, is triple-licensed under the
+              // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
+              // ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+              // LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+              // please see LICENSE-APACHE2.
+              //
+              // This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+              // either express or implied. See the LICENSE file for specific language governing
+              // rights and limitations of this software.
+              //
+              // If you have any questions regarding licensing, please contact us at
+              // [email protected].
+            </content>
+          </licenseHeader>
+        </configuration>
+      </plugin>
+
     </plugins>
     <extensions>
       <extension>

src/main/java/com/rabbitmq/client/impl/ChannelN.java

@@ -1164,26 +1164,28 @@ public GetResponse basicGet(String queue, boolean autoAck)
                                                   .queue(queue)
                                                   .noAck(autoAck)
                                                  .build());
-        Method method = replyCommand.getMethod();
-
-        if (method instanceof Basic.GetOk) {
-            Basic.GetOk getOk = (Basic.GetOk)method;
-            Envelope envelope = new Envelope(getOk.getDeliveryTag(),
-                                             getOk.getRedelivered(),
-                                             getOk.getExchange(),
-                                             getOk.getRoutingKey());
-            BasicProperties props = (BasicProperties)replyCommand.getContentHeader();
-            byte[] body = replyCommand.getContentBody();
-            int messageCount = getOk.getMessageCount();
-
-            metricsCollector.consumedMessage(this, getOk.getDeliveryTag(), autoAck);
-
-            return new GetResponse(envelope, props, body, messageCount);
-        } else if (method instanceof Basic.GetEmpty) {
-            return null;
-        } else {
-            throw new UnexpectedMethodError(method);
-        }
+        return this.observationCollector.basicGet(() -> {
+            Method method = replyCommand.getMethod();
+
+            if (method instanceof Basic.GetOk) {
+                Basic.GetOk getOk = (Basic.GetOk)method;
+                Envelope envelope = new Envelope(getOk.getDeliveryTag(),
+                    getOk.getRedelivered(),
+                    getOk.getExchange(),
+                    getOk.getRoutingKey());
+                BasicProperties props = (BasicProperties)replyCommand.getContentHeader();
+                byte[] body = replyCommand.getContentBody();
+                int messageCount = getOk.getMessageCount();
+
+                metricsCollector.consumedMessage(this, getOk.getDeliveryTag(), autoAck);
+
+                return new GetResponse(envelope, props, body, messageCount);
+            } else if (method instanceof Basic.GetEmpty) {
+                return null;
+            } else {
+                throw new UnexpectedMethodError(method);
+            }
+        }, queue);
     }
 
     /** Public API - {@inheritDoc} */

src/main/java/com/rabbitmq/client/observation/NoOpObservationCollector.java

@@ -12,11 +12,11 @@
 //
 // If you have any questions regarding licensing, please contact us at
 // [email protected].
-
 package com.rabbitmq.client.observation;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.GetResponse;
 import java.io.IOException;
 
 final class NoOpObservationCollector implements ObservationCollector {
@@ -36,4 +36,9 @@ public void publish(
   public Consumer basicConsume(String queue, String consumerTag, Consumer consumer) {
     return consumer;
   }
+
+  @Override
+  public GetResponse basicGet(BasicGetCall call, String queue) {
+    return call.get();
+  }
 }

src/main/java/com/rabbitmq/client/observation/ObservationCollector.java

@@ -12,20 +12,43 @@
 //
 // If you have any questions regarding licensing, please contact us at
 // [email protected].
-
 package com.rabbitmq.client.observation;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.GetResponse;
 import java.io.IOException;
 
 /**
+ * API to instrument operations in the AMQP client. The supported operations are publishing,
+ * asynchronous delivery, and synchronous delivery (<code>basic.get</code>).
+ *
+ * <p>Implementations can gather information and send it to tracing backends. This allows e.g.
+ * following the processing steps of a given message through different systems.
+ *
+ * <p>This is considered an SPI and is susceptible to change at any time.
+ *
  * @since 5.18.0
+ * @see com.rabbitmq.client.ConnectionFactory#setObservationCollector( ObservationCollector)
  */
 public interface ObservationCollector {
 
   ObservationCollector NO_OP = new NoOpObservationCollector();
 
+  /**
+   * Decorate message publishing.
+   *
+   * <p>Implementations are expected to call {@link PublishCall#publish( PublishCall,
+   * AMQP.Basic.Publish, AMQP.BasicProperties, byte[], ConnectionInfo)} to make sure the message is
+   * actually sent.
+   *
+   * @param call
+   * @param publish
+   * @param properties
+   * @param body
+   * @param connectionInfo
+   * @throws IOException
+   */
   void publish(
       PublishCall call,
       AMQP.Basic.Publish publish,
@@ -34,13 +57,44 @@ void publish(
       ConnectionInfo connectionInfo)
       throws IOException;
 
+  /**
+   * Decorate consumer registration.
+   *
+   * <p>Implementations are expected to decorate the appropriate {@link Consumer} callbacks. The
+   * original {@link Consumer} behavior should not be changed though.
+   *
+   * @param queue
+   * @param consumerTag
+   * @param consumer
+   * @return
+   */
   Consumer basicConsume(String queue, String consumerTag, Consumer consumer);
 
+  /**
+   * Decorate message polling with <code>basic.get</code>.
+   *
+   * <p>Implementations are expected to {@link BasicGetCall#basicGet( BasicGetCall, String)} and
+   * return the same result.
+   *
+   * @param call
+   * @param queue
+   * @return
+   */
+  GetResponse basicGet(BasicGetCall call, String queue);
+
+  /** Underlying publishing call. */
   interface PublishCall {
 
     void publish(AMQP.BasicProperties properties) throws IOException;
   }
 
+  /** Underlying <code>basic.get</code> call. */
+  interface BasicGetCall {
+
+    GetResponse get();
+  }
+
+  /** Connection information. */
   interface ConnectionInfo {
 
     String getPeerAddress();

…DefaultConsumeObservationConvention.java …DefaultDeliverObservationConvention.javasrc/main/java/com/rabbitmq/client/observation/micrometer/DefaultConsumeObservationConvention.java renamed to src/main/java/com/rabbitmq/client/observation/micrometer/DefaultDeliverObservationConvention.java src/main/java/com/rabbitmq/client/observation/micrometer/DefaultConsumeObservationConvention.java renamed to src/main/java/com/rabbitmq/client/observation/micrometer/DefaultDeliverObservationConvention.java

@@ -12,7 +12,6 @@
 //
 // If you have any questions regarding licensing, please contact us at
 // [email protected].
-
 package com.rabbitmq.client.observation.micrometer;
 
 import com.rabbitmq.client.observation.micrometer.RabbitMqObservationDocumentation.HighCardinalityTags;
@@ -21,21 +20,19 @@
 import io.micrometer.common.util.StringUtils;
 
 /**
- * Default implementation of {@link ConsumeObservationConvention}.
+ * Default implementation of {@link DeliverObservationConvention}.
  *
  * @since 5.18.0
- * @see ConsumeObservationConvention
+ * @see DeliverObservationConvention
  */
-public class DefaultConsumeObservationConvention implements ConsumeObservationConvention {
+public class DefaultDeliverObservationConvention implements DeliverObservationConvention {
 
   private final String name;
+  private final String operation;
 
-  public DefaultConsumeObservationConvention() {
-    this("rabbitmq.consume");
-  }
-
-  public DefaultConsumeObservationConvention(String name) {
+  public DefaultDeliverObservationConvention(String name, String operation) {
     this.name = name;
+    this.operation = operation;
   }
 
   @Override
@@ -44,8 +41,8 @@ public String getName() {
   }
 
   @Override
-  public String getContextualName(ConsumeContext context) {
-    return source(context.getQueue()) + " consume";
+  public String getContextualName(DeliverContext context) {
+    return source(context.getQueue()) + " " + operation;
   }
 
   private String exchange(String destination) {
@@ -57,14 +54,14 @@ private String source(String destination) {
   }
 
   @Override
-  public KeyValues getLowCardinalityKeyValues(ConsumeContext context) {
+  public KeyValues getLowCardinalityKeyValues(DeliverContext context) {
     return KeyValues.of(
-        LowCardinalityTags.MESSAGING_OPERATION.withValue("consume"),
+        LowCardinalityTags.MESSAGING_OPERATION.withValue(this.operation),
         LowCardinalityTags.MESSAGING_SYSTEM.withValue("rabbitmq"));
   }
 
   @Override
-  public KeyValues getHighCardinalityKeyValues(ConsumeContext context) {
+  public KeyValues getHighCardinalityKeyValues(DeliverContext context) {
     return KeyValues.of(
         HighCardinalityTags.MESSAGING_ROUTING_KEY.withValue(context.getRoutingKey()),
         HighCardinalityTags.MESSAGING_DESTINATION_NAME.withValue(exchange(context.getExchange())),

src/main/java/com/rabbitmq/client/observation/micrometer/DefaultPublishObservationConvention.java

@@ -12,7 +12,6 @@
 //
 // If you have any questions regarding licensing, please contact us at
 // [email protected].
-
 package com.rabbitmq.client.observation.micrometer;
 
 import com.rabbitmq.client.observation.micrometer.RabbitMqObservationDocumentation.HighCardinalityTags;
@@ -24,7 +23,6 @@
  * Default implementation of {@link PublishObservationConvention}.
  *
  * @since 5.18.0
- * @see RabbitMqObservationDocumentation
  */
 public class DefaultPublishObservationConvention implements PublishObservationConvention {
 

…servation/micrometer/ConsumeContext.java …servation/micrometer/DeliverContext.javasrc/main/java/com/rabbitmq/client/observation/micrometer/ConsumeContext.java renamed to src/main/java/com/rabbitmq/client/observation/micrometer/DeliverContext.java src/main/java/com/rabbitmq/client/observation/micrometer/ConsumeContext.java renamed to src/main/java/com/rabbitmq/client/observation/micrometer/DeliverContext.java

@@ -12,7 +12,6 @@
 //
 // If you have any questions regarding licensing, please contact us at
 // [email protected].
-
 package com.rabbitmq.client.observation.micrometer;
 
 import io.micrometer.observation.transport.ReceiverContext;
@@ -24,15 +23,19 @@
  *
  * @since 5.18.0
  */
-public class ConsumeContext extends ReceiverContext<Map<String, Object>> {
+public class DeliverContext extends ReceiverContext<Map<String, Object>> {
 
   private final String exchange;
   private final String routingKey;
   private final int payloadSizeBytes;
   private final String queue;
 
-  ConsumeContext(String exchange, String routingKey, String queue, Map<String, Object> headers,
-                 int payloadSizeBytes) {
+  DeliverContext(
+      String exchange,
+      String routingKey,
+      String queue,
+      Map<String, Object> headers,
+      int payloadSizeBytes) {
     super(
         (hdrs, key) -> {
           Object result = hdrs.get(key);
@@ -63,5 +66,4 @@ public int getPayloadSizeBytes() {
   public String getQueue() {
     return queue;
   }
-
 }

…ometer/ConsumeObservationConvention.java …ometer/DeliverObservationConvention.javasrc/main/java/com/rabbitmq/client/observation/micrometer/ConsumeObservationConvention.java renamed to src/main/java/com/rabbitmq/client/observation/micrometer/DeliverObservationConvention.java src/main/java/com/rabbitmq/client/observation/micrometer/ConsumeObservationConvention.java renamed to src/main/java/com/rabbitmq/client/observation/micrometer/DeliverObservationConvention.java

@@ -12,7 +12,6 @@
 //
 // If you have any questions regarding licensing, please contact us at
 // [email protected].
-
 package com.rabbitmq.client.observation.micrometer;
 
 import io.micrometer.observation.Observation;
@@ -22,12 +21,11 @@
  * {@link ObservationConvention} for RabbitMQ client instrumentation.
  *
  * @since 5.18.0
- * @see DefaultPublishObservationConvention
  */
-public interface ConsumeObservationConvention extends ObservationConvention<ConsumeContext> {
+public interface DeliverObservationConvention extends ObservationConvention<DeliverContext> {
 
   @Override
   default boolean supportsContext(Observation.Context context) {
-    return context instanceof ConsumeContext;
+    return context instanceof DeliverContext;
   }
 }