From 15186e80c64ad26b30b72ccf5d778591946a16a2 Mon Sep 17 00:00:00 2001 From: Ken Liao Date: Tue, 27 Aug 2024 02:22:58 -0700 Subject: [PATCH 1/2] Add attributes to display total number of queues and topics --- .../activemq/broker/jmx/BrokerView.java | 10 +++ .../activemq/broker/jmx/BrokerViewMBean.java | 8 +++ .../activemq/broker/jmx/BrokerViewTest.java | 63 +++++++++++++++++++ 3 files changed, 81 insertions(+) create mode 100644 activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/BrokerViewTest.java diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java index e8ec158dae0..97dab615933 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java @@ -280,11 +280,21 @@ public ObjectName[] getTopics() { return safeGetBroker().getTopicsNonSuppressed(); } + @Override + public int getTotalTopicsCount() { + return getTopics().length; + } + @Override public ObjectName[] getQueues() { return safeGetBroker().getQueuesNonSuppressed(); } + @Override + public int getTotalQueuesCount() { + return getQueues().length; + } + @Override public String queryQueues(String filter, int page, int pageSize) throws IOException { return DestinationsViewFilter.create(filter) diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerViewMBean.java index 7584a717349..08a6d19317c 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerViewMBean.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerViewMBean.java @@ -179,9 +179,17 @@ public interface BrokerViewMBean extends Service { @MBeanInfo("Topics (broadcasted 'queues'); generally system information.") ObjectName[] getTopics(); + @MBeanInfo("Total number of topics") + int getTotalTopicsCount(); + + @MBeanInfo("Standard Queues containing AIE messages.") ObjectName[] getQueues(); + @MBeanInfo("Total number of queues") + int getTotalQueuesCount(); + + /** * Queue Query API, take a look at {@link DestinationsViewFilter} for more information */ diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/BrokerViewTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/BrokerViewTest.java new file mode 100644 index 00000000000..5707b560900 --- /dev/null +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/BrokerViewTest.java @@ -0,0 +1,63 @@ +/** + * 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.broker.jmx; + +import jakarta.jms.*; +import jakarta.jms.Connection; +import org.apache.activemq.*; +import org.apache.activemq.broker.*; +import org.apache.activemq.util.*; +import org.junit.*; + +import static org.junit.Assert.assertTrue; + +public class BrokerViewTest { + protected BrokerService brokerService; + protected ActiveMQConnectionFactory factory; + protected Connection producerConnection; + + protected Session producerSession; + protected MessageConsumer consumer; + protected MessageProducer producer; + protected Queue queue; + + @Before + public void setUp() throws Exception { + brokerService = new BrokerService(); + brokerService.setPersistent(false); + brokerService.start(); + + factory = new ActiveMQConnectionFactory(BrokerRegistry.getInstance().findFirst().getVmConnectorURI()); + producerConnection = factory.createConnection(); + producerConnection.start(); + producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = producerSession.createQueue("testQueue"); + producer = producerSession.createProducer(queue); + producer.send(producerSession.createTextMessage("testMessage")); + } + + @Test(timeout=120000) + public void testBrokerViewRetrieveTotalQueuesAndTopicsCount() throws Exception { + assertTrue(Wait.waitFor(() -> (brokerService.getAdminView()) != null)); + + final BrokerView view = brokerService.getAdminView(); + // The total number of advisory topics + assert(view.getTotalTopicsCount() == 4); + // The queue created in setup + assert(view.getTotalQueuesCount() == 1); + } +} From 4faf9eac54e8024095a35d6647b49b79ed59efba Mon Sep 17 00:00:00 2001 From: Ken Liao Date: Fri, 25 Oct 2024 23:39:23 -0700 Subject: [PATCH 2/2] Addressed comments and add metrics for temporary queues and topics --- .../activemq/broker/jmx/BrokerView.java | 24 +++++++- .../activemq/broker/jmx/BrokerViewMBean.java | 10 +++ .../activemq/broker/jmx/BrokerViewTest.java | 61 ++++++++++--------- 3 files changed, 65 insertions(+), 30 deletions(-) diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java index 97dab615933..39fc8e1d98b 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java @@ -282,7 +282,17 @@ public ObjectName[] getTopics() { @Override public int getTotalTopicsCount() { - return getTopics().length; + return safeGetBroker().getTopicRegion().getDestinationMap().size(); + } + + @Override + public int getTotalNonSuppressedTopicsCount() { + return safeGetBroker().getTopicViews().size(); + } + + @Override + public int getTotalTemporaryTopicsCount() { + return safeGetBroker().getTempTopicRegion().getDestinationMap().size(); } @Override @@ -292,7 +302,17 @@ public ObjectName[] getQueues() { @Override public int getTotalQueuesCount() { - return getQueues().length; + return safeGetBroker().getQueueRegion().getDestinationMap().size(); + } + + @Override + public int getTotalNonSuppressedQueuesCount() { + return safeGetBroker().getQueueViews().size(); + } + + @Override + public int getTotalTemporaryQueuesCount() { + return safeGetBroker().getTempQueueRegion().getDestinationMap().size(); } @Override diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerViewMBean.java index 08a6d19317c..75a97a743f3 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerViewMBean.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerViewMBean.java @@ -182,6 +182,11 @@ public interface BrokerViewMBean extends Service { @MBeanInfo("Total number of topics") int getTotalTopicsCount(); + @MBeanInfo("Total number of non suppressed topics") + int getTotalNonSuppressedTopicsCount(); + + @MBeanInfo("Total number of temporary topics") + int getTotalTemporaryTopicsCount(); @MBeanInfo("Standard Queues containing AIE messages.") ObjectName[] getQueues(); @@ -189,6 +194,11 @@ public interface BrokerViewMBean extends Service { @MBeanInfo("Total number of queues") int getTotalQueuesCount(); + @MBeanInfo("Total number of non suppressed queues") + int getTotalNonSuppressedQueuesCount(); + + @MBeanInfo("Total number of temporary queues") + int getTotalTemporaryQueuesCount(); /** * Queue Query API, take a look at {@link DestinationsViewFilter} for more information diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/BrokerViewTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/BrokerViewTest.java index 5707b560900..87787f46171 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/BrokerViewTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/BrokerViewTest.java @@ -17,47 +17,52 @@ package org.apache.activemq.broker.jmx; import jakarta.jms.*; -import jakarta.jms.Connection; -import org.apache.activemq.*; -import org.apache.activemq.broker.*; -import org.apache.activemq.util.*; -import org.junit.*; +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerRegistry; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.util.Wait; +import org.junit.Test; import static org.junit.Assert.assertTrue; public class BrokerViewTest { - protected BrokerService brokerService; - protected ActiveMQConnectionFactory factory; - protected Connection producerConnection; - - protected Session producerSession; - protected MessageConsumer consumer; - protected MessageProducer producer; - protected Queue queue; - - @Before - public void setUp() throws Exception { - brokerService = new BrokerService(); + @Test(timeout=120000) + public void testBrokerViewRetrieveQueuesAndTopicsCount() throws Exception { + BrokerService brokerService = new BrokerService(); brokerService.setPersistent(false); brokerService.start(); - factory = new ActiveMQConnectionFactory(BrokerRegistry.getInstance().findFirst().getVmConnectorURI()); - producerConnection = factory.createConnection(); + ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BrokerRegistry.getInstance().findFirst().getVmConnectorURI()); + Connection producerConnection = factory.createConnection(); producerConnection.start(); - producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + // Create non-suppressed queue + Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue = producerSession.createQueue("testQueue"); - producer = producerSession.createProducer(queue); + MessageProducer producer = producerSession.createProducer(queue); producer.send(producerSession.createTextMessage("testMessage")); - } + // Create temporary queue + Session tempProducerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue tempQueue = tempProducerSession.createTemporaryQueue(); + MessageProducer tempProducer = tempProducerSession.createProducer(tempQueue); + tempProducer.send(tempProducerSession.createTextMessage("testMessage")); + // Create non-suppressed topic + Session topicProducerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Topic topic = topicProducerSession.createTopic("testTopic"); + MessageProducer topicProducer = topicProducerSession.createProducer(topic); + topicProducer.send(topicProducerSession.createTextMessage("testMessage")); + // Create temporary topic + Session tempTopicProducerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Topic tempTopic = tempTopicProducerSession.createTemporaryTopic(); + MessageProducer tempTopicProducer = tempTopicProducerSession.createProducer(tempTopic); + tempTopicProducer.send(tempTopicProducerSession.createTextMessage("testMessage")); - @Test(timeout=120000) - public void testBrokerViewRetrieveTotalQueuesAndTopicsCount() throws Exception { assertTrue(Wait.waitFor(() -> (brokerService.getAdminView()) != null)); - final BrokerView view = brokerService.getAdminView(); - // The total number of advisory topics - assert(view.getTotalTopicsCount() == 4); - // The queue created in setup + assert(view.getTotalTopicsCount() == 11); // Including advisory topics + assert(view.getTotalNonSuppressedTopicsCount() == 11); + assert(view.getTotalTemporaryTopicsCount() == 1); assert(view.getTotalQueuesCount() == 1); + assert(view.getTotalNonSuppressedQueuesCount() == 1); + assert(view.getTotalTemporaryQueuesCount() == 1); } }