activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jstrac...@apache.org
Subject svn commit: r689023 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/broker/jmx/QueueView.java test/java/org/apache/activemq/broker/jmx/MBeanTest.java
Date Tue, 26 Aug 2008 10:33:33 GMT
Author: jstrachan
Date: Tue Aug 26 03:33:32 2008
New Revision: 689023

URL: http://svn.apache.org/viewvc?rev=689023&view=rev
Log:
added a test case for AMQ-1909

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/QueueView.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/QueueView.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/QueueView.java?rev=689023&r1=689022&r2=689023&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/QueueView.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/QueueView.java
Tue Aug 26 03:33:32 2008
@@ -18,6 +18,7 @@
 
 import javax.management.openmbean.CompositeData;
 import javax.management.openmbean.OpenDataException;
+import javax.jms.JMSException;
 
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.Queue;
@@ -104,8 +105,13 @@
                 ConnectionContext context = BrokerView.getConnectionContext(broker.getContextBroker());
                 return queue.moveMessageTo(context, messageId, originalDestination);
             }
+            else {
+                throw new JMSException("No original destination for message: "+ messageId);
+            }
+        }
+        else {
+            throw new JMSException("Could not find message: "+ messageId);
         }
-        return false;
     }
     
     public int cursorSize() {

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java?rev=689023&r1=689022&r2=689023&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java
Tue Aug 26 03:33:32 2008
@@ -24,6 +24,7 @@
 import javax.jms.Message;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
+import javax.jms.MessageConsumer;
 import javax.management.MBeanServer;
 import javax.management.MBeanServerInvocationHandler;
 import javax.management.MalformedObjectNameException;
@@ -32,9 +33,11 @@
 import javax.management.openmbean.TabularData;
 import junit.textui.TestRunner;
 import org.apache.activemq.EmbeddedBrokerTestSupport;
+import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.region.BaseDestination;
+import org.apache.activemq.broker.region.policy.SharedDeadLetterStrategy;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -81,6 +84,142 @@
         assertConsumerCounts();
     }
 
