You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2011/10/31 19:54:19 UTC
svn commit: r1195615 - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/ test/java/org/apache/activemq/bugs/
test/java/org/apache/activemq/usecases/
Author: tabish
Date: Mon Oct 31 18:54:19 2011
New Revision: 1195615
URL: http://svn.apache.org/viewvc?rev=1195615&view=rev
Log:
Fix for: https://issues.apache.org/jira/browse/AMQ-1853
Tests included.
Added:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ1853Test.java (with props)
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/NonBlockingConsumerRedeliveryTest.java (with props)
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java?rev=1195615&r1=1195614&r2=1195615&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java Mon Oct 31 18:54:19 2011
@@ -191,8 +191,9 @@ public class ActiveMQConnection implemen
protected volatile CountDownLatch transportInterruptionProcessingComplete;
private long consumerFailoverRedeliveryWaitPeriod;
private final Scheduler scheduler;
- private boolean messagePrioritySupported=true;
+ private boolean messagePrioritySupported = true;
private boolean transactedIndividualAck = false;
+ private boolean nonBlockingRedelivery = false;
/**
* Construct an <code>ActiveMQConnection</code>
@@ -2417,6 +2418,14 @@ public class ActiveMQConnection implemen
this.transactedIndividualAck = transactedIndividualAck;
}
+ public boolean isNonBlockingRedelivery() {
+ return nonBlockingRedelivery;
+ }
+
+ public void setNonBlockingRedelivery(boolean nonBlockingRedelivery) {
+ this.nonBlockingRedelivery = nonBlockingRedelivery;
+ }
+
/**
* Removes any TempDestinations that this connection has cached, ignoring
* any exceptions generated because the destination is in use as they should
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java?rev=1195615&r1=1195614&r2=1195615&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java Mon Oct 31 18:54:19 2011
@@ -120,6 +120,7 @@ public class ActiveMQConnectionFactory e
private ClientInternalExceptionListener clientInternalExceptionListener;
private boolean messagePrioritySupported = true;
private boolean transactedIndividualAck = false;
+ private boolean nonBlockingRedelivery = false;
// /////////////////////////////////////////////
//
@@ -327,6 +328,7 @@ public class ActiveMQConnectionFactory e
connection.setCheckForDuplicates(isCheckForDuplicates());
connection.setMessagePrioritySupported(isMessagePrioritySupported());
connection.setTransactedIndividualAck(isTransactedIndividualAck());
+ connection.setNonBlockingRedelivery(isNonBlockingRedelivery());
if (transportListener != null) {
connection.addTransportListener(transportListener);
}
@@ -731,7 +733,7 @@ public class ActiveMQConnectionFactory e
props.setProperty("checkForDuplicates", Boolean.toString(isCheckForDuplicates()));
props.setProperty("messagePrioritySupported", Boolean.toString(isMessagePrioritySupported()));
props.setProperty("transactedIndividualAck", Boolean.toString(isTransactedIndividualAck()));
-
+ props.setProperty("nonBlockingRedelivery", Boolean.toString(isNonBlockingRedelivery()));
}
public boolean isUseCompression() {
@@ -1058,4 +1060,18 @@ public class ActiveMQConnectionFactory e
this.transactedIndividualAck = transactedIndividualAck;
}
+
+ public boolean isNonBlockingRedelivery() {
+ return nonBlockingRedelivery;
+ }
+
+ /**
+ * When true a MessageConsumer will not stop Message delivery before re-delivering Messages
+ * from a rolled back transaction. This implies that message order will not be preserved and
+ * also will result in the TransactedIndividualAck option to be enabled.
+ */
+ public void setNonBlockingRedelivery(boolean nonBlockingRedelivery) {
+ this.nonBlockingRedelivery = nonBlockingRedelivery;
+ }
+
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java?rev=1195615&r1=1195614&r2=1195615&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java Mon Oct 31 18:54:19 2011
@@ -153,6 +153,7 @@ public class ActiveMQMessageConsumer imp
private long optimizeAcknowledgeTimeOut = 0;
private long failoverRedeliveryWaitPeriod = 0;
private boolean transactedIndividualAck = false;
+ private boolean nonBlockingRedelivery = false;
/**
* Create a MessageConsumer
@@ -260,7 +261,8 @@ public class ActiveMQMessageConsumer imp
}
this.info.setOptimizedAcknowledge(this.optimizeAcknowledge);
this.failoverRedeliveryWaitPeriod = session.connection.getConsumerFailoverRedeliveryWaitPeriod();
- this.transactedIndividualAck = session.connection.isTransactedIndividualAck();
+ this.nonBlockingRedelivery = session.connection.isNonBlockingRedelivery();
+ this.transactedIndividualAck = session.connection.isTransactedIndividualAck() || this.nonBlockingRedelivery;
if (messageListener != null) {
setMessageListener(messageListener);
}
@@ -579,7 +581,6 @@ public class ActiveMQMessageConsumer imp
checkMessageListener();
if (timeout == 0) {
return this.receive();
-
}
sendPullCommand(timeout);
@@ -1184,30 +1185,52 @@ public class ActiveMQMessageConsumer imp
}
// stop the delivery of messages.
- unconsumedMessages.stop();
+ if (nonBlockingRedelivery) {
+ if (!unconsumedMessages.isClosed()) {
- for (Iterator<MessageDispatch> iter = deliveredMessages.iterator(); iter.hasNext();) {
- MessageDispatch md = iter.next();
- unconsumedMessages.enqueueFirst(md);
- }
+ final LinkedList<MessageDispatch> pendingRedeliveries =
+ new LinkedList<MessageDispatch>(deliveredMessages);
- if (redeliveryDelay > 0 && !unconsumedMessages.isClosed()) {
- // Start up the delivery again a little later.
- scheduler.executeAfterDelay(new Runnable() {
- public void run() {
- try {
- if (started.get()) {
- start();
+ // Start up the delivery again a little later.
+ scheduler.executeAfterDelay(new Runnable() {
+ public void run() {
+ try {
+ if (!unconsumedMessages.isClosed()) {
+ for(MessageDispatch dispatch : pendingRedeliveries) {
+ session.dispatch(dispatch);
+ }
+ }
+ } catch (Exception e) {
+ session.connection.onAsyncException(e);
}
- } catch (JMSException e) {
- session.connection.onAsyncException(e);
}
- }
- }, redeliveryDelay);
+ }, redeliveryDelay);
+ }
+
} else {
- start();
- }
+ unconsumedMessages.stop();
+
+ for (MessageDispatch md : deliveredMessages) {
+ unconsumedMessages.enqueueFirst(md);
+ }
+ if (redeliveryDelay > 0 && !unconsumedMessages.isClosed()) {
+ // Start up the delivery again a little later.
+ scheduler.executeAfterDelay(new Runnable() {
+ public void run() {
+ try {
+ if (started.get()) {
+ start();
+ }
+ } catch (JMSException e) {
+ session.connection.onAsyncException(e);
+ }
+ }
+ }, redeliveryDelay);
+ } else {
+ start();
+ }
+ }
}
deliveredCounter -= deliveredMessages.size();
deliveredMessages.clear();
@@ -1248,6 +1271,7 @@ public class ActiveMQMessageConsumer imp
}
}
}
+
/*
* called with deliveredMessages locked
*/
Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ1853Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ1853Test.java?rev=1195615&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ1853Test.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ1853Test.java Mon Oct 31 18:54:19 2011
@@ -0,0 +1,351 @@
+/**
+ * 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.bugs;
+
+import static org.junit.Assert.*;
+
+import java.net.URI;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQMessageProducer;
+import org.apache.activemq.ActiveMQSession;
+import org.apache.activemq.RedeliveryPolicy;
+import org.apache.activemq.broker.BrokerFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test validates that the AMQ consumer blocks on redelivery of a message,
+ * through all redeliveries, until the message is either successfully consumed
+ * or sent to the DLQ.
+ */
+public class AMQ1853Test {
+ private static BrokerService broker;
+
+ private static final Logger LOG = LoggerFactory.getLogger(AMQ1853Test.class);
+ static final String jmsConnectionURI = "failover:(vm://localhost)";
+
+ // Virtual Topic that the test publishes 10 messages to
+ private static final String queueFail = "Queue.BlockingConsumer.QueueFail";
+
+ // Number of messages
+
+ private final int producerMessages = 5;
+ private final int totalNumberMessages = producerMessages * 2;
+ private final int maxRedeliveries = 2;
+ private final int redeliveryDelay = 1000;
+
+ private Map<String, AtomicInteger> messageList = null;
+
+ @Before
+ public void setUp() throws Exception {
+ broker = BrokerFactory.createBroker(new URI("broker:()/localhost?persistent=false"));
+ broker.setUseJmx(false);
+ broker.start();
+ broker.waitUntilStarted();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (broker != null) {
+ broker.stop();
+ broker.waitUntilStopped();
+ broker = null;
+ }
+ }
+
+ @Test
+ public void testConsumerMessagesAreNotOrdered() throws Exception {
+
+ TestConsumer consumerAllFail = null;
+ messageList = new Hashtable<String, AtomicInteger>();
+
+ try {
+
+ // The first 2 consumers will rollback, ultimately causing messages to land on the DLQ
+
+ TestProducer producerAllFail = new TestProducer(queueFail);
+ thread(producerAllFail, false);
+
+ consumerAllFail = new TestConsumer(queueFail, true);
+ thread(consumerAllFail, false);
+
+ // Give the consumers a second to start
+ Thread.sleep(1000);
+
+ thread(producerAllFail, false);
+
+ // Give the consumers a second to start
+ Thread.sleep(1000);
+
+ producerAllFail.getLatch().await();
+
+ LOG.info("producer successful, count = " + producerAllFail.getLatch().getCount());
+
+ assertTrue("message list size = " + messageList.size(), totalNumberMessages == messageList.size());
+ LOG.info("final message list size = " + messageList.size());
+
+ consumerAllFail.getLatch().await();
+
+ LOG.info("consumerAllFail successful, count = " + consumerAllFail.getLatch().getCount());
+
+ Iterator<String> keys = messageList.keySet().iterator();
+ for (AtomicInteger counter : messageList.values()) {
+ String message = keys.next();
+ assertTrue("for message " + message + " counter = " + counter.get(), counter.get() == maxRedeliveries + 1);
+ LOG.info("final count for message " + message + " counter = " + counter.get());
+ }
+
+ assertFalse(consumerAllFail.messageReceiptIsOrdered());
+ } finally {
+ if (consumerAllFail != null) {
+ consumerAllFail.setStop(true);
+ }
+ }
+ }
+
+ private static Thread thread(Runnable runnable, boolean daemon) {
+ Thread brokerThread = new Thread(runnable);
+ brokerThread.setDaemon(daemon);
+ brokerThread.start();
+ return brokerThread;
+ }
+
+ private class TestProducer implements Runnable {
+
+ private CountDownLatch latch = null;
+ private String destinationName = null;
+
+ public TestProducer(String destinationName) {
+ this.destinationName = destinationName;
+ // We run the producer 2 times
+ latch = new CountDownLatch(totalNumberMessages);
+ }
+
+ public CountDownLatch getLatch() {
+ return latch;
+ }
+
+ public void run() {
+
+ ActiveMQConnectionFactory connectionFactory = null;
+ ActiveMQConnection connection = null;
+ ActiveMQSession session = null;
+ Destination destination = null;
+
+ try {
+ LOG.info("Started TestProducer for destination (" + destinationName + ")");
+
+ connectionFactory = new ActiveMQConnectionFactory(jmsConnectionURI);
+ connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.setCopyMessageOnSend(false);
+ connection.start();
+ session = (ActiveMQSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ destination = session.createQueue(this.destinationName);
+
+ // Create a MessageProducer from the Session to the Topic or Queue
+ ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(destination);
+ producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+
+ for (int i = 0; i < (producerMessages); i++) {
+ TextMessage message = (TextMessage) session.createTextMessage();
+ message.setLongProperty("TestTime", (System.currentTimeMillis()));
+ try {
+ producer.send(message);
+ LOG.info("Producer (" + destinationName + ")\n" + message.getJMSMessageID() + " = sent messageId\n");
+
+ latch.countDown();
+ LOG.info(" Latch count " + latch.getCount());
+ LOG.info("Producer message list size = " + messageList.keySet().size());
+ messageList.put(message.getJMSMessageID(), new AtomicInteger(0));
+ LOG.info("Producer message list size = " + messageList.keySet().size());
+
+ } catch (Exception deeperException) {
+ LOG.info("Producer for destination (" + destinationName + ") Caught: " + deeperException);
+ }
+
+ Thread.sleep(1000);
+ }
+
+ LOG.info("Finished TestProducer for destination (" + destinationName + ")");
+
+ } catch (Exception e) {
+ LOG.error("Terminating TestProducer(" + destinationName + ")Caught: " + e);
+ } finally {
+ try {
+ if (session != null) {
+ session.close();
+ }
+ if (connection != null) {
+ connection.close();
+ }
+ } catch (Exception e) {
+ LOG.error("Closing connection/session (" + destinationName + ")Caught: " + e);
+ }
+ }
+ }
+ }
+
+ private class TestConsumer implements Runnable, ExceptionListener, MessageListener {
+
+ private CountDownLatch latch = null;
+ private int receivedMessageCounter = 0;
+ private boolean bFakeFail = false;
+ String destinationName = null;
+ boolean bMessageReceiptIsOrdered = true;
+ boolean bStop = false;
+ String previousMessageId = null;
+
+ private ActiveMQConnectionFactory connectionFactory = null;
+ private ActiveMQConnection connection = null;
+ private Session session = null;
+ private MessageConsumer consumer = null;
+
+ public TestConsumer(String destinationName, boolean bFakeFail) {
+ this.bFakeFail = bFakeFail;
+ latch = new CountDownLatch(totalNumberMessages * (this.bFakeFail ? (maxRedeliveries + 1) : 1));
+ this.destinationName = destinationName;
+ }
+
+ public CountDownLatch getLatch() {
+ return latch;
+ }
+
+ public boolean messageReceiptIsOrdered() {
+ return bMessageReceiptIsOrdered;
+ }
+
+ public void run() {
+
+ try {
+ LOG.info("Started TestConsumer for destination (" + destinationName + ")");
+
+ connectionFactory = new ActiveMQConnectionFactory(jmsConnectionURI);
+ connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.setNonBlockingRedelivery(true);
+ session = connection.createSession(true, Session.SESSION_TRANSACTED);
+
+ RedeliveryPolicy policy = connection.getRedeliveryPolicy();
+ policy.setInitialRedeliveryDelay(redeliveryDelay);
+ policy.setBackOffMultiplier(-1);
+ policy.setRedeliveryDelay(redeliveryDelay);
+ policy.setMaximumRedeliveryDelay(-1);
+ policy.setUseExponentialBackOff(false);
+ policy.setMaximumRedeliveries(maxRedeliveries);
+
+ connection.setExceptionListener(this);
+ Destination destination = session.createQueue(destinationName);
+ consumer = session.createConsumer(destination);
+ consumer.setMessageListener(this);
+
+ connection.start();
+
+ while (!bStop) {
+ Thread.sleep(100);
+ }
+
+ LOG.info("Finished TestConsumer for destination name (" + destinationName + ") remaining " + this.latch.getCount()
+ + " messages " + this.toString());
+
+ } catch (Exception e) {
+ LOG.error("Consumer (" + destinationName + ") Caught: " + e);
+ } finally {
+ try {
+ if (consumer != null) {
+ consumer.close();
+ }
+ if (session != null) {
+ session.close();
+ }
+ if (connection != null) {
+ connection.close();
+ }
+ } catch (Exception e) {
+ LOG.error("Closing connection/session (" + destinationName + ")Caught: " + e);
+ }
+ }
+ }
+
+ public synchronized void onException(JMSException ex) {
+ LOG.error("Consumer for destination, (" + destinationName + "), JMS Exception occured. Shutting down client.");
+ }
+
+ public synchronized void setStop(boolean bStop) {
+ this.bStop = bStop;
+ }
+
+ public synchronized void onMessage(Message message) {
+ receivedMessageCounter++;
+ latch.countDown();
+
+ LOG.info("Consumer for destination (" + destinationName + ") latch countdown: " + latch.getCount() +
+ " :: Number messages received " + this.receivedMessageCounter);
+
+ try {
+
+ if (receivedMessageCounter % (maxRedeliveries + 1) == 1) {
+ previousMessageId = message.getJMSMessageID();
+ }
+
+ if (bMessageReceiptIsOrdered) {
+ bMessageReceiptIsOrdered = previousMessageId.trim().equals(message.getJMSMessageID());
+ }
+
+ AtomicInteger counter = messageList.get(message.getJMSMessageID());
+ counter.incrementAndGet();
+
+ LOG.info("Consumer for destination (" + destinationName + ")\n" + message.getJMSMessageID() + " = currentMessageId\n"
+ + previousMessageId + " = previousMessageId\n" + bMessageReceiptIsOrdered + "= bMessageReceiptIsOrdered\n"
+ + ">>LATENCY " + (System.currentTimeMillis() - message.getLongProperty("TestTime")) + "\n" + "message counter = "
+ + counter.get());
+
+ if (!bFakeFail) {
+ LOG.debug("Consumer on destination " + destinationName + " committing JMS Session for message: " + message.toString());
+ session.commit();
+ } else {
+ LOG.debug("Consumer on destination " + destinationName + " rolling back JMS Session for message: " + message.toString());
+ session.rollback(); // rolls back all the consumed messages on the session to
+ }
+
+ } catch (JMSException ex) {
+ ex.printStackTrace();
+ LOG.error("Error reading JMS Message from destination " + destinationName + ".");
+ }
+ }
+ }
+}
Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ1853Test.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/NonBlockingConsumerRedeliveryTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/NonBlockingConsumerRedeliveryTest.java?rev=1195615&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/NonBlockingConsumerRedeliveryTest.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/NonBlockingConsumerRedeliveryTest.java Mon Oct 31 18:54:19 2011
@@ -0,0 +1,381 @@
+/**
+ * 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.usecases;
+
+import static org.junit.Assert.*;
+
+import java.util.LinkedHashSet;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.RedeliveryPolicy;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NonBlockingConsumerRedeliveryTest {
+ private static final Logger LOG = LoggerFactory.getLogger(NonBlockingConsumerRedeliveryTest.class);
+
+ private final String destinationName = "Destination";
+ private final int MSG_COUNT = 100;
+
+ private BrokerService broker;
+ private String connectionUri;
+
+ private ActiveMQConnectionFactory connectionFactory;
+
+ @Test
+ public void testMessageDeleiveredWhenNonBlockingEnabled() throws Exception {
+
+ final LinkedHashSet<Message> received = new LinkedHashSet<Message>();
+ final LinkedHashSet<Message> beforeRollback = new LinkedHashSet<Message>();
+ final LinkedHashSet<Message> afterRollback = new LinkedHashSet<Message>();
+
+ Connection connection = connectionFactory.createConnection();
+ Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createQueue(destinationName);
+ MessageConsumer consumer = session.createConsumer(destination);
+
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ received.add(message);
+ }
+ });
+
+ sendMessages();
+
+ session.commit();
+ connection.start();
+
+ assertTrue("Pre-Rollback expects to receive: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + received.size() + " messages.");
+ return received.size() == MSG_COUNT;
+ }
+ }
+ ));
+
+ beforeRollback.addAll(received);
+ received.clear();
+ session.rollback();
+
+ assertTrue("Post-Rollback expects to receive: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + received.size() + " messages since rollback.");
+ return received.size() == MSG_COUNT;
+ }
+ }
+ ));
+
+ afterRollback.addAll(received);
+ received.clear();
+
+ assertEquals(beforeRollback.size(), afterRollback.size());
+ assertEquals(beforeRollback, afterRollback);
+ session.commit();
+ }
+
+ @Test
+ public void testMessageDeleiveryDoesntStop() throws Exception {
+
+ final LinkedHashSet<Message> received = new LinkedHashSet<Message>();
+ final LinkedHashSet<Message> beforeRollback = new LinkedHashSet<Message>();
+ final LinkedHashSet<Message> afterRollback = new LinkedHashSet<Message>();
+
+ Connection connection = connectionFactory.createConnection();
+ Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createQueue(destinationName);
+ MessageConsumer consumer = session.createConsumer(destination);
+
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ received.add(message);
+ }
+ });
+
+ sendMessages();
+ connection.start();
+
+ assertTrue("Pre-Rollback expects to receive: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + received.size() + " messages.");
+ return received.size() == MSG_COUNT;
+ }
+ }
+ ));
+
+ beforeRollback.addAll(received);
+ received.clear();
+ session.rollback();
+
+ sendMessages();
+
+ assertTrue("Post-Rollback expects to receive: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + received.size() + " messages since rollback.");
+ return received.size() == MSG_COUNT * 2;
+ }
+ }
+ ));
+
+ afterRollback.addAll(received);
+ received.clear();
+
+ assertEquals(beforeRollback.size() * 2, afterRollback.size());
+
+ session.commit();
+ }
+
+ @Test
+ public void testNonBlockingMessageDeleiveryIsDelayed() throws Exception {
+ final LinkedHashSet<Message> received = new LinkedHashSet<Message>();
+
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.getRedeliveryPolicy().setInitialRedeliveryDelay(TimeUnit.SECONDS.toMillis(6));
+ Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createQueue(destinationName);
+ MessageConsumer consumer = session.createConsumer(destination);
+
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ received.add(message);
+ }
+ });
+
+ sendMessages();
+ connection.start();
+
+ assertTrue("Pre-Rollback expects to receive: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + received.size() + " messages.");
+ return received.size() == MSG_COUNT;
+ }
+ }
+ ));
+
+ received.clear();
+ session.rollback();
+
+ assertFalse("Delayed redelivery test not expecting any messages yet.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ return received.size() > 0;
+ }
+ }, TimeUnit.SECONDS.toMillis(4)
+ ));
+
+ session.commit();
+ session.close();
+ }
+
+ @Test
+ public void testNonBlockingMessageDeleiveryWithRollbacks() throws Exception {
+ final LinkedHashSet<Message> received = new LinkedHashSet<Message>();
+
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ final Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
+ final Destination destination = session.createQueue(destinationName);
+ final MessageConsumer consumer = session.createConsumer(destination);
+
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ received.add(message);
+ }
+ });
+
+ sendMessages();
+ connection.start();
+
+ assertTrue("Pre-Rollback expects to receive: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + received.size() + " messages.");
+ return received.size() == MSG_COUNT;
+ }
+ }
+ ));
+
+ received.clear();
+
+ consumer.setMessageListener(new MessageListener() {
+
+ int count = 0;
+
+ @Override
+ public void onMessage(Message message) {
+
+ if (++count > 10) {
+ try {
+ session.rollback();
+ LOG.info("Rolling back session.");
+ count = 0;
+ } catch (JMSException e) {
+ LOG.warn("Caught an unexcepted exception: " + e.getMessage());
+ }
+ } else {
+ received.add(message);
+ try {
+ session.commit();
+ } catch (JMSException e) {
+ LOG.warn("Caught an unexcepted exception: " + e.getMessage());
+ }
+ }
+ }
+ });
+
+ session.rollback();
+
+ assertTrue("Post-Rollback expects to receive: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + received.size() + " messages since rollback.");
+ return received.size() == MSG_COUNT;
+ }
+ }
+ ));
+
+ assertEquals(MSG_COUNT, received.size());
+ session.commit();
+ }
+
+ @Test
+ public void testNonBlockingMessageDeleiveryWithAllRolledBack() throws Exception {
+ final LinkedHashSet<Message> received = new LinkedHashSet<Message>();
+ final LinkedHashSet<Message> dlqed = new LinkedHashSet<Message>();
+
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.getRedeliveryPolicy().setMaximumRedeliveries(5);
+ final Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
+ final Destination destination = session.createQueue(destinationName);
+ final Destination dlq = session.createQueue("ActiveMQ.DLQ");
+ final MessageConsumer consumer = session.createConsumer(destination);
+ final MessageConsumer dlqConsumer = session.createConsumer(dlq);
+
+ dlqConsumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ dlqed.add(message);
+ }
+ });
+
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ received.add(message);
+ }
+ });
+
+ sendMessages();
+ connection.start();
+
+ assertTrue("Pre-Rollback expects to receive: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + received.size() + " messages.");
+ return received.size() == MSG_COUNT;
+ }
+ }
+ ));
+
+ session.rollback();
+
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ try {
+ session.rollback();
+ } catch (JMSException e) {
+ LOG.warn("Caught an unexcepted exception: " + e.getMessage());
+ }
+ }
+ });
+
+ assertTrue("Post-Rollback expects to DLQ: " + MSG_COUNT + " messages.",
+ Wait.waitFor(new Wait.Condition(){
+ public boolean isSatisified() throws Exception {
+ LOG.info("Consumer has received " + dlqed.size() + " messages in DLQ.");
+ return dlqed.size() == MSG_COUNT;
+ }
+ }
+ ));
+
+ session.commit();
+ }
+
+ private void sendMessages() throws Exception {
+ Connection connection = connectionFactory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createQueue(destinationName);
+ MessageProducer producer = session.createProducer(destination);
+ for(int i = 0; i < MSG_COUNT; ++i) {
+ producer.send(session.createTextMessage("" + i));
+ }
+ }
+
+ @Before
+ public void startBroker() throws Exception {
+ broker = new BrokerService();
+ broker.setDeleteAllMessagesOnStartup(true);
+ broker.setPersistent(false);
+ broker.setUseJmx(false);
+ broker.addConnector("tcp://0.0.0.0:0");
+ broker.start();
+ broker.waitUntilStarted();
+
+ connectionUri = broker.getTransportConnectors().get(0).getPublishableConnectString();
+ connectionFactory = new ActiveMQConnectionFactory(connectionUri);
+ connectionFactory.setNonBlockingRedelivery(true);
+
+ RedeliveryPolicy policy = connectionFactory.getRedeliveryPolicy();
+ policy.setInitialRedeliveryDelay(TimeUnit.SECONDS.toMillis(2));
+ policy.setBackOffMultiplier(-1);
+ policy.setRedeliveryDelay(TimeUnit.SECONDS.toMillis(2));
+ policy.setMaximumRedeliveryDelay(-1);
+ policy.setUseExponentialBackOff(false);
+ policy.setMaximumRedeliveries(-1);
+ }
+
+ @After
+ public void stopBroker() throws Exception {
+ broker.stop();
+ broker.waitUntilStopped();
+ }
+
+}
Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/NonBlockingConsumerRedeliveryTest.java
------------------------------------------------------------------------------
svn:eol-style = native