test/NMSConsumerTest.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.Threading;
 using Apache.NMS.Test;
 using NUnit.Framework;
 using Apache.NMS.ActiveMQ.Commands;
 using System;
 using Apache.NMS.Util;

 namespace Apache.NMS.ActiveMQ.Test
 {
     [TestFixture]
     public class NMSConsumerTest : NMSTestSupport
     {
         protected static string DestinationName => "TEST" + TestContext.CurrentContext.Test.Name;

         protected static string TEST_CLIENT_ID = "MessageConsumerTestClientId";

         private CountDownLatch doneLatch;
         private int counter;
         private String errorMessage;

         [SetUp]
         public override void SetUp()
         {
             base.SetUp();
             this.doneLatch = new CountDownLatch(1);
             this.counter = 0;
             this.errorMessage = null;
         }

         [Test]
         public void TestBadSelectorDoesNotCloseConnection()
         {
             using (var context = CreateContext(TEST_CLIENT_ID))
             {
                 context.Start();
                 IDestination destination = context.CreateTemporaryQueue();

                 var producer = context.CreateProducer();
                 ITextMessage goodMsg = context.CreateTextMessage("testGood");
                 producer.Send(destination, goodMsg);

                 var consumer = context.CreateConsumer(destination);
                 Assert.NotNull(consumer.Receive(TimeSpan.FromMilliseconds(5000)));

                 try
                 {
                     context.CreateConsumer(destination, "badSelector;too");
                     Assert.Fail("Exception expected.");
                 }
                 catch (Exception e)
                 {
                     Tracer.DebugFormat("Caught Ex: {0}", e);
                 }

                 ITextMessage failMsg = context.CreateTextMessage("testFail");
                 producer.Send(destination, failMsg);
                 Assert.NotNull(consumer.Receive(TimeSpan.FromMilliseconds(5000)));
             }
         }

         [Test][Timeout(20_000)]
         public void TestAsyncDispatchExceptionRedelivers()
         {
             using (var context = CreateContext(TEST_CLIENT_ID))
             {
                 IQueue queue = context.GetQueue(DestinationName);

                 using (var producer = context.CreateProducer())
                 {
                     producer.DeliveryMode = MsgDeliveryMode.NonPersistent;
                     producer.Send(queue, producer.CreateTextMessage("First"));
                     producer.Send(queue, producer.CreateTextMessage("Second"));
                 }

                 using (var consumer = context.CreateConsumer(queue))
                 {
                     consumer.Listener += OnTestAsynchRedliversMessage;
                     context.Start();

                     if (doneLatch.await(TimeSpan.FromSeconds(10)))
                     {
                         if (!String.IsNullOrEmpty(errorMessage))
                         {
                             Assert.Fail(errorMessage);
                         }
                     }
                     else
                     {
                         Assert.Fail("Timeout waiting for async message delivery to complete.");
                     }
                 }
             }
         }

         private void OnTestAsynchRedliversMessage(IMessage msg)
         {
             counter++;
             try
             {
                 ITextMessage message = msg as ITextMessage;
                 switch (counter)
                 {
                     case 1:
                         Tracer.Debug("Got first Message: " + message.Text);
                         Assert.AreEqual("First", message.Text);
                         Assert.IsFalse(message.NMSRedelivered);
                         break;
                     case 2:
                         Tracer.Debug("Got Second Message: " + message.Text);
                         Assert.AreEqual("Second", message.Text);
                         Assert.IsFalse(message.NMSRedelivered);
                         throw new Exception("Ignore Me");
                     case 3:
                         Tracer.Debug("Got Third Message: " + message.Text);
                         Assert.AreEqual("Second", message.Text);
                         Assert.IsTrue(message.NMSRedelivered);
                         doneLatch.countDown();
                         break;
                     default:
                         errorMessage = "Got too many messages: " + counter;
                         Tracer.Debug(errorMessage);
                         doneLatch.countDown();
                         break;
                 }
             }
             catch (Exception e)
             {
                 if (e.Message.Equals("Ignore Me"))
                 {
                     throw;
                 }

                 errorMessage = "Got exception: " + e.Message;
                 Tracer.Warn("Exception on Message Receive: " + e.Message);
                 doneLatch.countDown();
             }
         }

         [Test]
         public void ConsumeInTwoThreads()
         {
             ParameterizedThreadStart threadStart =
                 delegate(object o)
                 {
                     var consumer = (INMSConsumer)o;
                     IMessage message = consumer.Receive(TimeSpan.FromSeconds(2));
                     Assert.IsNotNull(message);
                 };

             using (var context = CreateContext(TEST_CLIENT_ID, AcknowledgementMode.Transactional))
             {
                 context.Start();

                 IQueue queue = context.GetQueue(DestinationName);

                 // enqueue 2 messages
                 using (var consumer = context.CreateConsumer(queue))
                 using (var producer = context.CreateProducer())
                 {
                     producer.DeliveryMode = MsgDeliveryMode.Persistent;
                     producer.Send(queue, producer.CreateMessage());
                     producer.Send(queue, producer.CreateMessage());
                     context.Commit();

                     // receive first using a dedicated thread. This works
                     Thread thread = new Thread(threadStart);
                     thread.Start(consumer);
                     thread.Join();
                     context.Commit();

                     // receive second using main thread. This FAILS
                     IMessage message =
                         consumer.Receive(TimeSpan
                             .FromSeconds(2)); // throws System.Threading.AbandonedMutexException
                     Assert.IsNotNull(message);
                     context.Commit();
                 }
             }
         }

         [Test][Timeout(20_000)]
         public void TestReceiveIgnoreExpirationMessage(
             [Values(AcknowledgementMode.AutoAcknowledge, AcknowledgementMode.ClientAcknowledge,
                 AcknowledgementMode.DupsOkAcknowledge, AcknowledgementMode.Transactional)]
             AcknowledgementMode ackMode,
             [Values(MsgDeliveryMode.NonPersistent, MsgDeliveryMode.Persistent)]
             MsgDeliveryMode deliveryMode,
             [Values(ExpirationOptions.DEFAULT, ExpirationOptions.IGNORE, ExpirationOptions.DO_NOT_IGNORE)]
             ExpirationOptions expirationOption)
         {
             using (var context = CreateContext(TEST_CLIENT_ID, ackMode))
             {
                 context.Start();
                 // using (Session context = connection.CreateSession(ackMode) as Session)
                 {
                     string destinationName = DestinationName;

                     if (ExpirationOptions.IGNORE == expirationOption)
                     {
                         destinationName += "?consumer.nms.ignoreExpiration=true";
                     }
                     else if (ExpirationOptions.DO_NOT_IGNORE == expirationOption)
                     {
                         destinationName += "?consumer.nms.ignoreExpiration=false";
                     }

                     try
                     {
                         IDestination destination = context.GetQueue(destinationName);

                         using (var consumer = context.CreateConsumer(destination))
                         using (var producer = context.CreateProducer())
                         {
                             producer.DeliveryMode = deliveryMode;

                             string msgText = string.Format("ExpiredMessage: {0}", Guid.NewGuid().ToString());

                             ActiveMQTextMessage msg = context.CreateTextMessage(msgText) as ActiveMQTextMessage;

                             // Give it two seconds to live.
                             msg.NMSTimeToLive = TimeSpan.FromMilliseconds(2000);

                             producer.Send(destination, msg);

                             if (AcknowledgementMode.Transactional == ackMode)
                             {
                                 context.Commit();
                             }

                             // Wait for four seconds before processing it.  The broker will have sent it to our local
                             // client dispatch queue, but we won't attempt to process the message until it has had
                             // a chance to expire within our internal queue system.
                             Thread.Sleep(4000);

                             ActiveMQTextMessage rcvMsg = consumer.ReceiveNoWait() as ActiveMQTextMessage;

                             if (ExpirationOptions.IGNORE == expirationOption)
                             {
                                 Assert.IsNotNull(rcvMsg, "Did not receive expired message.");
                                 rcvMsg.Acknowledge();

                                 Assert.AreEqual(msgText, rcvMsg.Text, "Message text does not match.");
                                 Assert.IsTrue(rcvMsg.IsExpired());

                                 if (AcknowledgementMode.Transactional == ackMode)
                                 {
                                     context.Commit();
                                 }
                             }
                             else
                             {
                                 // Should not receive a message.
                                 Assert.IsNull(rcvMsg, "Received an expired message!");
                             }

                             consumer.Close();
                             producer.Close();
                         }
                     }
                     finally
                     {
                         try
                         {
                             // Ensure that Session resources on the Broker release transacted Consumers.
                             context.Close();
                             // Give the Broker some time to remove the subscriptions.
                             Thread.Sleep(2000);
                         }
                         catch
                         {
                         }
                     }
                 }
             }
         }

         public virtual INMSContext CreateContext() => this.CreateContext(null);

         public virtual INMSContext CreateContext(string newClientId,
             AcknowledgementMode ackMode = AcknowledgementMode.AutoAcknowledge)
         {
             var context = this.Factory.CreateContext(this.userName, this.passWord, ackMode);
             Assert.IsNotNull((object)context, "connection not created");
             if (newClientId != null)
                 context.ClientId = newClientId;
             return context;
         }
     }
 }
	/*
	* 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.Threading;
	using Apache.NMS.Test;
	using NUnit.Framework;
	using Apache.NMS.ActiveMQ.Commands;
	using System;
	using Apache.NMS.Util;

	namespace Apache.NMS.ActiveMQ.Test
	{
	[TestFixture]
	public class NMSConsumerTest : NMSTestSupport
	{
	protected static string DestinationName => "TEST" + TestContext.CurrentContext.Test.Name;

	protected static string TEST_CLIENT_ID = "MessageConsumerTestClientId";

	private CountDownLatch doneLatch;
	private int counter;
	private String errorMessage;

	[SetUp]
	public override void SetUp()
	{
	base.SetUp();
	this.doneLatch = new CountDownLatch(1);
	this.counter = 0;
	this.errorMessage = null;
	}

	[Test]
	public void TestBadSelectorDoesNotCloseConnection()
	{
	using (var context = CreateContext(TEST_CLIENT_ID))
	{
	context.Start();
	IDestination destination = context.CreateTemporaryQueue();

	var producer = context.CreateProducer();
	ITextMessage goodMsg = context.CreateTextMessage("testGood");
	producer.Send(destination, goodMsg);

	var consumer = context.CreateConsumer(destination);
	Assert.NotNull(consumer.Receive(TimeSpan.FromMilliseconds(5000)));

	try
	{
	context.CreateConsumer(destination, "badSelector;too");
	Assert.Fail("Exception expected.");
	}
	catch (Exception e)
	{
	Tracer.DebugFormat("Caught Ex: {0}", e);
	}

	ITextMessage failMsg = context.CreateTextMessage("testFail");
	producer.Send(destination, failMsg);
	Assert.NotNull(consumer.Receive(TimeSpan.FromMilliseconds(5000)));
	}
	}

	[Test][Timeout(20_000)]
	public void TestAsyncDispatchExceptionRedelivers()
	{
	using (var context = CreateContext(TEST_CLIENT_ID))
	{
	IQueue queue = context.GetQueue(DestinationName);

	using (var producer = context.CreateProducer())
	{
	producer.DeliveryMode = MsgDeliveryMode.NonPersistent;
	producer.Send(queue, producer.CreateTextMessage("First"));
	producer.Send(queue, producer.CreateTextMessage("Second"));
	}

	using (var consumer = context.CreateConsumer(queue))
	{
	consumer.Listener += OnTestAsynchRedliversMessage;
	context.Start();

	if (doneLatch.await(TimeSpan.FromSeconds(10)))
	{
	if (!String.IsNullOrEmpty(errorMessage))
	{
	Assert.Fail(errorMessage);
	}
	}
	else
	{
	Assert.Fail("Timeout waiting for async message delivery to complete.");
	}
	}
	}
	}

	private void OnTestAsynchRedliversMessage(IMessage msg)
	{
	counter++;
	try
	{
	ITextMessage message = msg as ITextMessage;
	switch (counter)
	{
	case 1:
	Tracer.Debug("Got first Message: " + message.Text);
	Assert.AreEqual("First", message.Text);
	Assert.IsFalse(message.NMSRedelivered);
	break;
	case 2:
	Tracer.Debug("Got Second Message: " + message.Text);
	Assert.AreEqual("Second", message.Text);
	Assert.IsFalse(message.NMSRedelivered);
	throw new Exception("Ignore Me");
	case 3:
	Tracer.Debug("Got Third Message: " + message.Text);
	Assert.AreEqual("Second", message.Text);
	Assert.IsTrue(message.NMSRedelivered);
	doneLatch.countDown();
	break;
	default:
	errorMessage = "Got too many messages: " + counter;
	Tracer.Debug(errorMessage);
	doneLatch.countDown();
	break;
	}
	}
	catch (Exception e)
	{
	if (e.Message.Equals("Ignore Me"))
	{
	throw;
	}

	errorMessage = "Got exception: " + e.Message;
	Tracer.Warn("Exception on Message Receive: " + e.Message);
	doneLatch.countDown();
	}
	}

	[Test]
	public void ConsumeInTwoThreads()
	{
	ParameterizedThreadStart threadStart =
	delegate(object o)
	{
	var consumer = (INMSConsumer)o;
	IMessage message = consumer.Receive(TimeSpan.FromSeconds(2));
	Assert.IsNotNull(message);
	};

	using (var context = CreateContext(TEST_CLIENT_ID, AcknowledgementMode.Transactional))
	{
	context.Start();

	IQueue queue = context.GetQueue(DestinationName);

	// enqueue 2 messages
	using (var consumer = context.CreateConsumer(queue))
	using (var producer = context.CreateProducer())
	{
	producer.DeliveryMode = MsgDeliveryMode.Persistent;
	producer.Send(queue, producer.CreateMessage());
	producer.Send(queue, producer.CreateMessage());
	context.Commit();

	// receive first using a dedicated thread. This works
	Thread thread = new Thread(threadStart);
	thread.Start(consumer);
	thread.Join();
	context.Commit();

	// receive second using main thread. This FAILS
	IMessage message =
	consumer.Receive(TimeSpan
	.FromSeconds(2)); // throws System.Threading.AbandonedMutexException
	Assert.IsNotNull(message);
	context.Commit();
	}
	}
	}

	[Test][Timeout(20_000)]
	public void TestReceiveIgnoreExpirationMessage(
	[Values(AcknowledgementMode.AutoAcknowledge, AcknowledgementMode.ClientAcknowledge,
	AcknowledgementMode.DupsOkAcknowledge, AcknowledgementMode.Transactional)]
	AcknowledgementMode ackMode,
	[Values(MsgDeliveryMode.NonPersistent, MsgDeliveryMode.Persistent)]
	MsgDeliveryMode deliveryMode,
	[Values(ExpirationOptions.DEFAULT, ExpirationOptions.IGNORE, ExpirationOptions.DO_NOT_IGNORE)]
	ExpirationOptions expirationOption)
	{
	using (var context = CreateContext(TEST_CLIENT_ID, ackMode))
	{
	context.Start();
	// using (Session context = connection.CreateSession(ackMode) as Session)
	{
	string destinationName = DestinationName;

	if (ExpirationOptions.IGNORE == expirationOption)
	{
	destinationName += "?consumer.nms.ignoreExpiration=true";
	}
	else if (ExpirationOptions.DO_NOT_IGNORE == expirationOption)
	{
	destinationName += "?consumer.nms.ignoreExpiration=false";
	}

	try
	{
	IDestination destination = context.GetQueue(destinationName);

	using (var consumer = context.CreateConsumer(destination))
	using (var producer = context.CreateProducer())
	{
	producer.DeliveryMode = deliveryMode;

	string msgText = string.Format("ExpiredMessage: {0}", Guid.NewGuid().ToString());

	ActiveMQTextMessage msg = context.CreateTextMessage(msgText) as ActiveMQTextMessage;

	// Give it two seconds to live.
	msg.NMSTimeToLive = TimeSpan.FromMilliseconds(2000);

	producer.Send(destination, msg);

	if (AcknowledgementMode.Transactional == ackMode)
	{
	context.Commit();
	}

	// Wait for four seconds before processing it. The broker will have sent it to our local
	// client dispatch queue, but we won't attempt to process the message until it has had
	// a chance to expire within our internal queue system.
	Thread.Sleep(4000);

	ActiveMQTextMessage rcvMsg = consumer.ReceiveNoWait() as ActiveMQTextMessage;

	if (ExpirationOptions.IGNORE == expirationOption)
	{
	Assert.IsNotNull(rcvMsg, "Did not receive expired message.");
	rcvMsg.Acknowledge();

	Assert.AreEqual(msgText, rcvMsg.Text, "Message text does not match.");
	Assert.IsTrue(rcvMsg.IsExpired());

	if (AcknowledgementMode.Transactional == ackMode)
	{
	context.Commit();
	}
	}
	else
	{
	// Should not receive a message.
	Assert.IsNull(rcvMsg, "Received an expired message!");
	}

	consumer.Close();
	producer.Close();
	}
	}
	finally
	{
	try
	{
	// Ensure that Session resources on the Broker release transacted Consumers.
	context.Close();
	// Give the Broker some time to remove the subscriptions.
	Thread.Sleep(2000);
	}
	catch
	{
	}
	}
	}
	}
	}

	public virtual INMSContext CreateContext() => this.CreateContext(null);

	public virtual INMSContext CreateContext(string newClientId,
	AcknowledgementMode ackMode = AcknowledgementMode.AutoAcknowledge)
	{
	var context = this.Factory.CreateContext(this.userName, this.passWord, ackMode);
	Assert.IsNotNull((object)context, "connection not created");
	if (newClientId != null)
	context.ClientId = newClientId;
	return context;
	}
	}
	}