test/MessageListenerRedeliveryTest.cs - activemq-nms-openwire - Git at Google

 /*
  * 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.
  */

 using System;
 using System.Threading;
 using System.Collections;
 using Apache.NMS;
 using Apache.NMS.Util;
 using Apache.NMS.Test;
 using Apache.NMS.Policies;
 using NUnit.Framework;

 namespace Apache.NMS.ActiveMQ.Test
 {
     [TestFixture]
     public class MessageListenerRedeliveryTest : NMSTestSupport
     {
         private Connection connection;
         private volatile int counter;
         private ISession session;
         private ArrayList received;
         private ArrayList dlqMessages;
         private int maxDeliveries;

         private CountDownLatch gotOneMessage;
         private CountDownLatch gotTwoMessages;
         private CountDownLatch gotOneDlqMessage;
         private CountDownLatch gotMaxRedeliveries;

         [SetUp]
         public override void SetUp()
         {
             this.connection = (Connection) CreateConnection();
             this.connection.RedeliveryPolicy = GetRedeliveryPolicy();
             this.gotOneMessage = new CountDownLatch(1);
             this.gotTwoMessages = new CountDownLatch(2);
             this.gotOneDlqMessage = new CountDownLatch(1);
             this.maxDeliveries = GetRedeliveryPolicy().MaximumRedeliveries;
             this.gotMaxRedeliveries = new CountDownLatch(maxDeliveries);
             this.received = new ArrayList();
             this.dlqMessages = new ArrayList();
             this.counter = 0;
         }

         [TearDown]
         public override void TearDown()
         {
             this.session = null;

             if(this.connection != null)
             {
                 this.connection.Close();
                 this.connection = null;
             }

             base.TearDown();
         }

         protected IRedeliveryPolicy GetRedeliveryPolicy()
         {
             RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
             redeliveryPolicy.InitialRedeliveryDelay = 1000;
             redeliveryPolicy.MaximumRedeliveries = 3;
             redeliveryPolicy.BackOffMultiplier = (short)2;
             redeliveryPolicy.UseExponentialBackOff = true;
             return redeliveryPolicy;
         }

         private void OnMessageListener(IMessage message)
         {
             counter++;
             if(this.counter <= 4)
             {
                 session.Rollback();
             }
             else
             {
                 message.Acknowledge();
                 session.Commit();
             }
         }

         private void OnTracedReceiveMessage(IMessage message)
         {
             try
             {
                 received.Add(((ITextMessage) message).Text);
             }
             catch (Exception e)
             {
                 Assert.Fail("Error: " + e.Message);
             }

             if (++counter < maxDeliveries)
             {
                 throw new Exception("force a redelivery");
             }

             // new blood
             counter = 0;
             gotTwoMessages.countDown();
         }

         private void OnDlqMessage(IMessage message)
         {
             dlqMessages.Add(message);
             gotOneDlqMessage.countDown();
         }

         private void OnRedeliveredMessage(IMessage message)
         {
             gotMaxRedeliveries.countDown();
             throw new Exception("Test Forcing a Rollback");
         }

         [Test]
         public void TestQueueRollbackConsumerListener()
         {
             connection.Start();

             this.session = connection.CreateSession(AcknowledgementMode.Transactional);
             ITemporaryQueue queue = session.CreateTemporaryQueue();
             IMessageProducer producer = session.CreateProducer(queue);
             IMessage message = session.CreateTextMessage("Test Message");
             producer.Send(message);
             session.Commit();

             IMessageConsumer consumer = session.CreateConsumer(queue);

             consumer.Listener += new MessageListener(OnMessageListener);

             Thread.Sleep(500);

             // first try.. should get 2 since there is no delay on the
             // first redeliver..
             Assert.AreEqual(2, counter);

             Thread.Sleep(1000);

             // 2nd redeliver (redelivery after 1 sec)
             Assert.AreEqual(3, counter);

             Thread.Sleep(2000);

             // 3rd redeliver (redelivery after 2 seconds) - it should give up after
             // that
             Assert.AreEqual(4, counter);

             // create new message
             producer.Send(session.CreateTextMessage("Test Message Again"));
             session.Commit();

             Thread.Sleep(500);

             // it should be committed, so no redelivery
             Assert.AreEqual(5, counter);

             Thread.Sleep(1500);

             // no redelivery, counter should still be 5
             Assert.AreEqual(5, counter);

             session.Close();
         }

         [Test]
         public void TestQueueRollbackSessionListener()
         {
             connection.Start();

             this.session = connection.CreateSession(AcknowledgementMode.Transactional);
             IQueue queue = session.CreateTemporaryQueue();
             IMessageProducer producer = CreateProducer(session, queue);
             IMessage message = CreateTextMessage(session);
             producer.Send(message);
             session.Commit();

             IMessageConsumer consumer = session.CreateConsumer(queue);
             consumer.Listener += new MessageListener(OnMessageListener);

             Thread.Sleep(1000);

             // first try
             Assert.AreEqual(2, counter);

             Thread.Sleep(1500);

             // second try (redelivery after 1 sec)
             Assert.AreEqual(3, counter);

             Thread.Sleep(3000);

             // third try (redelivery after 2 seconds) - it should give up after that
             Assert.AreEqual(4, counter);

             // create new message
             producer.Send(CreateTextMessage(session));
             session.Commit();

             Thread.Sleep(1000);

             // it should be committed, so no redelivery
             Assert.AreEqual(5, counter);

             Thread.Sleep(2000);

             // no redelivery, counter should still be 4
             Assert.AreEqual(5, counter);

             session.Close();
         }

         [Test]
         public void TestQueueSessionListenerExceptionRetry()
         {
             connection.Start();

             ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
             IQueue queue = session.CreateTemporaryQueue();
             IMessageProducer producer = CreateProducer(session, queue);
             IMessage message = CreateTextMessage(session, "1");
             producer.Send(message);
             message = CreateTextMessage(session, "2");
             producer.Send(message);

             IMessageConsumer consumer = session.CreateConsumer(queue);
             consumer.Listener += new MessageListener(OnTracedReceiveMessage);

             Assert.IsTrue(gotTwoMessages.await(TimeSpan.FromSeconds(20)), "got message before retry expiry");

             for (int i = 0; i < maxDeliveries; i++)
             {
                 Assert.AreEqual("1", received[i], "got first redelivered: " + i);
             }
             for (int i = maxDeliveries; i < maxDeliveries * 2; i++)
             {
                 Assert.AreEqual("2", received[i], "got first redelivered: " + i);
             }

             session.Close();
         }

         [Test]
         public void TestQueueSessionListenerExceptionDlq()
         {
             connection.Start();

             session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
             IQueue queue = session.CreateTemporaryQueue();
             IMessageProducer producer = CreateProducer(session, queue);
             IMessage message = CreateTextMessage(session);
             producer.Send(message);

             IDestination dlqDestination = session.GetQueue("ActiveMQ.DLQ");
             connection.DeleteDestination(dlqDestination);
             IMessageConsumer dlqConsumer = session.CreateConsumer(dlqDestination);
             dlqConsumer.Listener += new MessageListener(OnDlqMessage);

             IMessageConsumer consumer = session.CreateConsumer(queue);
             consumer.Listener += new MessageListener(OnRedeliveredMessage);

             Assert.IsTrue(gotMaxRedeliveries.await(TimeSpan.FromSeconds(20)), "got message before retry expiry");

             // check DLQ
             Assert.IsTrue(gotOneDlqMessage.await(TimeSpan.FromSeconds(20)), "got dlq message");

             // check DLQ message cause is captured
             message = dlqMessages[0] as IMessage;
             Assert.IsNotNull(message, "dlq message captured");
             String cause = message.Properties.GetString("dlqDeliveryFailureCause");

             Assert.IsTrue(cause.Contains("JMSException"), "cause 'cause' exception is remembered");
             Assert.IsTrue(cause.Contains("Test"), "is correct exception");
             Assert.IsTrue(cause.Contains("RedeliveryPolicy"), "cause policy is remembered");

             session.Close();
         }

         private void OnMessageThenRollback(IMessage message)
         {
             gotOneMessage.countDown();
             try
             {
                 session.Rollback();
             }
             catch (Exception)
             {
             }

             throw new Exception("Test force a redelivery");
         }

         [Test]
         public void TestTransactedQueueSessionListenerExceptionDlq()
         {
             connection.Start();

             session = connection.CreateSession(AcknowledgementMode.Transactional);
             IQueue queue = session.CreateTemporaryQueue();
             IMessageProducer producer = CreateProducer(session, queue);
             IMessage message = CreateTextMessage(session);
             producer.Send(message);
             session.Commit();

             IDestination dlqDestination = session.GetQueue("ActiveMQ.DLQ");
             connection.DeleteDestination(dlqDestination);
             IMessageConsumer dlqConsumer = session.CreateConsumer(dlqDestination);
             dlqConsumer.Listener += new MessageListener(OnDlqMessage);

             IMessageConsumer consumer = session.CreateConsumer(queue);
             consumer.Listener += new MessageListener(OnMessageThenRollback);

             Assert.IsTrue(gotOneMessage.await(TimeSpan.FromSeconds(20)), "got message before retry expiry");

             // check DLQ
             Assert.IsTrue(gotOneDlqMessage.await(TimeSpan.FromSeconds(20)), "got dlq message");

             // check DLQ message cause is captured
             message = dlqMessages[0] as IMessage;
             Assert.IsNotNull(message, "dlq message captured");
             String cause = message.Properties.GetString("dlqDeliveryFailureCause");

             Assert.IsTrue(cause.Contains("JMSException"), "cause 'cause' exception is remembered");
             Assert.IsTrue(cause.Contains("Test force"), "is correct exception");
             Assert.IsTrue(cause.Contains("RedeliveryPolicy"), "cause policy is remembered");

             session.Close();
         }

         private ITextMessage CreateTextMessage(ISession session, String text)
         {
             return session.CreateTextMessage(text);
         }

         private ITextMessage CreateTextMessage(ISession session)
         {
             return session.CreateTextMessage("Hello");
         }

         private IMessageProducer CreateProducer(ISession session, IDestination queue)
         {
             IMessageProducer producer = session.CreateProducer(queue);
             producer.DeliveryMode = GetDeliveryMode();
             return producer;
         }

         protected MsgDeliveryMode GetDeliveryMode()
         {
             return MsgDeliveryMode.Persistent;
         }
     }
 }
	/*
	* 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.
	*/

	using System;
	using System.Threading;
	using System.Collections;
	using Apache.NMS;
	using Apache.NMS.Util;
	using Apache.NMS.Test;
	using Apache.NMS.Policies;
	using NUnit.Framework;

	namespace Apache.NMS.ActiveMQ.Test
	{
	[TestFixture]
	public class MessageListenerRedeliveryTest : NMSTestSupport
	{
	private Connection connection;
	private volatile int counter;
	private ISession session;
	private ArrayList received;
	private ArrayList dlqMessages;
	private int maxDeliveries;

	private CountDownLatch gotOneMessage;
	private CountDownLatch gotTwoMessages;
	private CountDownLatch gotOneDlqMessage;
	private CountDownLatch gotMaxRedeliveries;

	[SetUp]
	public override void SetUp()
	{
	this.connection = (Connection) CreateConnection();
	this.connection.RedeliveryPolicy = GetRedeliveryPolicy();
	this.gotOneMessage = new CountDownLatch(1);
	this.gotTwoMessages = new CountDownLatch(2);
	this.gotOneDlqMessage = new CountDownLatch(1);
	this.maxDeliveries = GetRedeliveryPolicy().MaximumRedeliveries;
	this.gotMaxRedeliveries = new CountDownLatch(maxDeliveries);
	this.received = new ArrayList();
	this.dlqMessages = new ArrayList();
	this.counter = 0;
	}

	[TearDown]
	public override void TearDown()
	{
	this.session = null;

	if(this.connection != null)
	{
	this.connection.Close();
	this.connection = null;
	}

	base.TearDown();
	}

	protected IRedeliveryPolicy GetRedeliveryPolicy()
	{
	RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
	redeliveryPolicy.InitialRedeliveryDelay = 1000;
	redeliveryPolicy.MaximumRedeliveries = 3;
	redeliveryPolicy.BackOffMultiplier = (short)2;
	redeliveryPolicy.UseExponentialBackOff = true;
	return redeliveryPolicy;
	}

	private void OnMessageListener(IMessage message)
	{
	counter++;
	if(this.counter <= 4)
	{
	session.Rollback();
	}
	else
	{
	message.Acknowledge();
	session.Commit();
	}
	}

	private void OnTracedReceiveMessage(IMessage message)
	{
	try
	{
	received.Add(((ITextMessage) message).Text);
	}
	catch (Exception e)
	{
	Assert.Fail("Error: " + e.Message);
	}

	if (++counter < maxDeliveries)
	{
	throw new Exception("force a redelivery");
	}

	// new blood
	counter = 0;
	gotTwoMessages.countDown();
	}

	private void OnDlqMessage(IMessage message)
	{
	dlqMessages.Add(message);
	gotOneDlqMessage.countDown();
	}

	private void OnRedeliveredMessage(IMessage message)
	{
	gotMaxRedeliveries.countDown();
	throw new Exception("Test Forcing a Rollback");
	}

	[Test]
	public void TestQueueRollbackConsumerListener()
	{
	connection.Start();

	this.session = connection.CreateSession(AcknowledgementMode.Transactional);
	ITemporaryQueue queue = session.CreateTemporaryQueue();
	IMessageProducer producer = session.CreateProducer(queue);
	IMessage message = session.CreateTextMessage("Test Message");
	producer.Send(message);
	session.Commit();

	IMessageConsumer consumer = session.CreateConsumer(queue);

	consumer.Listener += new MessageListener(OnMessageListener);

	Thread.Sleep(500);

	// first try.. should get 2 since there is no delay on the
	// first redeliver..
	Assert.AreEqual(2, counter);

	Thread.Sleep(1000);

	// 2nd redeliver (redelivery after 1 sec)
	Assert.AreEqual(3, counter);

	Thread.Sleep(2000);

	// 3rd redeliver (redelivery after 2 seconds) - it should give up after
	// that
	Assert.AreEqual(4, counter);

	// create new message
	producer.Send(session.CreateTextMessage("Test Message Again"));
	session.Commit();

	Thread.Sleep(500);

	// it should be committed, so no redelivery
	Assert.AreEqual(5, counter);

	Thread.Sleep(1500);

	// no redelivery, counter should still be 5
	Assert.AreEqual(5, counter);

	session.Close();
	}

	[Test]
	public void TestQueueRollbackSessionListener()
	{
	connection.Start();

	this.session = connection.CreateSession(AcknowledgementMode.Transactional);
	IQueue queue = session.CreateTemporaryQueue();
	IMessageProducer producer = CreateProducer(session, queue);
	IMessage message = CreateTextMessage(session);
	producer.Send(message);
	session.Commit();

	IMessageConsumer consumer = session.CreateConsumer(queue);
	consumer.Listener += new MessageListener(OnMessageListener);

	Thread.Sleep(1000);

	// first try
	Assert.AreEqual(2, counter);

	Thread.Sleep(1500);

	// second try (redelivery after 1 sec)
	Assert.AreEqual(3, counter);

	Thread.Sleep(3000);

	// third try (redelivery after 2 seconds) - it should give up after that
	Assert.AreEqual(4, counter);

	// create new message
	producer.Send(CreateTextMessage(session));
	session.Commit();

	Thread.Sleep(1000);

	// it should be committed, so no redelivery
	Assert.AreEqual(5, counter);

	Thread.Sleep(2000);

	// no redelivery, counter should still be 4
	Assert.AreEqual(5, counter);

	session.Close();
	}

	[Test]
	public void TestQueueSessionListenerExceptionRetry()
	{
	connection.Start();

	ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
	IQueue queue = session.CreateTemporaryQueue();
	IMessageProducer producer = CreateProducer(session, queue);
	IMessage message = CreateTextMessage(session, "1");
	producer.Send(message);
	message = CreateTextMessage(session, "2");
	producer.Send(message);

	IMessageConsumer consumer = session.CreateConsumer(queue);
	consumer.Listener += new MessageListener(OnTracedReceiveMessage);

	Assert.IsTrue(gotTwoMessages.await(TimeSpan.FromSeconds(20)), "got message before retry expiry");

	for (int i = 0; i < maxDeliveries; i++)
	{
	Assert.AreEqual("1", received[i], "got first redelivered: " + i);
	}
	for (int i = maxDeliveries; i < maxDeliveries * 2; i++)
	{
	Assert.AreEqual("2", received[i], "got first redelivered: " + i);
	}

	session.Close();
	}

	[Test]
	public void TestQueueSessionListenerExceptionDlq()
	{
	connection.Start();

	session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
	IQueue queue = session.CreateTemporaryQueue();
	IMessageProducer producer = CreateProducer(session, queue);
	IMessage message = CreateTextMessage(session);
	producer.Send(message);

	IDestination dlqDestination = session.GetQueue("ActiveMQ.DLQ");
	connection.DeleteDestination(dlqDestination);
	IMessageConsumer dlqConsumer = session.CreateConsumer(dlqDestination);
	dlqConsumer.Listener += new MessageListener(OnDlqMessage);

	IMessageConsumer consumer = session.CreateConsumer(queue);
	consumer.Listener += new MessageListener(OnRedeliveredMessage);

	Assert.IsTrue(gotMaxRedeliveries.await(TimeSpan.FromSeconds(20)), "got message before retry expiry");

	// check DLQ
	Assert.IsTrue(gotOneDlqMessage.await(TimeSpan.FromSeconds(20)), "got dlq message");

	// check DLQ message cause is captured
	message = dlqMessages[0] as IMessage;
	Assert.IsNotNull(message, "dlq message captured");
	String cause = message.Properties.GetString("dlqDeliveryFailureCause");

	Assert.IsTrue(cause.Contains("JMSException"), "cause 'cause' exception is remembered");
	Assert.IsTrue(cause.Contains("Test"), "is correct exception");
	Assert.IsTrue(cause.Contains("RedeliveryPolicy"), "cause policy is remembered");

	session.Close();
	}

	private void OnMessageThenRollback(IMessage message)
	{
	gotOneMessage.countDown();
	try
	{
	session.Rollback();
	}
	catch (Exception)
	{
	}

	throw new Exception("Test force a redelivery");
	}

	[Test]
	public void TestTransactedQueueSessionListenerExceptionDlq()
	{
	connection.Start();

	session = connection.CreateSession(AcknowledgementMode.Transactional);
	IQueue queue = session.CreateTemporaryQueue();
	IMessageProducer producer = CreateProducer(session, queue);
	IMessage message = CreateTextMessage(session);
	producer.Send(message);
	session.Commit();

	IDestination dlqDestination = session.GetQueue("ActiveMQ.DLQ");
	connection.DeleteDestination(dlqDestination);
	IMessageConsumer dlqConsumer = session.CreateConsumer(dlqDestination);
	dlqConsumer.Listener += new MessageListener(OnDlqMessage);

	IMessageConsumer consumer = session.CreateConsumer(queue);
	consumer.Listener += new MessageListener(OnMessageThenRollback);

	Assert.IsTrue(gotOneMessage.await(TimeSpan.FromSeconds(20)), "got message before retry expiry");

	// check DLQ
	Assert.IsTrue(gotOneDlqMessage.await(TimeSpan.FromSeconds(20)), "got dlq message");

	// check DLQ message cause is captured
	message = dlqMessages[0] as IMessage;
	Assert.IsNotNull(message, "dlq message captured");
	String cause = message.Properties.GetString("dlqDeliveryFailureCause");

	Assert.IsTrue(cause.Contains("JMSException"), "cause 'cause' exception is remembered");
	Assert.IsTrue(cause.Contains("Test force"), "is correct exception");
	Assert.IsTrue(cause.Contains("RedeliveryPolicy"), "cause policy is remembered");

	session.Close();
	}

	private ITextMessage CreateTextMessage(ISession session, String text)
	{
	return session.CreateTextMessage(text);
	}

	private ITextMessage CreateTextMessage(ISession session)
	{
	return session.CreateTextMessage("Hello");
	}

	private IMessageProducer CreateProducer(ISession session, IDestination queue)
	{
	IMessageProducer producer = session.CreateProducer(queue);
	producer.DeliveryMode = GetDeliveryMode();
	return producer;
	}

	protected MsgDeliveryMode GetDeliveryMode()
	{
	return MsgDeliveryMode.Persistent;
	}
	}
	}