activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject svn commit: r1715303 - in /activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src: main/csharp/MessageConsumer.cs test/csharp/MessageListenerRedeliveryTest.cs
Date Thu, 19 Nov 2015 23:18:39 GMT
Author: tabish
Date: Thu Nov 19 23:18:39 2015
New Revision: 1715303

URL: http://svn.apache.org/viewvc?rev=1715303&view=rev
Log:
https://issues.apache.org/jira/browse/AMQNET-513

Preserve the rollback cause when poisoning a message.

Modified:
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/MessageConsumer.cs
    activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/MessageListenerRedeliveryTest.cs

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/MessageConsumer.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/MessageConsumer.cs?rev=1715303&r1=1715302&r2=1715303&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/MessageConsumer.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/MessageConsumer.cs
Thu Nov 19 23:18:39 2015
@@ -883,10 +883,10 @@ namespace Apache.NMS.ActiveMQ
                     }
                     catch(Exception e)
                     {
+                        dispatch.RollbackCause = e;
                         if(IsAutoAcknowledgeBatch || IsAutoAcknowledgeEach || IsIndividualAcknowledge)
                         {
                             // Schedule redelivery and possible dlq processing
-                            dispatch.RollbackCause = e;
                             Rollback();
                         }
                         else
@@ -1094,9 +1094,9 @@ namespace Apache.NMS.ActiveMQ
             }
         }
 
-        private bool ConsumeExpiredMessage(MessageDispatch dispatch) 
+        private bool ConsumeExpiredMessage(MessageDispatch dispatch)
         {
-            if (dispatch.Message.IsExpired()) 
+            if (dispatch.Message.IsExpired())
             {
                 return !info.Browser && !IgnoreExpiration;
             }
@@ -1480,7 +1480,9 @@ namespace Apache.NMS.ActiveMQ
                             cause.ExceptionClass = "javax.jms.JMSException";
                             cause.Message = lastMd.RollbackCause.Message;
                             poisonCause.Cause = cause;
+                            poisonCause.Message = poisonCause.Message + " cause: " + lastMd.RollbackCause.Message;
                         }
+
                         ack.FirstMessageId = firstMsgId;
                         ack.PoisonCause = poisonCause;
 

Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/MessageListenerRedeliveryTest.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/MessageListenerRedeliveryTest.cs?rev=1715303&r1=1715302&r2=1715303&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/MessageListenerRedeliveryTest.cs
(original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/MessageListenerRedeliveryTest.cs
Thu Nov 19 23:18:39 2015
@@ -15,7 +15,11 @@
  * 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;
@@ -26,15 +30,29 @@ namespace Apache.NMS.ActiveMQ.Test
     public class MessageListenerRedeliveryTest : NMSTestSupport
     {
         private Connection connection;
-        private int counter;
+        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;
         }
 
@@ -76,6 +94,39 @@ namespace Apache.NMS.ActiveMQ.Test
             }
         }
 
+        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() 
         {
@@ -91,7 +142,7 @@ namespace Apache.NMS.ActiveMQ.Test
             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
@@ -125,6 +176,191 @@ namespace Apache.NMS.ActiveMQ.Test
     
             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;
+        }
     }
 }



Mime
View raw message