svn commit: r1479925 - in /activemq/trunk: activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java activemq-unit-tests/src/test/java/org/apache/activemq/bugs/MessageExpirationReaperTest.java

[ce...@apache.org]

Author: ceposta
Date: Tue May  7 14:39:25 2013
New Revision: 1479925

URL: http://svn.apache.org/r1479925
Log:
Fix for https://issues.apache.org/jira/browse/AMQ-4516 - Don't decrement topic enqueues when messages are expired
Removed the call to decrement the counters, and added unit test

Modified:
    activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java
    activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/MessageExpirationReaperTest.java

Modified: activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java?rev=1479925&r1=1479924&r2=1479925&view=diff
==============================================================================
--- activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java (original)
+++ activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java Tue May  7 14:39:25 2013
@@ -731,7 +731,6 @@ public class Topic extends BaseDestinati
         // AMQ-2586: Better to leave this stat at zero than to give the user
         // misleading metrics.
         // destinationStatistics.getMessages().decrement();
-        destinationStatistics.getEnqueues().decrement();
         destinationStatistics.getExpired().increment();
         MessageAck ack = new MessageAck();
         ack.setAckType(MessageAck.STANDARD_ACK_TYPE);

Modified: activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/MessageExpirationReaperTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/MessageExpirationReaperTest.java?rev=1479925&r1=1479924&r2=1479925&view=diff
==============================================================================
--- activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/MessageExpirationReaperTest.java (original)
+++ activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/MessageExpirationReaperTest.java Tue May  7 14:39:25 2013
@@ -19,12 +19,7 @@ package org.apache.activemq.bugs;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 
-import javax.jms.ConnectionFactory;
-import javax.jms.MessageProducer;
-import javax.jms.Queue;
-import javax.jms.QueueBrowser;
-import javax.jms.Session;
-import javax.jms.TextMessage;
+import javax.jms.*;
 import javax.management.ObjectName;
 
 import org.apache.activemq.ActiveMQConnection;
@@ -34,6 +29,7 @@ import org.apache.activemq.broker.jmx.De
 import org.apache.activemq.broker.region.policy.PolicyEntry;
 import org.apache.activemq.broker.region.policy.PolicyMap;
 import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQTopic;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -60,6 +56,7 @@ public class MessageExpirationReaperTest
 
         factory = createConnectionFactory();
         connection = (ActiveMQConnection) factory.createConnection();
+        connection.setClientID("test-connection");
         connection.start();
     }
 
@@ -135,6 +132,45 @@ public class MessageExpirationReaperTest
         assertEquals("Wrong inFlightCount: " + view.getInFlightCount(), 0, view.getInFlightCount());
     }
 
+    @Test
+    public void testExpiredMessagesOnTopic() throws Exception{
+        Session session = createSession();
+
+        // use a zero prefetch so messages don't go inflight
+        ActiveMQTopic destination = new ActiveMQTopic(destinationName + "?consumer.prefetchSize=0");
+
+        MessageProducer producer = session.createProducer(destination);
+
+        // should have a durable sub because it's a little tricky to get messages to expire in
+        // non-durable subs.. with durable subs, we can just expire in the topic using the expire
+        // period.. also.. durable sub has to be "inactive" for the expire checker to actually
+        // expire the messages
+        MessageConsumer consumer = session.createDurableSubscriber(destination, "test-durable");
+
+        producer.setTimeToLive(500);
+
+        final int count = 3;
+        // Send some messages with an expiration
+        for (int i = 0; i < count; i++) {
+            TextMessage message = session.createTextMessage("" + i);
+            producer.send(message);
+        }
+
+        DestinationViewMBean view = createView(destination);
+        // not expired yet...
+        assertEquals("Incorrect enqueue count", 3, view.getEnqueueCount() );
+
+        // close consumer so topic thinks consumer is inactive
+        consumer.close();
+
+        // Let the messages reach an expiry time
+        Thread.sleep(2000);
+
+        assertEquals("Incorrect inflight count: " + view.getInFlightCount(), 0, view.getInFlightCount());
+        assertEquals("Incorrect queue size count", 0, view.getQueueSize());
+        assertEquals("Incorrect expired size count", view.getEnqueueCount(), view.getExpiredCount());
+    }
+
     protected DestinationViewMBean createView(ActiveMQDestination destination) throws Exception {
         String domain = "org.apache.activemq";
         ObjectName name;