[activemq] branch main updated: [AMQ-8597] Show active consumers in web console

[jb...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jbonofre pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git


The following commit(s) were added to refs/heads/main by this push:
     new 4cedd8077 [AMQ-8597] Show active consumers in web console
     new e04faac37 Merge pull request #838 from lucastetreault/console-consumers
4cedd8077 is described below

commit 4cedd80774a91ac14f50ad20d4e32b49bc9f4563
Author: Lucas Tétreault <te...@amazon.com>
AuthorDate: Sat May 7 00:30:08 2022 -0700

    [AMQ-8597] Show active consumers in web console
---
 .../apache/activemq/web/BrokerFacadeSupport.java   | 11 +++--
 .../activemq/web/BrokerFacadeSupportTest.java      | 57 ++++++++++++++++++++++
 2 files changed, 64 insertions(+), 4 deletions(-)

diff --git a/activemq-web/src/main/java/org/apache/activemq/web/BrokerFacadeSupport.java b/activemq-web/src/main/java/org/apache/activemq/web/BrokerFacadeSupport.java
index 09e7c9f3d..5f8d7df61 100644
--- a/activemq-web/src/main/java/org/apache/activemq/web/BrokerFacadeSupport.java
+++ b/activemq-web/src/main/java/org/apache/activemq/web/BrokerFacadeSupport.java
@@ -240,8 +240,9 @@ public abstract class BrokerFacadeSupport implements BrokerFacade {
     public Collection<SubscriptionViewMBean> getQueueConsumers(String queueName) throws Exception {
         String brokerName = getBrokerName();
         queueName = StringUtils.replace(queueName, "\"", "_");
-        ObjectName query = new ObjectName("org.apache.activemq:type=Broker,brokerName=\"" + brokerName
-                + "\",destinationType=Queue,destinationName=\"" + queueName + "\",endpoint=Consumer,*");
+        queueName = StringUtils.replace(queueName, ":", "_");
+        ObjectName query = new ObjectName("org.apache.activemq:type=Broker,brokerName=" + brokerName
+                + ",destinationType=Queue,destinationName=" + queueName + ",endpoint=Consumer,*");
         Set<ObjectName> queryResult = queryNames(query, null);
         return getManagedObjects(queryResult.toArray(new ObjectName[queryResult.size()]), SubscriptionViewMBean.class);
     }
@@ -251,8 +252,9 @@ public abstract class BrokerFacadeSupport implements BrokerFacade {
     public Collection<ProducerViewMBean> getQueueProducers(String queueName) throws Exception {
         String brokerName = getBrokerName();
         queueName = StringUtils.replace(queueName, "\"", "_");
-        ObjectName query = new ObjectName("org.apache.activemq:type=Broker,brokerName=\"" + brokerName
-                + "\",destinationType=Queue,destinationName=\"" + queueName + "\",endpoint=Producer,*");
+        queueName = StringUtils.replace(queueName, ":", "_");
+        ObjectName query = new ObjectName("org.apache.activemq:type=Broker,brokerName=" + brokerName
+                + ",destinationType=Queue,destinationName=" + queueName + ",endpoint=Producer,*");
         Set<ObjectName> queryResult = queryNames(query, null);
         return getManagedObjects(queryResult.toArray(new ObjectName[queryResult.size()]), ProducerViewMBean.class);
     }
@@ -262,6 +264,7 @@ public abstract class BrokerFacadeSupport implements BrokerFacade {
     public Collection<ProducerViewMBean> getTopicProducers(String topicName) throws Exception {
         String brokerName = getBrokerName();
         topicName = StringUtils.replace(topicName, "\"", "_");
+        topicName = StringUtils.replace(topicName, ":", "_");
         ObjectName query = new ObjectName("org.apache.activemq:type=Broker,brokerName=" + brokerName
                 + ",destinationType=Topic,destinationName=" + topicName + ",endpoint=Producer,*");
         Set<ObjectName> queryResult = queryNames(query, null);
diff --git a/activemq-web/src/test/java/org/apache/activemq/web/BrokerFacadeSupportTest.java b/activemq-web/src/test/java/org/apache/activemq/web/BrokerFacadeSupportTest.java
new file mode 100644
index 000000000..f6c1acd71
--- /dev/null
+++ b/activemq-web/src/test/java/org/apache/activemq/web/BrokerFacadeSupportTest.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.web;
+
+import org.apache.activemq.broker.BrokerTestSupport;
+import org.apache.activemq.broker.StubConnection;
+import org.apache.activemq.broker.jmx.SubscriptionViewMBean;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ConnectionInfo;
+import org.apache.activemq.command.ConsumerInfo;
+import org.apache.activemq.command.SessionInfo;
+
+import java.util.Collection;
+
+public class BrokerFacadeSupportTest extends BrokerTestSupport {
+    public void testGetQueueConsumerWorksWithSimpleQueue() throws Exception {
+        ActiveMQDestination destination = new ActiveMQQueue("TEST");
+        setupConsumer(destination);
+        BrokerFacadeSupport brokerFacadeSupport = new LocalBrokerFacade(broker);
+        Collection<SubscriptionViewMBean> subscriptionViewMBeans = brokerFacadeSupport.getQueueConsumers("TEST");
+        assertEquals(1, subscriptionViewMBeans.size());
+    }
+
+    public void testGetQueueConsumerWorksWithQueueWithColon() throws Exception {
+        ActiveMQDestination destination = new ActiveMQQueue("TEST:TEST");
+        setupConsumer(destination);
+        BrokerFacadeSupport brokerFacadeSupport = new LocalBrokerFacade(broker);
+        Collection<SubscriptionViewMBean> subscriptionViewMBeans = brokerFacadeSupport.getQueueConsumers("TEST:TEST");
+        assertEquals(1, subscriptionViewMBeans.size());
+    }
+
+    private void setupConsumer(ActiveMQDestination destination) throws Exception {
+        StubConnection connection1 = createConnection();
+        ConnectionInfo connectionInfo1 = createConnectionInfo();
+        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
+        connection1.send(connectionInfo1);
+        connection1.send(sessionInfo1);
+        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
+        consumerInfo1.setPrefetchSize(1);
+        connection1.request(consumerInfo1);
+    }
+}