/*
 * 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.Specialized;
using System.Collections.Generic;
using NUnit.Framework;
using NUnit.Framework.Interfaces;
using Apache.NMS;
using Apache.NMS.AMQP.Test.Util;
using Apache.NMS.AMQP.Test.Attribute;
using System.Text;

namespace Apache.NMS.AMQP.Test.TestCase
{
    [TestFixture]
    class ConsumerTest : BaseTestCase
    {
        

        const string CUSTOM_CLIENT_ID = "foobar";

        public override void Setup()
        {
            base.Setup();
            msgCount = 0;
            asyncEx = null;
            waiter = new System.Threading.ManualResetEvent(false);
        }


        protected void SendMessages(IMessageProducer producer, int messageCount)
        {
            ITextMessage message = producer.CreateTextMessage();

            for (int i = 0; i < messageCount; i++)
            {
                message.Text = "num:" + i;
                producer.Send(message, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.Zero);
            }
        }

        [Test]
        [ConnectionSetup(null, "default")]
        [SessionSetup("default", "default", AckMode = AcknowledgementMode.AutoAcknowledge)]
        [TopicSetup("default", "testdest1", Name = "nms.test")]
        [ConsumerSetup("default", "testdest1", "con1")]
        [ProducerSetup("default", "testdest1", "pro1")]
        public void TestReceiveMessageSync()
        {
            const int NUM_MSGS = 100;
            const int CONUSMER_TIMEOUT = 25; // 25 ms to pull from consumer.
            const int BACKOFF_SLEEP = 500; // 0.5 secs to sleep when consumer pull times out.
            using (IConnection connection = this.GetConnection())
            using (IMessageConsumer consumer = this.GetConsumer("con1"))
            using (IMessageProducer producer = this.GetProducer("pro1"))
            {
                try
                {
                    connection.ExceptionListener += DefaultExceptionListener;
                    connection.Start();

                    SendMessages(producer, NUM_MSGS);

                    ITextMessage textMessage = null;
                    IMessage message = null;
                    int backoffCount = 0;
                    DateTime start = DateTime.Now;
                    TimeSpan duration;
                    while (msgCount < NUM_MSGS && backoffCount < 3)
                    {
                        while ((message = consumer.Receive(TimeSpan.FromMilliseconds(CONUSMER_TIMEOUT))) != null)
                        {
                            textMessage = message as ITextMessage;
                            Assert.AreEqual("num:" + msgCount, textMessage.Text, "Received message out of order.");
                            msgCount++;
                        }
                        backoffCount++;
                        if (backoffCount < 3 && msgCount < NUM_MSGS)
                        {
                            System.Threading.Thread.Sleep(BACKOFF_SLEEP);
                        }
                    }

                    duration = (DateTime.Now - start);
                    int milis = Convert.ToInt32(duration.TotalMilliseconds);
                    Assert.AreEqual(NUM_MSGS, msgCount, "Received {0} messages out of {1} in {0}ms.", msgCount, NUM_MSGS, milis);

                }
                catch (Exception ex)
                {
                    this.PrintTestFailureAndAssert(this.GetMethodName(), "Unexpected Exception.", ex);
                }
            }
        }

        [Test]
        [ConnectionSetup(null, "default")]
        [SessionSetup("default", "default", AckMode = AcknowledgementMode.AutoAcknowledge)]
        [TopicSetup("default", "testdest1", Name = "nms.test")]
        [ConsumerSetup("default", "testdest1", "con1")]
        [ProducerSetup("default", "testdest1", "pro1", DeliveryMode = MsgDeliveryMode.NonPersistent)]
        public void TestReceiveStopStartMessageAsync()
        {
            const int NUM_MSGS = 100;
            const int MESSAGE_RECEIVE_TIMEOUT = 1000; // 1.0s

            //Apache.NMS.Util.Atomic<bool> stopped = new Apache.NMS.Util.Atomic<bool>(false);

            using (IConnection connection = this.GetConnection())
            using (IMessageConsumer consumer = this.GetConsumer("con1"))
            using (IMessageProducer producer = this.GetProducer("pro1"))
            {
                try
                {
                    MessageListener ackcallback = CreateListener(NUM_MSGS);
                    MessageListener callback = (m) =>
                    {
                        if (!connection.IsStarted)
                        {
                            waiter.Set();
                            throw new Exception("Received Message " + msgCount + " on stopped connection.");
                        }
                        ackcallback(m);
                        Logger.Info("MsgCount : " + msgCount);
                    };

                    consumer.Listener += callback;
                    connection.ExceptionListener += DefaultExceptionListener;
                    connection.Start();

                    ITextMessage message = producer.CreateTextMessage();

                    for (int i = 0; i < NUM_MSGS; i++)
                    {
                        message.Text = "num:" + i;
                        producer.Send(message);
                        if (i == NUM_MSGS / 2)
                        {
                            connection.Stop();
                            Logger.Info("Stopping Connection. At message " + i + ".");
                        }
                        else if (i == NUM_MSGS - 1)
                        {
                            Logger.Info("Starting Connection. At message " + i + ".");
                            connection.Start();

                        }
                    }

                    if (!waiter.WaitOne(MESSAGE_RECEIVE_TIMEOUT))
                    {
                        Assert.IsNull(asyncEx, "OnExceptionListener Event when raised. Message : {0}.", asyncEx?.Message);
                        Assert.Fail("Received {0} of {1} messages in {2}ms.", msgCount, NUM_MSGS, MESSAGE_RECEIVE_TIMEOUT);
                    }

                    Assert.IsNull(asyncEx, "OnExceptionListener Event when raised. Message : {0}.", asyncEx?.Message);
                    Assert.AreEqual(NUM_MSGS, msgCount, "Failed to Received All Messages sent.");

                }
                catch (Exception ex)
                {
                    Logger.Warn("Async execption: " + this.GetTestException(asyncEx));
                    this.PrintTestFailureAndAssert(this.GetMethodName(), "Unexpected Exception.", ex);
                }

            }
        }

        [Test]
        //[Repeat(1000)]
        [ConnectionSetup(null, "default")]
        [SessionSetup("default", "default", AckMode = AcknowledgementMode.ClientAcknowledge)]
        [TopicSetup("default", "testdest1", Name = "nms.test")]
        [ConsumerSetup("default", "testdest1", "con1")]
        [ProducerSetup("default", "testdest1", "pro1", DeliveryMode = MsgDeliveryMode.Persistent)]
        public void TestReceiveMessageAsyncClientAck()
        {
            const int NUM_MSGS = 100;
            const int MESSAGE_RECEIVE_TIMEOUT = 1000; // 1.0s

            //Apache.NMS.Util.Atomic<bool> stopped = new Apache.NMS.Util.Atomic<bool>(false);

            using (IConnection connection = this.GetConnection())
            using (ISession session = this.GetSession("default"))
            using (IMessageConsumer consumer = this.GetConsumer("con1"))
            using (IMessageProducer producer = this.GetProducer("pro1"))
            {
                try
                {
                    Assert.AreEqual(AcknowledgementMode.ClientAcknowledge, session.AcknowledgementMode, "Session acknowkedgement mode is not Client mode.");
                    MessageListener ackcallback = CreateListener(NUM_MSGS);
                    MessageListener callback = (m) =>
                    {
                        if (!connection.IsStarted)
                        {
                            waiter.Set();
                            throw new Exception("Received Message " + msgCount + " on stopped connection.");
                        }
                        m.Properties["NMS.AMQP.ACK.TYPE"] = 0; // AMQP Accepted
                        m.Acknowledge();
                        ackcallback(m);
                        Logger.Info("MsgCount : " + msgCount);
                    };

                    consumer.Listener += callback;
                    connection.ExceptionListener += DefaultExceptionListener;
                    connection.Start();

                    ITextMessage message = producer.CreateTextMessage();

                    for (int i = 0; i < NUM_MSGS; i++)
                    {
                        message.Text = "num:" + i;
                        producer.Send(message, 
                            MsgDeliveryMode.NonPersistent, 
                            MsgPriority.Normal,
                            TimeSpan.Zero);
                    }

                    if (!waiter.WaitOne(MESSAGE_RECEIVE_TIMEOUT))
                    {
                        Assert.IsNull(asyncEx, "OnExceptionListener Event when raised. Message : {0}.", asyncEx?.Message);
                        Assert.Fail("Received {0} of {1} messages in {2}ms.", msgCount, NUM_MSGS, MESSAGE_RECEIVE_TIMEOUT);
                    }

                    Assert.IsNull(asyncEx, "OnExceptionListener Event when raised. Message : {0}.", asyncEx?.Message);
                    Assert.AreEqual(NUM_MSGS, msgCount, "Failed to Received All Messages sent.");

                }
                catch (Exception ex)
                {
                    if (asyncEx != null)
                        Logger.Warn("Async execption: " + this.GetTestException(asyncEx));
                    this.PrintTestFailureAndAssert(this.GetMethodName(), "Unexpected Exception.", ex);
                }

            }
        }

        [Test]
        //[Repeat(1000)]
        [ConnectionSetup(null, "default")]
        [SessionSetup("default", "default", AckMode = AcknowledgementMode.AutoAcknowledge)]
        [TopicSetup("default", "testdest1", Name = "nms.test")]
        [ConsumerSetup("default", "testdest1", "con1")]
        [ProducerSetup("default", "testdest1", "pro1", DeliveryMode = MsgDeliveryMode.NonPersistent)]
        public void TestReceiveMessageAsync()
        {
            const int NUM_MSGS = 100;
            const int MESSAGE_RECEIVE_TIMEOUT = 1000; // 1.0s

            //Apache.NMS.Util.Atomic<bool> stopped = new Apache.NMS.Util.Atomic<bool>(false);

            using (IConnection connection = this.GetConnection())
            using (IMessageConsumer consumer = this.GetConsumer("con1"))
            using (IMessageProducer producer = this.GetProducer("pro1"))
            {
                try
                {
                    MessageListener ackcallback = CreateListener(NUM_MSGS);
                    MessageListener callback = (m) =>
                    {
                        if (!connection.IsStarted)
                        {
                            waiter.Set();
                            throw new Exception("Received Message " + msgCount + " on stopped connection.");
                        }
                        ackcallback(m);
                        Logger.Info("MsgCount : " + msgCount);
                    };

                    consumer.Listener += callback;
                    connection.ExceptionListener += DefaultExceptionListener;
                    connection.Start();

                    ITextMessage message = producer.CreateTextMessage();

                    for (int i = 0; i < NUM_MSGS; i++)
                    {
                        message.Text = "num:" + i;
                        producer.Send(message);
                    }

                    if (!waiter.WaitOne(MESSAGE_RECEIVE_TIMEOUT))
                    {
                        Assert.IsNull(asyncEx, "OnExceptionListener Event when raised. Message : {0}.", asyncEx?.Message);
                        Assert.Fail("Received {0} of {1} messages in {2}ms.", msgCount, NUM_MSGS, MESSAGE_RECEIVE_TIMEOUT);
                    }

                    Assert.IsNull(asyncEx, "OnExceptionListener Event when raised. Message : {0}.", asyncEx?.Message);
                    Assert.AreEqual(NUM_MSGS, msgCount, "Failed to Received All Messages sent.");

                }
                catch (Exception ex)
                {
                    Logger.Warn("Async execption: " + this.GetTestException(asyncEx));
                    this.PrintTestFailureAndAssert(this.GetMethodName(), "Unexpected Exception.", ex);
                }

            }
        }

        private void LogSummary(long[] missingCount, long[] lastId, long Msgbatch)
        {
            long missingMsgs = 0;
            long undeliveredMsgs = 0;

            for (int i = 0; i < missingCount.Length; i++)
            {
                missingMsgs += missingCount[i];
                undeliveredMsgs += Msgbatch - lastId[i] - 1;
            }

            long TotalMsgsDelivered = msgCount + missingMsgs + undeliveredMsgs;
            Logger.Warn(string.Format("Msgs Received: {0}, Msgs Lost: {1}, Msgs Remaining Delivery: {2} TotalMsgs: {3}",
                msgCount, missingMsgs, undeliveredMsgs, TotalMsgsDelivered));
        }

        [Test]
        [ConnectionSetup(null, "default")]
        [SessionSetup("default", "default", AckMode = AcknowledgementMode.AutoAcknowledge)]
        [TopicSetup("default", "testdest1", Name = "nms.test.1")]
        [TopicSetup("default", "testdest2", Name = "nms.test.2")]
        [TopicSetup("default", "testdest3", Name = "nms.test.3")]
        [TopicSetup("default", "testdest4", Name = "nms.test.4")]
        [ConsumerSetup("default", "testdest1", "con1")]
        [ConsumerSetup("default", "testdest2", "con2")]
        [ConsumerSetup("default", "testdest3", "con3")]
        [ConsumerSetup("default", "testdest4", "con4")]//*/
        [ProducerSetup("default", "testdest1", "pro1")]
        [ProducerSetup("default", "testdest2", "pro2")]
        [ProducerSetup("default", "testdest3", "pro3")]
        [ProducerSetup("default", "testdest4", "pro4")]//*/
        public void TestMultipleConsumerMessageListenerReceive()
        {
            const int NUM_MSGS = 100000;

            string[] conIds = new string[] { "con1", "con2", "con3", "con4" };
            string[] proIds = new string[] { "pro1", "pro2", "pro3", "pro4" };
            IList<IMessageConsumer> consumers = this.GetConsumers(conIds);
            IList<IMessageProducer> producers = this.GetProducers(proIds);
            int ProducerCount = producers.Count;
            int ConsumerCount = consumers.Count;
            long[] lastMsgId = new long[ConsumerCount];
            long[] MissingMsgCount = new long[ConsumerCount];
            long lostMsgCount = 0;
            int TotalMsgs = NUM_MSGS * ProducerCount;
            long undeliveredMsgCount = TotalMsgs;
            int timeout = Math.Max(TotalMsgs / 1000, 1) * 1100;
            MsgDeliveryMode mode = MsgDeliveryMode.NonPersistent;

            using (IConnection connection = this.GetConnection("default"))
            {
                
                connection.ExceptionListener += DefaultExceptionListener;

                for (int i = 0; i < ConsumerCount; i++)
                {
                    lastMsgId[i] = -1;
                    MissingMsgCount[i] = 0;
                }
                try
                {
                    MessageListener callback = CreateListener(TotalMsgs);
                    int index = 0;
                    foreach (IMessageConsumer consumer in consumers)
                    {
                        int num = index;
                        MessageListener countingCallback = (m) =>
                        {
                            long lastId = lastMsgId[num];
                            long msgId = ExtractMsgId(m.NMSMessageId);
                            if (msgId > lastId)
                            {
                                lastMsgId[num] = msgId;
                                if (lastId != -1)
                                {
                                    MissingMsgCount[num] += (msgId - (lastId + 1));
                                    lostMsgCount += (msgId - (lastId + 1));

                                }
                            }
                            callback(m);

                            // signal envent waiter when the last expected msg is delivered on all consumers
                            if (lostMsgCount + msgCount == TotalMsgs)
                            {
                                // signal if detected lost msgs from id gap and delivered msgs make the total msgs.
                                waiter?.Set();
                            }
                            else if (lastMsgId[num] == NUM_MSGS - 1)
                            {
                                // signal if final msg id on every consumer is detected.
                                undeliveredMsgCount -= NUM_MSGS;
                                if (undeliveredMsgCount <= 0)
                                    waiter?.Set();
                            }
                        };

                        consumer.Listener += countingCallback;
                        index++;
                    }
                    
                    connection.Start();

                    // send messages to Destinations
                    ITextMessage sendMsg = producers[0].CreateTextMessage();
                    for (int i = 0; i < NUM_MSGS; i++)
                    {
                        int link = 0;
                        foreach (IMessageProducer producer in producers)
                        {
                            sendMsg.Text = "Link: " + link + ", num:" + i;
                            link++;
                            producer.Send(sendMsg, mode, MsgPriority.Normal, TimeSpan.Zero);
                        }
                    }

                    if (!waiter.WaitOne(timeout))
                    {
                        if (mode.Equals(MsgDeliveryMode.NonPersistent))
                        {
                            if (msgCount != TotalMsgs)
                            {
                                Logger.Warn(string.Format("Only received {0} of {1} messages in {2}ms.", msgCount, TotalMsgs, timeout));

                                LogSummary(MissingMsgCount, lastMsgId, NUM_MSGS);
                            }

                            Assert.IsNull(asyncEx, "Received unexpected asynchronous exception. Message : {0}", asyncEx?.Message);
                        }
                        else
                        {
                            Assert.Fail("Only received {0} of {1} messages in {2}ms.", msgCount, TotalMsgs, timeout);
                        }
                    }
                    else
                    {

                        if (msgCount != TotalMsgs)
                        {
                            Logger.Warn(string.Format("Only received {0} of {1} messages in {2}ms.", msgCount, TotalMsgs, timeout));

                            LogSummary(MissingMsgCount, lastMsgId, NUM_MSGS);
                        }

                        Assert.IsNull(asyncEx, "Received unexpected asynchronous exception. Message : {0}", asyncEx?.Message);

                        long ActualMsgs = mode.Equals(MsgDeliveryMode.NonPersistent) ? msgCount + lostMsgCount : msgCount;

                        Assert.AreEqual(TotalMsgs, ActualMsgs, "Only received {0} (delivered = {1}, lost = {2}) of {3} messages in {4}ms.", ActualMsgs, msgCount, lostMsgCount, TotalMsgs, timeout);
                    }
                }
                catch (Exception ex)
                {
                    this.PrintTestFailureAndAssert(GetTestMethodName(), "Unexpected Exception.", ex);
                }
                finally
                {
                    // sleep for 2 seconds to allow for pending procuder acknowledgements to be received from the broker.
                    System.Threading.Thread.Sleep(TimeSpan.FromSeconds(2));
                }
            }
        }

        [Test]
        [ConnectionSetup(null, "default")]
        [ConnectionSetup(null, "sender")]
        [SessionSetup("sender", "s1", AckMode = AcknowledgementMode.AutoAcknowledge)]
        [SessionSetup("default", "default", AckMode = AcknowledgementMode.AutoAcknowledge)]
        [TopicSetup("default", "testdest1", Name = "nms.test")]
        [ConsumerSetup("default", "testdest1", "con1")]
        [ProducerSetup("s1", "testdest1", "pro1")]
        public void TestMessageListenerThrowsOnCloseInMessageCallback()
        {
            const int NUM_MSGS = 2;
            const int MESSAGE_RECEIVE_TIMEOUT = 5000; // 5.0s

            using (IConnection connection = this.GetConnection("default"))
            using (IMessageConsumer consumer = this.GetConsumer("con1"))
            using (IMessageProducer producer = this.GetProducer("pro1"))
            {
                try
                {
                    //bool close = false;
                    MessageListener ackcallback = CreateListener(NUM_MSGS);
                    MessageListener callback = (m) =>
                    {
                        try
                        {
                            connection.Close();
                        }
                        catch (NMSException ex)
                        {
                            asyncEx = ex;
                        }
                        ackcallback(m);
                    };

                    consumer.Listener += callback;
                    connection.ExceptionListener += DefaultExceptionListener;
                    connection.Start();

                    ITextMessage message = producer.CreateTextMessage();
                    // Send a couple of messages and verify that an exception is 
                    // thrown if the listener trys to Close() the connection.
                    producer.Send(message);
                    producer.Send(message);
                    if (!waiter.WaitOne(MESSAGE_RECEIVE_TIMEOUT))
                    {
                        Assert.Fail("Received {0} of {1} messages in {2}ms.", msgCount, NUM_MSGS, MESSAGE_RECEIVE_TIMEOUT);
                    }
                    connection.Stop();

                    Assert.NotNull(asyncEx, "Failed to receive Exception on MessageListener after {0} messages.", msgCount);
                    Assert.True(asyncEx is IllegalStateException, "Failed to Recieve the correct IllegalStateException Exception, received : {0}", asyncEx);

                }
                catch (Exception ex)
                {
                    this.PrintTestFailureAndAssert(this.GetMethodName(), "Unexpected Exception.", ex);
                }
            }
        }


        [Test]
        [ConnectionSetup(null, "default")]
        [SessionSetup("default", "default")]
        [QueueSetup("default", "q1", Name = "nms.queue")]
        [ConsumerSetup("default", "q1", new[] { "c1", "c2", "c3" })]
        public void TestConsumerMessageListenerEventAddRemove()
        {
            IMessageConsumer consumer = null;
            IMessage msg = null;
            using (IConnection connection = GetConnection("default"))
            {
                consumer = GetConsumer("c1");
                try
                {
                    consumer.Listener += DefaultMessageListener;
                }
                catch (NMSException ex)
                {
                    this.PrintTestFailureAndAssert(this.GetTestMethodName(), "Unexpected Exception.", ex);
                }

                try
                {
                    msg = consumer.ReceiveNoWait();
                    Assert.Fail("Expected Exception when receiving synchously on asynchrous consumer.");
                }
                catch (NMSException ex)
                {
                    Assert.IsTrue(ex is IllegalStateException, "Expected exception was not IllealStateException. Exception: {0}", ex);
                    Assert.IsTrue(
                        ex.Message.StartsWith("Cannot synchronously receive message on a synchronous consumer"),
                        "Exception Message does not match. Message: {0}", ex.Message
                        );
                }

                connection.Start();
                consumer = GetConsumer("c2");
                try
                {
                    consumer.Listener += DefaultMessageListener;
                    Assert.Fail("Expected Exception when adding callback event on started connection.");
                }
                catch (NMSException ex)
                {
                    Assert.IsTrue(ex is IllegalStateException, "Expected exception was not IllealStateException. Exception: {0}", ex);
                    Assert.IsTrue(
                        ex.Message.StartsWith("Cannot add MessageListener to consumer"),
                        "Exception Message does not match. Message: {0}", ex.Message
                        );
                }

                try
                {
                    consumer.Listener -= DefaultMessageListener;
                    Assert.Fail("Expected Exception when removing callback event on started connection.");
                }
                catch (NMSException ex)
                {
                    Assert.IsTrue(ex is IllegalStateException, "Expected exception was not IllealStateException. Exception: {0}", ex);
                    Assert.IsTrue(
                        ex.Message.StartsWith("Cannot remove MessageListener to consumer"),
                        "Exception Message does not match. Message: {0}", ex.Message
                        );
                }


            }
        }

        /*
         * Test Durable Consumer Create (subscribe) and Unsubscribe capabilities.
         */
        [Test]
        [ConnectionSetup(null,"c1")]
        [SessionSetup("c1", "s1")]
        [TopicSetup("s1", "dt1", Name = DURABLE_TOPIC_NAME)]
        public void TestDurableConsumerCreateUnsubscribe()
        {
            string name = DURABLE_SUBSRIPTION_NAME;
            bool cleaned = false;
            IMessageConsumer durableConsumer = null;
            using (IConnection connection = GetConnection("c1"))
            using (ISession session = GetSession("s1"))
            {

                try
                {
                    ITopic topic = GetDestination("dt1") as ITopic;
                    durableConsumer = session.CreateDurableConsumer(topic, name, null, false);

                    try
                    {
                        IMessageConsumer other = session.CreateDurableConsumer(topic, name, null, false);
                        Assert.Fail("Expected exception for using the same durable consumer name {0} on the same connection", name);
                    }
                    catch (NMSException nmse)
                    {
                        Assert.AreEqual(nmse.ErrorCode, "nms:internal");
                    }

                    try
                    {
                        session.DeleteDurableConsumer(name);
                        Assert.Fail("Expected exception for deleting active durable consumer {0}.", name);
                    }
                    catch(IllegalStateException)
                    {
                        // pass
                    }

                    // make durable consumer inactive
                    durableConsumer.Close();

                    durableConsumer = null;

                    // unsubscribe the subscription
                    session.DeleteDurableConsumer(name);

                    cleaned = true;

                    try
                    {
                        session.DeleteDurableConsumer(name);
                        Assert.Fail("Expected InvalidDestinationException On Delete Consumer Operation for non-existent durable consumer with name {0}", name);
                    }
                    catch (InvalidDestinationException ide)
                    {
                        // pass
                        Logger.Info(ide.Message);
                    }

                }
                catch (Exception ex)
                {
                    this.PrintTestFailureAndAssert(this.GetTestMethodName(), "Unexpected Exception.", ex);
                }
                finally
                {
                    if(durableConsumer != null)
                    {
                        durableConsumer.Close();
                    }

                    if (!cleaned)
                    {
                        try
                        {
                            session.DeleteDurableConsumer(name);
                        }
                        catch (Exception ex)
                        {
                            Logger.Warn(string.Format("Failed to clean up Durable Consumer {0}. Cause : {1}", name, ex));
                        }
                    }
                }
            }

        }

        /*
         * Tests the durability accross multiple connections. This test will create three connections where one
         * connection creates the durable subscription and consumes from it, one the sends message after its created,
         * and one where the consumer reactiviates the subscription and consumes message while the subscription
         * was inactive. The test then deletes the subscription using the producer connection.
         */
        [Test]
        [ConnectionSetup(null, "c1", ClientId = CUSTOM_CLIENT_ID)]
        [ConnectionSetup(null, "c2", "c3")]
        [SessionSetup("c1", "s1")]
        [SessionSetup("c3", "s2")]
        [TopicSetup("s1", "t1", Name = DURABLE_TOPIC_NAME)]
        [ProducerSetup("s2", "t1", "sender", DeliveryMode = MsgDeliveryMode.NonPersistent)]
        public void TestSubscriptionDurabilityOverConnections()
        {
            const int MSG_BATCH_SIZE = 100;
            const int TIMEOUT = 1000 * MSG_BATCH_SIZE;
            const string MSG_BATCH_ID_PROP_KEY = "batch_id";
            string subName = DURABLE_SUBSRIPTION_NAME;
            ITopic topic = this.GetDestination("t1") as ITopic;
            int msgSentCount = 0;
            bool cleaned = false;
            try
            {
                string clientId = null;
                IMessageProducer producer = this.GetProducer("sender");
                ITextMessage sendMsg = producer.CreateTextMessage();

                using (IConnection initialConnection = this.GetConnection("c1"))
                {
                    initialConnection.ExceptionListener += DefaultExceptionListener;
                    clientId = initialConnection.ClientId;
                    // create subscription
                    ISession subFactory = this.GetSession("s1");
                    IMessageConsumer durableConsumer = subFactory.CreateDurableConsumer(topic, subName, null, false);
                    durableConsumer.Listener += CreateListener(MSG_BATCH_SIZE);

                    // send messages to subscription
                    for (int i = 0; i < MSG_BATCH_SIZE; i++)
                    {
                        sendMsg.Text = String.Format("msg : {0}", msgSentCount);
                        sendMsg.Properties.SetInt(MSG_BATCH_ID_PROP_KEY, i);
                        producer.Send(sendMsg);
                        msgSentCount++;
                    }

                    initialConnection.Start();

                    // assert messages are received
                    Assert.IsTrue(this.waiter.WaitOne(TIMEOUT), 
                        "Timed out waiting to receive messages, received {0} of {1} in {2}ms", 
                        msgCount, msgSentCount, TIMEOUT);
                    Assert.IsNull(asyncEx, "Caught asynchronous exception. {0}", asyncEx?.ToString());
                    Assert.AreEqual(msgSentCount, msgCount, "Messages sent do not match messages received");


                }
                // initial connection is closed and the subscription is inactive.

                // send more messages to the subscription.
                for (int i = 0; i < MSG_BATCH_SIZE; i++)
                {
                    sendMsg.Text = String.Format("msg : {0}", msgSentCount);
                    sendMsg.Properties.SetInt(MSG_BATCH_ID_PROP_KEY, i);
                    producer.Send(sendMsg);
                    msgSentCount++;
                }

                this.waiter.Reset();

                // re-activate the subscription
                using (IConnection connection = this.GetConnection("c2"))
                {
                    connection.ClientId = clientId;
                    connection.ExceptionListener += DefaultExceptionListener;

                    ISession subscriptionFactory = connection.CreateSession();
                    IMessageConsumer durableConsumer = subscriptionFactory.CreateDurableConsumer(topic, subName, null, false);
                    durableConsumer.Listener += CreateListener(MSG_BATCH_SIZE * 2);

                    connection.Start();

                    // assert messages sent while inactive are received
                    Assert.IsTrue(this.waiter.WaitOne(TIMEOUT),
                        "Timed out waiting to receive messages, received {0} of {1} in {2}ms",
                        msgCount, msgSentCount, TIMEOUT);
                    Assert.IsNull(asyncEx, "Caught asynchronous exception. {0}", asyncEx?.ToString());
                    Assert.AreEqual(msgSentCount, msgCount, "Messages sent while inactive do not match messages received");

                    // unsubscribe

                    durableConsumer.Close();

                    subscriptionFactory.DeleteDurableConsumer(subName);

                    durableConsumer = null;

                    cleaned = true;
                }
            }
            catch (Exception ex)
            {
                this.PrintTestFailureAndAssert(this.GetTestMethodName(), "Unexpected Exception.", ex);
            }
            finally
            {
                // clean up code to prevent leaving a durbale subscription on the test broker.
                if (!cleaned)
                {
                    try
                    {
                        this.GetSession("s2").DeleteDurableConsumer(subName);
                    }
                    catch (InvalidDestinationException ide)
                    {
                        Logger.Info(string.Format("Unable to unsubscribe from {0}, Cause : {1}", subName, ide));
                    }
                    catch (Exception ex)
                    {
                        Logger.Warn(string.Format("Caught unexpected failure while unsubscribing from {0}. Failure : {1}", subName, ex));
                    }
                }
            }

        }

        // next test
    }
}