activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject [1/2] activemq git commit: https://issues.apache.org/jira/browse/AMQ-5813
Date Fri, 05 Jun 2015 18:13:13 GMT
Repository: activemq
Updated Branches:
  refs/heads/master 11da37b99 -> 3649f13b8


https://issues.apache.org/jira/browse/AMQ-5813

Fixing countBeforeFull for TopicSubscriptions to report a positive value


Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/ea03bb1f
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/ea03bb1f
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/ea03bb1f

Branch: refs/heads/master
Commit: ea03bb1f8cb2bce97fabdaa8fa287f4c9f977a3a
Parents: 11da37b
Author: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Authored: Thu May 28 11:33:17 2015 +0000
Committer: Timothy Bish <tabish121@gmail.com>
Committed: Fri Jun 5 14:08:15 2015 -0400

----------------------------------------------------------------------
 .../broker/region/AbstractSubscription.java     |   2 +-
 .../broker/region/TopicSubscription.java        |   5 +
 .../TopicSubscriptionCountBeforeFullTest.java   | 117 +++++++++++++++++++
 3 files changed, 123 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/ea03bb1f/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
index 37056a2..92d1c0d 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
@@ -263,7 +263,7 @@ public abstract class AbstractSubscription implements Subscription {
 
     @Override
     public int countBeforeFull() {
-        return getDispatchedQueueSize() - info.getPrefetchSize();
+        return info.getPrefetchSize() - getDispatchedQueueSize();
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/activemq/blob/ea03bb1f/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
index ba755ee..c59c359 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
@@ -380,6 +380,11 @@ public class TopicSubscription extends AbstractSubscription {
     }
 
     @Override
+    public int countBeforeFull() {
+        return getPrefetchSize() == 0 ? prefetchExtension.get() : info.getPrefetchSize()
+ prefetchExtension.get() - getDispatchedQueueSize();
+    }
+
+    @Override
     public int getPendingQueueSize() {
         return matched();
     }

http://git-wip-us.apache.org/repos/asf/activemq/blob/ea03bb1f/activemq-unit-tests/src/test/java/org/apache/activemq/TopicSubscriptionCountBeforeFullTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/TopicSubscriptionCountBeforeFullTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/TopicSubscriptionCountBeforeFullTest.java
new file mode 100644
index 0000000..ca4a90c
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/TopicSubscriptionCountBeforeFullTest.java
@@ -0,0 +1,117 @@
+/**
+ * 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;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.broker.region.Destination;
+import org.apache.activemq.broker.region.Subscription;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.junit.Test;
+
+/**
+ * This test shows that the countBeforeFull statistic that is part of a Subscription is correct
+ * for TopicSubscriptions.
+ */
+public class TopicSubscriptionCountBeforeFullTest extends TestSupport {
+
+    protected BrokerService brokerService;
+    private Connection connection;
+    private String brokerUrlString;
+    private Session session;
+    private Topic topic;
+    private Destination amqDestination;
+    private int prefetch = 10;
+
+    protected void setUp() throws Exception {
+        super.setUp();
+        brokerService = new BrokerService();
+        brokerService.setDeleteAllMessagesOnStartup(true);
+        TransportConnector tcp = brokerService
+                .addConnector("tcp://localhost:0");
+        brokerService.start();
+        brokerUrlString = tcp.getPublishableConnectString();
+        connection = createConnection();
+        connection.start();
+        session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
+        topic = session.createTopic("test");
+        session.createConsumer(topic);
+        amqDestination = TestSupport.getDestination(brokerService,new ActiveMQTopic("test"));
+    }
+
+    @Override
+    protected ActiveMQConnectionFactory createConnectionFactory()
+            throws Exception {
+        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrlString);
+        ActiveMQPrefetchPolicy prefecthPolicy = new ActiveMQPrefetchPolicy();
+        prefecthPolicy.setTopicPrefetch(prefetch);
+        factory.setPrefetchPolicy(prefecthPolicy);
+        return factory;
+    }
+
+    /**
+     * @see junit.framework.TestCase#tearDown()
+     */
+    protected void tearDown() throws Exception {
+        if (connection != null) {
+            connection.close();
+            connection = null;
+        }
+        brokerService.stop();
+        super.tearDown();
+    }
+
+    /**
+     * Tests that countBeforeFull is 0 if prefetch is filled
+     *
+     * @throws javax.jms.JMSException
+     */
+    @Test
+    public void testCountBeforeFullPrefetchFilled() throws JMSException {
+        sendMessages(10);
+        assertEquals(getSubscription().countBeforeFull(), 0);
+    }
+
+    /**
+     * Tests that countBeforeFull is a positive number when no messages have been sent
+     * and prefetch is greater than 0
+     *
+     * @throws javax.jms.JMSException
+     */
+    @Test
+    public void testCountBeforeFullNotNull() throws JMSException {
+        assertTrue(getSubscription().countBeforeFull() == prefetch);
+    }
+
+    protected void sendMessages(int count) throws JMSException {
+        MessageProducer producer = session.createProducer(topic);
+        for (int i = 0; i < count; i++) {
+            producer.send(session.createTextMessage("testMessage"));
+        }
+    }
+
+    protected Subscription getSubscription() {
+        return amqDestination.getConsumers().get(0);
+    }
+
+}


Mime
View raw message