Add support for metrics collection

Add a MetricsCollector property to RMQConnectionFactory.

[#157312331]

Fixes #49

(cherry picked from commit 2195e91)

Conflicts:
	src/main/java/com/rabbitmq/jms/admin/RMQConnectionFactory.java
	src/test/java/com/rabbitmq/integration/tests/AbstractITQueue.java

Author: acogoluegnes

src/main/java/com/rabbitmq/jms/admin/RMQConnectionFactory.java

@@ -2,6 +2,8 @@
 package com.rabbitmq.jms.admin;
 
 import com.rabbitmq.client.Address;
+import com.rabbitmq.client.MetricsCollector;
+import com.rabbitmq.client.NoOpMetricsCollector;
 import com.rabbitmq.jms.client.AmqpPropertiesCustomiser;
 import com.rabbitmq.jms.client.ConnectionParams;
 import com.rabbitmq.jms.client.RMQConnection;
@@ -11,8 +13,21 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.jms.*;
-import javax.naming.*;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.QueueConnection;
+import javax.jms.QueueConnectionFactory;
+import javax.jms.TopicConnection;
+import javax.jms.TopicConnectionFactory;
+import javax.naming.NamingException;
+import javax.naming.RefAddr;
+import javax.naming.Reference;
+import javax.naming.Referenceable;
+import javax.naming.StringRefAddr;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLException;
 import java.io.IOException;
@@ -21,7 +36,9 @@
 import java.util.List;
 import java.util.concurrent.TimeoutException;
 
-import static com.rabbitmq.jms.util.UriCodec.*;
+import static com.rabbitmq.jms.util.UriCodec.encHost;
+import static com.rabbitmq.jms.util.UriCodec.encSegment;
+import static com.rabbitmq.jms.util.UriCodec.encUserinfo;
 
 /**
  * RabbitMQ Implementation of JMS {@link ConnectionFactory}
@@ -79,6 +96,13 @@ public class RMQConnectionFactory implements ConnectionFactory, Referenceable, S
      */
     private AmqpPropertiesCustomiser amqpPropertiesCustomiser;
 
+    /**
+     * Collector for AMQP-client metrics.
+     *
+     * @since 1.10.0
+     */
+    private MetricsCollector metricsCollector = new NoOpMetricsCollector();
+
     /** Default not to use ssl */
     private boolean ssl = false;
     private String tlsProtocol;
@@ -130,6 +154,8 @@ public Connection createConnection(String username, String password) throws JMSE
         com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();
         setRabbitUri(logger, this, factory, this.getUri());
         maybeEnableTLS(factory);
+
+        factory.setMetricsCollector(this.metricsCollector);
         com.rabbitmq.client.Connection rabbitConnection = instantiateNodeConnection(factory);
 
         RMQConnection conn = new RMQConnection(new ConnectionParams()
@@ -706,5 +732,15 @@ public void setAmqpPropertiesCustomiser(AmqpPropertiesCustomiser amqpPropertiesC
         this.amqpPropertiesCustomiser = amqpPropertiesCustomiser;
     }
 
+    /**
+     * Set the collector for AMQP-client metrics.
+     *
+     * @param metricsCollector
+     * @since 1.10.0
+     * @see com.rabbitmq.client.ConnectionFactory#setMetricsCollector(MetricsCollector)
+     */
+    public void setMetricsCollector(MetricsCollector metricsCollector) {
+        this.metricsCollector = metricsCollector;
+    }
 }
 

src/test/java/com/rabbitmq/integration/tests/AbstractITQueue.java

@@ -12,6 +12,7 @@
 
 import org.junit.After;
 import org.junit.Before;
+import com.rabbitmq.jms.admin.RMQConnectionFactory;
 
 public abstract class AbstractITQueue {
     QueueConnectionFactory connFactory;
@@ -21,9 +22,14 @@ public abstract class AbstractITQueue {
     public void beforeTests() throws Exception {
         this.connFactory = (QueueConnectionFactory) AbstractTestConnectionFactory.getTestConnectionFactory()
                 .getConnectionFactory();
+        customise((RMQConnectionFactory) connFactory);
         this.queueConn = connFactory.createQueueConnection();
     }
 
+    protected void customise(RMQConnectionFactory connectionFactory) {
+
+    }
+
     protected static void drainQueue(QueueSession session, Queue queue) throws Exception {
         QueueReceiver receiver = session.createReceiver(queue);
         int n=0;

src/test/java/com/rabbitmq/integration/tests/MetricsCollectionIT.java

@@ -0,0 +1,117 @@
+/* Copyright (c) 2018 Pivotal Software, Inc. All rights reserved. */
+package com.rabbitmq.integration.tests;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.impl.AbstractMetricsCollector;
+import com.rabbitmq.jms.admin.RMQConnectionFactory;
+import org.junit.Test;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Queue;
+import javax.jms.QueueReceiver;
+import javax.jms.QueueSender;
+import javax.jms.QueueSession;
+import javax.jms.Session;
+import java.io.Serializable;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Integration test to check metrics collection is properly set up.
+ */
+public class MetricsCollectionIT extends AbstractITQueue {
+
+    private static final String QUEUE_NAME = "test.queue." + MetricsCollectionIT.class.getCanonicalName();
+    private static final long TEST_RECEIVE_TIMEOUT = 1000; // one second
+
+    TestMetricsCollector metricsCollector = new TestMetricsCollector();
+
+    @Override
+    protected void customise(RMQConnectionFactory connectionFactory) {
+        connectionFactory.setMetricsCollector(metricsCollector);
+    }
+
+    @Test
+    public void testSendAndReceiveLongTextMessage() throws Exception {
+        assertEquals(1, metricsCollector.connectionCount);
+        assertEquals(0, metricsCollector.channelCount);
+        assertEquals(0, metricsCollector.publishedMessageCount);
+        assertEquals(0, metricsCollector.consumedMessageCount);
+        sendAndConsumeMessage();
+        assertEquals(1, metricsCollector.channelCount);
+        assertEquals(1, metricsCollector.publishedMessageCount);
+        assertEquals(1, metricsCollector.consumedMessageCount);
+    }
+
+    protected void sendAndConsumeMessage() throws Exception {
+        MessageTestType mtt = MessageTestType.TEXT;
+        try {
+            queueConn.start();
+            QueueSession queueSession = queueConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+            Queue queue = queueSession.createQueue(QUEUE_NAME);
+
+            drainQueue(queueSession, queue);
+
+            QueueSender queueSender = queueSession.createSender(queue);
+            queueSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+            queueSender.send(mtt.gen(queueSession, (Serializable) queue));
+        } finally {
+            reconnect();
+        }
+
+        queueConn.start();
+        QueueSession queueSession = queueConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+        Queue queue = queueSession.createQueue(QUEUE_NAME);
+        QueueReceiver queueReceiver = queueSession.createReceiver(queue);
+        mtt.check(queueReceiver.receive(TEST_RECEIVE_TIMEOUT), (Serializable) queue);
+    }
+
+    static class TestMetricsCollector extends AbstractMetricsCollector {
+
+        volatile int connectionCount = 0;
+        volatile int channelCount = 0;
+        volatile int publishedMessageCount = 0;
+        volatile int consumedMessageCount = 0;
+
+        @Override
+        protected void incrementConnectionCount(Connection connection) {
+            connectionCount++;
+        }
+
+        @Override
+        protected void decrementConnectionCount(Connection connection) {
+            connectionCount--;
+        }
+
+        @Override
+        protected void incrementChannelCount(Channel channel) {
+            channelCount++;
+        }
+
+        @Override
+        protected void decrementChannelCount(Channel channel) {
+            channelCount--;
+        }
+
+        @Override
+        protected void markPublishedMessage() {
+            publishedMessageCount++;
+        }
+
+        @Override
+        protected void markConsumedMessage() {
+            consumedMessageCount++;
+        }
+
+        @Override
+        protected void markAcknowledgedMessage() {
+
+        }
+
+        @Override
+        protected void markRejectedMessage() {
+
+        }
+    }
+}