You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by de...@apache.org on 2009/12/29 11:46:48 UTC
svn commit: r894347 - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/ test/java/org/apache/activemq/broker/policy/
test/java/org/apache/activemq/transport/stomp/
Author: dejanb
Date: Tue Dec 29 10:46:47 2009
New Revision: 894347
URL: http://svn.apache.org/viewvc?rev=894347&view=rev
Log:
https://issues.apache.org/activemq/browse/AMQ-2553 - browsing dlq over transacted session
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQQueueBrowser.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTest.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTestSupport.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/IndividualDeadLetterTest.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQQueueBrowser.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQQueueBrowser.java?rev=894347&r1=894346&r2=894347&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQQueueBrowser.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQQueueBrowser.java Tue Dec 29 10:46:47 2009
@@ -113,6 +113,9 @@
return;
}
try {
+ if (session.getTransacted()) {
+ session.commit();
+ }
consumer.close();
consumer = null;
} catch (JMSException e) {
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTest.java?rev=894347&r1=894346&r2=894347&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTest.java Tue Dec 29 10:46:47 2009
@@ -33,7 +33,7 @@
public class DeadLetterTest extends DeadLetterTestSupport {
private static final Log LOG = LogFactory.getLog(DeadLetterTest.class);
- private int rollbackCount;
+ protected int rollbackCount;
protected void doTest() throws Exception {
connection.start();
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTestSupport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTestSupport.java?rev=894347&r1=894346&r2=894347&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTestSupport.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DeadLetterTestSupport.java Tue Dec 29 10:46:47 2009
@@ -23,6 +23,8 @@
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
@@ -51,6 +53,7 @@
protected boolean durableSubscriber;
protected Destination dlqDestination;
protected MessageConsumer dlqConsumer;
+ protected QueueBrowser dlqBrowser;
protected BrokerService broker;
protected boolean transactedMode;
protected int acknowledgeMode = Session.CLIENT_ACKNOWLEDGE;
@@ -108,6 +111,13 @@
LOG.info("Consuming from dead letter on: " + dlqDestination);
dlqConsumer = session.createConsumer(dlqDestination);
}
+
+ protected void makeDlqBrowser() throws JMSException {
+ dlqDestination = createDlqDestination();
+
+ LOG.info("Browsing dead letter on: " + dlqDestination);
+ dlqBrowser = session.createBrowser((Queue)dlqDestination);
+ }
protected void sendMessages() throws JMSException {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/IndividualDeadLetterTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/IndividualDeadLetterTest.java?rev=894347&r1=894346&r2=894347&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/IndividualDeadLetterTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/IndividualDeadLetterTest.java Tue Dec 29 10:46:47 2009
@@ -16,19 +16,29 @@
*/
package org.apache.activemq.broker.policy;
+import java.util.Enumeration;
+
+import javax.jms.DeliveryMode;
import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Queue;
+import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.policy.DeadLetterStrategy;
import org.apache.activemq.broker.region.policy.IndividualDeadLetterStrategy;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
/**
* @version $Revision$
*/
public class IndividualDeadLetterTest extends DeadLetterTest {
+ private static final Log LOG = LogFactory.getLog(IndividualDeadLetterTest.class);
protected BrokerService createBroker() throws Exception {
BrokerService broker = super.createBroker();
@@ -50,4 +60,48 @@
String prefix = topic ? "ActiveMQ.DLQ.Topic." : "ActiveMQ.DLQ.Queue.";
return new ActiveMQQueue(prefix + getClass().getName() + "." + getName());
}
+
+ public void testDLQBrowsing() throws Exception {
+ super.topic = false;
+ deliveryMode = DeliveryMode.PERSISTENT;
+ durableSubscriber = false;
+ messageCount = 1;
+
+ connection.start();
+
+ ActiveMQConnection amqConnection = (ActiveMQConnection) connection;
+ rollbackCount = amqConnection.getRedeliveryPolicy().getMaximumRedeliveries() + 1;
+ LOG.info("Will redeliver messages: " + rollbackCount + " times");
+
+ sendMessages();
+
+ // now lets receive and rollback N times
+ for (int i = 0; i < rollbackCount; i++) {
+ makeConsumer();
+ Message message = consumer.receive(5000);
+ assertNotNull("No message received: ", message);
+
+ session.rollback();
+ LOG.info("Rolled back: " + rollbackCount + " times");
+ consumer.close();
+ }
+
+ makeDlqBrowser();
+ browseDlq();
+ dlqBrowser.close();
+ session.close();
+ Thread.sleep(1000);
+ session = connection.createSession(transactedMode, acknowledgeMode);
+ Queue testQueue = new ActiveMQQueue("ActiveMQ.DLQ.Queue.ActiveMQ.DLQ.Queue." + getClass().getName() + "." + getName());
+ MessageConsumer testConsumer = session.createConsumer(testQueue);
+ assertNull("The message shouldn't be sent to another DLQ", testConsumer.receive(1000));
+
+ }
+
+ protected void browseDlq() throws Exception {
+ Enumeration messages = dlqBrowser.getEnumeration();
+ while (messages.hasMoreElements()) {
+ LOG.info("Browsing: " + messages.nextElement());
+ }
+ }
}
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java?rev=894347&r1=894346&r2=894347&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java Tue Dec 29 10:46:47 2009
@@ -29,8 +29,8 @@
final int msgCount = 10000;
final int producerCount = 5;
final int consumerCount = 5;
- final int testTime = 10 * 60 * 1000;
- final String bindAddress = "stomp://0.0.0.0:61613";
+ final int testTime = 30 * 60 * 1000;
+ final String bindAddress = "stomp://0.0.0.0:61612";
public void testLoad() throws Exception {