src/test/csharp/NonBlockingConsumerRedeliveryTest.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.Collections.Generic;
 using System.Threading;
 using Apache.NMS.Test;
 using Apache.NMS.ActiveMQ.Commands;
 using Apache.NMS.Util;
 using NUnit.Framework;

 namespace Apache.NMS.ActiveMQ.Test
 {
 	[TestFixture]
 	public class NonBlockingConsumerRedeliveryTest  : NMSTestSupport
 	{
         private Connection connection;
 		private ISession session;
     	private readonly int MSG_COUNT = 5;
 		private int count;

         private List<IMessage> received = new List<IMessage>();
         private List<IMessage> dlqed = new List<IMessage>();
         private List<IMessage> beforeRollback = new List<IMessage>();
         private List<IMessage> afterRollback = new List<IMessage>();

         [SetUp]
         public override void SetUp()
         {
 			base.SetUp();

 			session = null;
         	connection = (Connection) CreateConnection();
         	connection.NonBlockingRedelivery = true;

 			DeleteDLQ();

 			this.received.Clear();
 			this.beforeRollback.Clear();
 			this.afterRollback.Clear();
 			this.dlqed.Clear();
 			this.count = 0;
         }

         [TearDown]
         public override void TearDown()
         {
             if(this.connection != null)
             {
                 this.connection.Close();
                 this.connection = null;
             }

             base.TearDown();
         }

 		public void OnMessage(IMessage message)
 		{
 			this.received.Add(message);
 		}

 		private void AssertReceived(int count, String message)
 		{
 			AssertReceived(this.received, count, message);
 		}

 		private void AssertReceived(List<IMessage> target, int count, String message)
 		{
 			for (int i = 0; i < 30; ++i)
 			{
 				if (target.Count == count)
 				{
 					break;
 				}
 				Thread.Sleep(1000);
 			}
 			Assert.AreEqual(count, target.Count, message);
 		}

 		[Test]
 		public void testMessageDeleiveredWhenNonBlockingEnabled()
 		{
 	        session = connection.CreateSession(AcknowledgementMode.Transactional);
 			IDestination destination = session.CreateTemporaryQueue();
 	        IMessageConsumer consumer = session.CreateConsumer(destination);

 			consumer.Listener += OnMessage;

 	        SendMessages(destination);
 	        session.Commit();

 			connection.Start();

 			AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        this.beforeRollback.AddRange(received);
 	        this.received.Clear();
 	        session.Rollback();

 			AssertReceived(MSG_COUNT, "Post-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        this.afterRollback.AddRange(received);
 	        this.received.Clear();

 	        Assert.AreEqual(this.beforeRollback.Count, this.afterRollback.Count);
 	        Assert.AreEqual(this.beforeRollback, this.afterRollback);
 	        session.Commit();
 		}

 		[Test]
 		public void testMessageDeleiveredInCorrectOrder()
 		{
 	        session = connection.CreateSession(AcknowledgementMode.Transactional);
 	        IDestination destination = session.CreateTemporaryQueue();
 	        IMessageConsumer consumer = session.CreateConsumer(destination);

 			consumer.Listener += OnMessage;

 	        SendMessages(destination);

 	        session.Commit();
 	        connection.Start();

 			AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        beforeRollback.AddRange(received);
 	        received.Clear();
 	        session.Rollback();

 			AssertReceived(MSG_COUNT, "Post-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        afterRollback.AddRange(received);
 	        received.Clear();

 	        Assert.AreEqual(beforeRollback.Count, afterRollback.Count);
 	        Assert.AreEqual(beforeRollback, afterRollback);

 			IEnumerator<IMessage> after = afterRollback.GetEnumerator();
 			IEnumerator<IMessage> before = beforeRollback.GetEnumerator();

 			while (after.MoveNext() && before.MoveNext())
 			{
 				ITextMessage original = before.Current as ITextMessage;
 				ITextMessage rolledBack = after.Current as ITextMessage;

 				int originalId = Int32.Parse(original.Text);
 				int rolledBackId = Int32.Parse (rolledBack.Text);

 				Assert.AreEqual(originalId, rolledBackId);
 			}

 	        session.Commit();
 		}

 		[Test]
 		public void testMessageDeleiveryDoesntStop()
 		{
 	        session = connection.CreateSession(AcknowledgementMode.Transactional);
 			IDestination destination = session.CreateTemporaryQueue();
 	        IMessageConsumer consumer = session.CreateConsumer(destination);

 			consumer.Listener += OnMessage;

 	        SendMessages(destination);
 	        connection.Start();

 			AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        beforeRollback.AddRange(received);
 	        received.Clear();
 	        session.Rollback();

 	        SendMessages(destination);

 			AssertReceived(MSG_COUNT * 2, "Post-Rollback expects to receive: " + MSG_COUNT * 2 + " messages.");

 	        afterRollback.AddRange(received);
 	        received.Clear();

 	        Assert.AreEqual(beforeRollback.Count * 2, afterRollback.Count);

 	        session.Commit();
 		}

 		[Test]
 		public void testNonBlockingMessageDeleiveryIsDelayed()
 		{
 	        connection.RedeliveryPolicy.InitialRedeliveryDelay = 7000;
 	        session = connection.CreateSession(AcknowledgementMode.Transactional);
 	        IDestination destination = session.CreateTemporaryQueue();
 	        IMessageConsumer consumer = session.CreateConsumer(destination);

 			consumer.Listener += OnMessage;

 	        SendMessages(destination);
 	        connection.Start();

 			AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        received.Clear();
 	        session.Rollback();

 			Thread.Sleep(4000);
 			Assert.IsFalse(this.received.Count > 0, "Delayed redelivery test not expecting any messages yet.");

 			AssertReceived(MSG_COUNT, "Post-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        session.Commit();
 	        session.Close();
 		}

         public void OnMessageWithSomeRollbacks(IMessage message)
 		{
             if (++count > 10)
 			{
                 try
 				{
                     session.Rollback();
                     Tracer.Info("Rolling back session.");
                     count = 0;
                 }
 				catch (Exception e)
 				{
 					Tracer.WarnFormat("Caught an unexcepted exception: {0}", e.Message);
                 }
             }
 			else
 			{
                 received.Add(message);
                 try
 				{
                     session.Commit();
 				}
 				catch (Exception e)
 				{
 					Tracer.WarnFormat("Caught an unexcepted exception: {0}", e.Message);
                 }
             }
         }

 		[Test]
 		public void testNonBlockingMessageDeleiveryWithRollbacks()
 		{
 	        session = connection.CreateSession(AcknowledgementMode.Transactional);
 	        IDestination destination = session.CreateTemporaryQueue();
 	        IMessageConsumer consumer = session.CreateConsumer(destination);

 			consumer.Listener += OnMessage;

 	        SendMessages(destination);
 	        connection.Start();

 			AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        received.Clear();

 			consumer.Listener -= OnMessage;
 			consumer.Listener += OnMessageWithSomeRollbacks;

 	        session.Rollback();

 			AssertReceived(MSG_COUNT, "Post-Rollback expects to receive: " + MSG_COUNT + " messages.");

 	        Assert.AreEqual(MSG_COUNT, received.Count);
 	        session.Commit();
 		}

 		private void OnDLQMessage(IMessage message)
 		{
 			Tracer.DebugFormat("DLQ Message {0}", message);
             dlqed.Add(message);
 		}

 		private void OnMessageAlwaysRollsBack(IMessage message)
 		{
 			session.Rollback();
 		}

 		[Test]
 		public void testNonBlockingMessageDeleiveryWithAllRolledBack()
 		{
 	        connection.RedeliveryPolicy.MaximumRedeliveries = 3;
 	        session = connection.CreateSession(AcknowledgementMode.Transactional);
 	        IDestination destination = session.CreateTemporaryQueue();
 	        IDestination dlq = session.GetQueue("ActiveMQ.DLQ");
 	        IMessageConsumer consumer = session.CreateConsumer(destination);
 	        IMessageConsumer dlqConsumer = session.CreateConsumer(dlq);

 			dlqConsumer.Listener += OnDLQMessage;
 			consumer.Listener += OnMessage;

 	        SendMessages(destination);
 	        connection.Start();

 			AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

 			consumer.Listener -= OnMessage;
 			consumer.Listener += OnMessageAlwaysRollsBack;

 			session.Rollback();

 			AssertReceived(dlqed, MSG_COUNT, "Post-Rollback expects to DLQ: " + MSG_COUNT + " messages.");

 	        session.Commit();
 		}

 		private void DeleteDLQ()
 		{
 	        session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
 	        IDestination dlq = session.GetQueue("ActiveMQ.DLQ");
 			try
 			{
 				connection.DeleteDestination(dlq);
 			}
 			catch
 			{
 			}
 		}

 	    private void SendMessages(IDestination destination)
 		{
 	        IConnection connection = CreateConnection();
 	        ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
 	        IMessageProducer producer = session.CreateProducer(destination);
 	        for(int i = 0; i < MSG_COUNT; ++i)
 			{
 	            producer.Send(session.CreateTextMessage("" + i));
 	        }
 			connection.Close();
 	    }
 	}
 }
	/*
	* 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.Collections.Generic;
	using System.Threading;
	using Apache.NMS.Test;
	using Apache.NMS.ActiveMQ.Commands;
	using Apache.NMS.Util;
	using NUnit.Framework;

	namespace Apache.NMS.ActiveMQ.Test
	{
	[TestFixture]
	public class NonBlockingConsumerRedeliveryTest : NMSTestSupport
	{
	private Connection connection;
	private ISession session;
	private readonly int MSG_COUNT = 5;
	private int count;

	private List<IMessage> received = new List<IMessage>();
	private List<IMessage> dlqed = new List<IMessage>();
	private List<IMessage> beforeRollback = new List<IMessage>();
	private List<IMessage> afterRollback = new List<IMessage>();

	[SetUp]
	public override void SetUp()
	{
	base.SetUp();

	session = null;
	connection = (Connection) CreateConnection();
	connection.NonBlockingRedelivery = true;

	DeleteDLQ();

	this.received.Clear();
	this.beforeRollback.Clear();
	this.afterRollback.Clear();
	this.dlqed.Clear();
	this.count = 0;
	}

	[TearDown]
	public override void TearDown()
	{
	if(this.connection != null)
	{
	this.connection.Close();
	this.connection = null;
	}

	base.TearDown();
	}

	public void OnMessage(IMessage message)
	{
	this.received.Add(message);
	}

	private void AssertReceived(int count, String message)
	{
	AssertReceived(this.received, count, message);
	}

	private void AssertReceived(List<IMessage> target, int count, String message)
	{
	for (int i = 0; i < 30; ++i)
	{
	if (target.Count == count)
	{
	break;
	}
	Thread.Sleep(1000);
	}
	Assert.AreEqual(count, target.Count, message);
	}

	[Test]
	public void testMessageDeleiveredWhenNonBlockingEnabled()
	{
	session = connection.CreateSession(AcknowledgementMode.Transactional);
	IDestination destination = session.CreateTemporaryQueue();
	IMessageConsumer consumer = session.CreateConsumer(destination);

	consumer.Listener += OnMessage;

	SendMessages(destination);
	session.Commit();

	connection.Start();

	AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

	this.beforeRollback.AddRange(received);
	this.received.Clear();
	session.Rollback();

	AssertReceived(MSG_COUNT, "Post-Rollback expects to receive: " + MSG_COUNT + " messages.");

	this.afterRollback.AddRange(received);
	this.received.Clear();

	Assert.AreEqual(this.beforeRollback.Count, this.afterRollback.Count);
	Assert.AreEqual(this.beforeRollback, this.afterRollback);
	session.Commit();
	}

	[Test]
	public void testMessageDeleiveredInCorrectOrder()
	{
	session = connection.CreateSession(AcknowledgementMode.Transactional);
	IDestination destination = session.CreateTemporaryQueue();
	IMessageConsumer consumer = session.CreateConsumer(destination);

	consumer.Listener += OnMessage;

	SendMessages(destination);

	session.Commit();
	connection.Start();

	AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

	beforeRollback.AddRange(received);
	received.Clear();
	session.Rollback();

	AssertReceived(MSG_COUNT, "Post-Rollback expects to receive: " + MSG_COUNT + " messages.");

	afterRollback.AddRange(received);
	received.Clear();

	Assert.AreEqual(beforeRollback.Count, afterRollback.Count);
	Assert.AreEqual(beforeRollback, afterRollback);

	IEnumerator<IMessage> after = afterRollback.GetEnumerator();
	IEnumerator<IMessage> before = beforeRollback.GetEnumerator();

	while (after.MoveNext() && before.MoveNext())
	{
	ITextMessage original = before.Current as ITextMessage;
	ITextMessage rolledBack = after.Current as ITextMessage;

	int originalId = Int32.Parse(original.Text);
	int rolledBackId = Int32.Parse (rolledBack.Text);

	Assert.AreEqual(originalId, rolledBackId);
	}

	session.Commit();
	}

	[Test]
	public void testMessageDeleiveryDoesntStop()
	{
	session = connection.CreateSession(AcknowledgementMode.Transactional);
	IDestination destination = session.CreateTemporaryQueue();
	IMessageConsumer consumer = session.CreateConsumer(destination);

	consumer.Listener += OnMessage;

	SendMessages(destination);
	connection.Start();

	AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

	beforeRollback.AddRange(received);
	received.Clear();
	session.Rollback();

	SendMessages(destination);

	AssertReceived(MSG_COUNT * 2, "Post-Rollback expects to receive: " + MSG_COUNT * 2 + " messages.");

	afterRollback.AddRange(received);
	received.Clear();

	Assert.AreEqual(beforeRollback.Count * 2, afterRollback.Count);

	session.Commit();
	}

	[Test]
	public void testNonBlockingMessageDeleiveryIsDelayed()
	{
	connection.RedeliveryPolicy.InitialRedeliveryDelay = 7000;
	session = connection.CreateSession(AcknowledgementMode.Transactional);
	IDestination destination = session.CreateTemporaryQueue();
	IMessageConsumer consumer = session.CreateConsumer(destination);

	consumer.Listener += OnMessage;

	SendMessages(destination);
	connection.Start();

	AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

	received.Clear();
	session.Rollback();

	Thread.Sleep(4000);
	Assert.IsFalse(this.received.Count > 0, "Delayed redelivery test not expecting any messages yet.");

	AssertReceived(MSG_COUNT, "Post-Rollback expects to receive: " + MSG_COUNT + " messages.");

	session.Commit();
	session.Close();
	}

	public void OnMessageWithSomeRollbacks(IMessage message)
	{
	if (++count > 10)
	{
	try
	{
	session.Rollback();
	Tracer.Info("Rolling back session.");
	count = 0;
	}
	catch (Exception e)
	{
	Tracer.WarnFormat("Caught an unexcepted exception: {0}", e.Message);
	}
	}
	else
	{
	received.Add(message);
	try
	{
	session.Commit();
	}
	catch (Exception e)
	{
	Tracer.WarnFormat("Caught an unexcepted exception: {0}", e.Message);
	}
	}
	}

	[Test]
	public void testNonBlockingMessageDeleiveryWithRollbacks()
	{
	session = connection.CreateSession(AcknowledgementMode.Transactional);
	IDestination destination = session.CreateTemporaryQueue();
	IMessageConsumer consumer = session.CreateConsumer(destination);

	consumer.Listener += OnMessage;

	SendMessages(destination);
	connection.Start();

	AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

	received.Clear();

	consumer.Listener -= OnMessage;
	consumer.Listener += OnMessageWithSomeRollbacks;

	session.Rollback();

	AssertReceived(MSG_COUNT, "Post-Rollback expects to receive: " + MSG_COUNT + " messages.");

	Assert.AreEqual(MSG_COUNT, received.Count);
	session.Commit();
	}

	private void OnDLQMessage(IMessage message)
	{
	Tracer.DebugFormat("DLQ Message {0}", message);
	dlqed.Add(message);
	}

	private void OnMessageAlwaysRollsBack(IMessage message)
	{
	session.Rollback();
	}

	[Test]
	public void testNonBlockingMessageDeleiveryWithAllRolledBack()
	{
	connection.RedeliveryPolicy.MaximumRedeliveries = 3;
	session = connection.CreateSession(AcknowledgementMode.Transactional);
	IDestination destination = session.CreateTemporaryQueue();
	IDestination dlq = session.GetQueue("ActiveMQ.DLQ");
	IMessageConsumer consumer = session.CreateConsumer(destination);
	IMessageConsumer dlqConsumer = session.CreateConsumer(dlq);

	dlqConsumer.Listener += OnDLQMessage;
	consumer.Listener += OnMessage;

	SendMessages(destination);
	connection.Start();

	AssertReceived(MSG_COUNT, "Pre-Rollback expects to receive: " + MSG_COUNT + " messages.");

	consumer.Listener -= OnMessage;
	consumer.Listener += OnMessageAlwaysRollsBack;

	session.Rollback();

	AssertReceived(dlqed, MSG_COUNT, "Post-Rollback expects to DLQ: " + MSG_COUNT + " messages.");

	session.Commit();
	}

	private void DeleteDLQ()
	{
	session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
	IDestination dlq = session.GetQueue("ActiveMQ.DLQ");
	try
	{
	connection.DeleteDestination(dlq);
	}
	catch
	{
	}
	}

	private void SendMessages(IDestination destination)
	{
	IConnection connection = CreateConnection();
	ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
	IMessageProducer producer = session.CreateProducer(destination);
	for(int i = 0; i < MSG_COUNT; ++i)
	{
	producer.Send(session.CreateTextMessage("" + i));
	}
	connection.Close();
	}
	}
	}