+    public void testMoveMessages() throws Exception {
+        connection = connectionFactory.createConnection();
+        useConnection(connection);
+
+        ObjectName queueViewMBeanName = assertRegisteredObjectName(domain + ":Type=Queue,Destination="
+ getDestinationString() + ",BrokerName=localhost");
+
+        QueueViewMBean queue = (QueueViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer,
queueViewMBeanName, QueueViewMBean.class, true);
+
+        CompositeData[] compdatalist = queue.browse();
+        int initialQueueSize = compdatalist.length;
+        if (initialQueueSize == 0) {
+            fail("There is no message in the queue:");
+        }
+        else {
+            echo("Current queue size: " + initialQueueSize);
+        }
+        // TODO uncommenting this line causes a hang!
+        //int messageCount = initialQueueSize;
+        int messageCount = 10;
+        String[] messageIDs = new String[messageCount];
+        for (int i = 0; i < messageCount; i++) {
+            CompositeData cdata = compdatalist[i];
+            String messageID = (String) cdata.get("JMSMessageID");
+            assertNotNull("Should have a message ID for message " + i, messageID);
+            messageIDs[i] = messageID;
+        }
+
+
+        echo("About to move " + messageCount + " messages");
+
+        String newDestination = getSecondDestinationString();
+        for (String messageID : messageIDs) {
+            echo("Moving message: " + messageID);
+            queue.moveMessageTo(messageID, newDestination);
+        }
+
+        echo("Now browsing the queue");
+        compdatalist = queue.browse();
+        int actualCount = compdatalist.length;
+        echo("Current queue size: " + actualCount);
+        // TODO we seem to have browsed the queue and now there are messages missing!
+        //assertEquals("Should now have empty queue but was", initialQueueSize - messageCount,
actualCount);
+
+        echo("Now browsing the second queue");
+
+        queueViewMBeanName = assertRegisteredObjectName(domain + ":Type=Queue,Destination="
+ newDestination + ",BrokerName=localhost");
+        queue = (QueueViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer,
queueViewMBeanName, QueueViewMBean.class, true);
+
+        long newQueuesize = queue.getQueueSize();
+        echo("Second queue size: " + newQueuesize);
+        assertEquals("Unexpected number of messages ",messageCount, newQueuesize);
+    }
+
+    public void TODO_testRetryMessages() throws Exception {
+        // lets speed up redelivery
+        ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory) connectionFactory;
+        factory.getRedeliveryPolicy().setCollisionAvoidancePercent((short) 0);
+        factory.getRedeliveryPolicy().setMaximumRedeliveries(1);
+        factory.getRedeliveryPolicy().setInitialRedeliveryDelay(0);
+        factory.getRedeliveryPolicy().setUseCollisionAvoidance(false);
+        factory.getRedeliveryPolicy().setUseExponentialBackOff(false);
+        factory.getRedeliveryPolicy().setBackOffMultiplier((short) 0);
+
+        connection = connectionFactory.createConnection();
+        useConnection(connection);
+
+
+        ObjectName queueViewMBeanName = assertRegisteredObjectName(domain + ":Type=Queue,Destination="
+ getDestinationString() + ",BrokerName=localhost");
+        QueueViewMBean queue = (QueueViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer,
queueViewMBeanName, QueueViewMBean.class, true);
+
+        long initialQueueSize = queue.getQueueSize();
+        echo("current queue size: " + initialQueueSize);
+
+
+        // lets create a duff consumer which keeps rolling back...
+        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        MessageConsumer consumer = session.createConsumer(new ActiveMQQueue(getDestinationString()));
+        Message message = consumer.receive(5000);
+        while (message != null) {
+            echo("Message: " + message.getJMSMessageID() + " redelivered " + message.getJMSRedelivered()
+ " counter " + message.getObjectProperty("JMSXDeliveryCount"));
+            session.rollback();
+            message = consumer.receive(2000);
+        }
+        consumer.close();
+        session.close();
+
+
+        // now lets get the dead letter queue
+        Thread.sleep(1000);
+
+        ObjectName dlqQueueViewMBeanName = assertRegisteredObjectName(domain + ":Type=Queue,Destination="
+ SharedDeadLetterStrategy.DEFAULT_DEAD_LETTER_QUEUE_NAME + ",BrokerName=localhost");
+        QueueViewMBean dlq = (QueueViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer,
dlqQueueViewMBeanName, QueueViewMBean.class, true);
+
+        long initialDlqSize = dlq.getQueueSize();
+        CompositeData[] compdatalist = dlq.browse();
+        int dlqQueueSize = compdatalist.length;
+        if (dlqQueueSize == 0) {
+            fail("There are no messages in the queue:");
+        }
+        else {
+            echo("Current DLQ queue size: " + dlqQueueSize);
+        }
+
+        // TODO uncommenting this line causes a hang!
+        //int messageCount = dlqQueueSize;
+        int messageCount = 10;
+        String[] messageIDs = new String[messageCount];
+        for (int i = 0; i < messageCount; i++) {
+            CompositeData cdata = compdatalist[i];
+            String messageID = (String) cdata.get("JMSMessageID");
+            assertNotNull("Should have a message ID for message " + i, messageID);
+            messageIDs[i] = messageID;
+        }
+
+
+        echo("About to retry " + messageCount + " messages");
+
+        for (String messageID : messageIDs) {
+            echo("Retrying message: " + messageID);
+            dlq.retryMessage(messageID);
+        }
+
+        long queueSize = queue.getQueueSize();
+        compdatalist = queue.browse();
+        int actualCount = compdatalist.length;
+        echo("Orginal queue size is now " + queueSize);
+        echo("Original browse queue size: " + actualCount);
+
+        long dlqSize = dlq.getQueueSize();
+        echo("DLQ size: " + dlqSize);
+
+        assertEquals("DLQ size", initialDlqSize - messageCount, dlqSize);
+        assertEquals("queue size", initialQueueSize, queueSize);
+        assertEquals("browse queue size", initialQueueSize, actualCount);
+    }
+
     public void testMoveMessagesBySelector() throws Exception {
         connection = connectionFactory.createConnection();
         useConnection(connection);
@@ -89,7 +228,7 @@
 
         QueueViewMBean queue = (QueueViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer,
queueViewMBeanName, QueueViewMBean.class, true);
 
-        String newDestination = "test.new.destination." + getClass() + "." + getName();
+        String newDestination = getSecondDestinationString();
         queue.moveMatchingMessagesTo("counter > 2", newDestination);
 
         queueViewMBeanName = assertRegisteredObjectName(domain + ":Type=Queue,Destination="
+ newDestination + ",BrokerName=localhost");
@@ -112,7 +251,7 @@
 
         QueueViewMBean queue = (QueueViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer,
queueViewMBeanName, QueueViewMBean.class, true);
 
-        String newDestination = "test.new.destination." + getClass() + "." + getName();
+        String newDestination = getSecondDestinationString();
         long queueSize = queue.getQueueSize();
         queue.copyMatchingMessagesTo("counter > 2", newDestination);
 
@@ -375,10 +514,10 @@
 
     protected BrokerService createBroker() throws Exception {
         BrokerService answer = new BrokerService();
+        answer.setPersistent(false);
         answer.setDeleteAllMessagesOnStartup(true);
         answer.setUseJmx(true);
         //answer.setEnableStatistics(true);
-        answer.setPersistent(false);
         answer.addConnector(bindAddress);
         return answer;
     }
@@ -404,4 +543,9 @@
     protected void echo(String text) {
         LOG.info(text);
     }
+
+
+    protected String getSecondDestinationString() {
+        return "test.new.destination." + getClass() + "." + getName();
+    }
 }



Mime
View raw message