/*
 * 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 Apache.NMS;
using Apache.NMS.Util;
using Apache.NMS.Test;
using NUnit.Framework;
using NUnit.Framework.Extensions;

namespace Apache.NMS.Stomp.Test
{
    [TestFixture]
    public class ConsumerTest : NMSTestSupport
    {
        protected static string TEST_CLIENT_ID = "TestConsumerClientId";
        protected const int COUNT = 25;
        protected const string VALUE_NAME = "value";

        private bool dontAck;

// The .NET CF does not have the ability to interrupt threads, so this test is impossible.
#if !NETCF
//        [RowTest]
//        [Row(AcknowledgementMode.AutoAcknowledge)]
//        [Row(AcknowledgementMode.ClientAcknowledge)]
//        [Row(AcknowledgementMode.DupsOkAcknowledge)]
//        [Row(AcknowledgementMode.Transactional)]
        public void TestNoTimeoutConsumer(AcknowledgementMode ackMode)
        {
            // Launch a thread to perform IMessageConsumer.Receive().
            // If it doesn't fail in less than three seconds, no exception was thrown.
            Thread receiveThread = new Thread(new ThreadStart(TimeoutConsumerThreadProc));
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                using(ISession session = connection.CreateSession(ackMode))
                {
                    ITemporaryQueue queue = session.CreateTemporaryQueue();
                    using(this.timeoutConsumer = session.CreateConsumer(queue))
                    {
                        receiveThread.Start();
                        if(receiveThread.Join(3000))
                        {
                            Assert.Fail("IMessageConsumer.Receive() returned without blocking.  Test failed.");
                        }
                        else
                        {
                            // Kill the thread - otherwise it'll sit in Receive() until a message arrives.
                            receiveThread.Interrupt();
                        }
                    }
                }
            }
        }

        protected IMessageConsumer timeoutConsumer;

        protected void TimeoutConsumerThreadProc()
        {
            try
            {
                timeoutConsumer.Receive();
            }
            catch(ArgumentOutOfRangeException e)
            {
                // The test failed.  We will know because the timeout will expire inside TestNoTimeoutConsumer().
                Assert.Fail("Test failed with exception: " + e.Message);
            }
            catch(ThreadInterruptedException)
            {
                // The test succeeded!  We were still blocked when we were interrupted.
            }
            catch(Exception e)
            {
                // Some other exception occurred.
                Assert.Fail("Test failed with exception: " + e.Message);
            }
        }

//        [RowTest]
//        [Row(AcknowledgementMode.AutoAcknowledge)]
//        [Row(AcknowledgementMode.ClientAcknowledge)]
//        [Row(AcknowledgementMode.DupsOkAcknowledge)]
//        [Row(AcknowledgementMode.Transactional)]
        public void TestSyncReceiveConsumerClose(AcknowledgementMode ackMode)
        {
            // Launch a thread to perform IMessageConsumer.Receive().
            // If it doesn't fail in less than three seconds, no exception was thrown.
            Thread receiveThread = new Thread(new ThreadStart(TimeoutConsumerThreadProc));
            using (IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                using (ISession session = connection.CreateSession(ackMode))
                {
                    ITemporaryQueue queue = session.CreateTemporaryQueue();
                    using (this.timeoutConsumer = session.CreateConsumer(queue))
                    {
                        receiveThread.Start();
                        if (receiveThread.Join(3000))
                        {
                            Assert.Fail("IMessageConsumer.Receive() returned without blocking.  Test failed.");
                        }
                        else
                        {
                            // Kill the thread - otherwise it'll sit in Receive() until a message arrives.
                            this.timeoutConsumer.Close();
                            receiveThread.Join(10000);
                            if (receiveThread.IsAlive)
                            {
                                // Kill the thread - otherwise it'll sit in Receive() until a message arrives.
                                receiveThread.Interrupt();
                                Assert.Fail("IMessageConsumer.Receive() thread is still alive, Close should have killed it.");
                            }
                        }
                    }
                }
            }
        }

        internal class ThreadArg
        {
            internal IConnection connection = null;
            internal ISession session = null;
            internal IDestination destination = null;
        }

        protected void DelayedProducerThreadProc(Object arg)
        {
            try
            {
                ThreadArg args = arg as ThreadArg;

                using(ISession session = args.connection.CreateSession())
                {
                    using(IMessageProducer producer = session.CreateProducer(args.destination))
                    {
                        // Give the consumer time to enter the receive.
                        Thread.Sleep(5000);

                        producer.Send(args.session.CreateTextMessage("Hello World"));
                    }
                }
            }
            catch(Exception e)
            {
                // Some other exception occurred.
                Assert.Fail("Test failed with exception: " + e.Message);
            }
        }

//        [RowTest]
//        [Row(AcknowledgementMode.AutoAcknowledge)]
//        [Row(AcknowledgementMode.ClientAcknowledge)]
//        [Row(AcknowledgementMode.DupsOkAcknowledge)]
//        [Row(AcknowledgementMode.Transactional)]
        public void TestDoChangeSentMessage(AcknowledgementMode ackMode)
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                using(ISession session = connection.CreateSession(ackMode))
                {
                    ITemporaryQueue queue = session.CreateTemporaryQueue();
                    using(IMessageConsumer consumer = session.CreateConsumer(queue))
                    {
                        IMessageProducer producer = session.CreateProducer(queue);
                        ITextMessage message = session.CreateTextMessage();

                        string prefix = "ConsumerTest - TestDoChangeSentMessage: ";

                        for(int i = 0; i < COUNT; i++)
                        {
                            message.Properties[VALUE_NAME] = i;
                            message.Text = prefix + Convert.ToString(i);

                            producer.Send(message);

                            message.ClearBody();
                            message.ClearProperties();
                        }

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

                        for(int i = 0; i < COUNT; i++)
                        {
                            ITextMessage msg = consumer.Receive(TimeSpan.FromMilliseconds(2000)) as ITextMessage;
                            Assert.AreEqual(msg.Text, prefix + Convert.ToString(i));
                            Assert.AreEqual(msg.Properties.GetInt(VALUE_NAME), i);
                        }

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

                    }
                }
            }
        }

//        [RowTest]
//        [Row(AcknowledgementMode.AutoAcknowledge)]
//        [Row(AcknowledgementMode.ClientAcknowledge)]
//        [Row(AcknowledgementMode.DupsOkAcknowledge)]
//        [Row(AcknowledgementMode.Transactional)]
        public void TestConsumerReceiveBeforeMessageDispatched(AcknowledgementMode ackMode)
        {
            // Launch a thread to perform a delayed send.
            Thread sendThread = new Thread(DelayedProducerThreadProc);
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                using(ISession session = connection.CreateSession(ackMode))
                {
                    ITemporaryQueue queue = session.CreateTemporaryQueue();
                    using(IMessageConsumer consumer = session.CreateConsumer(queue))
                    {
                        ThreadArg arg = new ThreadArg();

                        arg.connection = connection;
                        arg.session = session;
                        arg.destination = queue;

                        sendThread.Start(arg);
                        IMessage message = consumer.Receive(TimeSpan.FromMinutes(1));
                        Assert.IsNotNull(message);
                    }
                }
            }
        }

//        [RowTest]
//        [Row(MsgDeliveryMode.NonPersistent, DestinationType.Queue)]
//        [Row(MsgDeliveryMode.NonPersistent, DestinationType.Topic)]
        public void TestDontStart(MsgDeliveryMode deliveryMode, DestinationType destinationType )
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                ISession session = connection.CreateSession();
                IDestination destination = CreateDestination(session, destinationType);
                IMessageConsumer consumer = session.CreateConsumer(destination);

                // Send the messages
                SendMessages(session, destination, deliveryMode, 1);

                // Make sure no messages were delivered.
                Assert.IsNull(consumer.Receive(TimeSpan.FromMilliseconds(1000)));
            }
        }

//        [RowTest]
//        [Row(MsgDeliveryMode.NonPersistent, DestinationType.Queue)]
//        [Row(MsgDeliveryMode.Persistent, DestinationType.Queue)]
//        [Row(MsgDeliveryMode.NonPersistent, DestinationType.Topic)]
//        [Row(MsgDeliveryMode.Persistent, DestinationType.Topic)]
//        [Row(MsgDeliveryMode.NonPersistent, DestinationType.TemporaryQueue)]
//        [Row(MsgDeliveryMode.Persistent, DestinationType.TemporaryQueue)]
//        [Row(MsgDeliveryMode.NonPersistent, DestinationType.TemporaryTopic)]
//        [Row(MsgDeliveryMode.Persistent, DestinationType.TemporaryTopic)]
        public void TestSendReceiveTransacted(MsgDeliveryMode deliveryMode, DestinationType destinationType)
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                // Send a message to the broker.
                connection.Start();
                ISession session = connection.CreateSession(AcknowledgementMode.Transactional);
                IDestination destination = CreateDestination(session, destinationType);
                IMessageConsumer consumer = session.CreateConsumer(destination);
                IMessageProducer producer = session.CreateProducer(destination);

                producer.DeliveryMode = deliveryMode;
                producer.Send(session.CreateTextMessage("Test"));

                // Message should not be delivered until commit.
                Thread.Sleep(1000);
                Assert.IsNull(consumer.ReceiveNoWait());
                session.Commit();

                // Make sure only 1 message was delivered.
                IMessage message = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNotNull(message);
                Assert.IsFalse(message.NMSRedelivered);
                Assert.IsNull(consumer.ReceiveNoWait());

                // Message should be redelivered is rollback is used.
                session.Rollback();

                // Make sure only 1 message was delivered.
                message = consumer.Receive(TimeSpan.FromMilliseconds(2000));
                Assert.IsNotNull(message);
                Assert.IsTrue(message.NMSRedelivered);
                Assert.IsNull(consumer.ReceiveNoWait());

                // If we commit now, the message should not be redelivered.
                session.Commit();
                Thread.Sleep(1000);
                Assert.IsNull(consumer.ReceiveNoWait());
            }
        }

        //[Test]
        public void TestAckedMessageAreConsumed()
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                IQueue queue = session.GetQueue(Guid.NewGuid().ToString());
                IMessageProducer producer = session.CreateProducer(queue);
                producer.Send(session.CreateTextMessage("Hello"));

                // Consume the message...
                IMessageConsumer consumer = session.CreateConsumer(queue);
                IMessage msg = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNotNull(msg);
                msg.Acknowledge();

                // Reset the session.
                session.Close();
                session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);

                // Attempt to Consume the message...
                consumer = session.CreateConsumer(queue);
                msg = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNull(msg);

                session.Close();
            }
        }

        //[Test]
        public void TestLastMessageAcked()
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                IQueue queue = session.GetQueue(Guid.NewGuid().ToString());
                IMessageProducer producer = session.CreateProducer(queue);
                producer.Send(session.CreateTextMessage("Hello"));
                producer.Send(session.CreateTextMessage("Hello2"));
                producer.Send(session.CreateTextMessage("Hello3"));

                // Consume the message...
                IMessageConsumer consumer = session.CreateConsumer(queue);
                IMessage msg = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNotNull(msg);
                msg = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNotNull(msg);
                msg = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNotNull(msg);
                msg.Acknowledge();

                // Reset the session.
                session.Close();
                session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);

                // Attempt to Consume the message...
                consumer = session.CreateConsumer(queue);
                msg = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNull(msg);

                session.Close();
            }
        }

        //[Test]
        public void TestUnAckedMessageAreNotConsumedOnSessionClose()
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                IQueue queue = session.GetQueue(Guid.NewGuid().ToString());
                IMessageProducer producer = session.CreateProducer(queue);
                producer.Send(session.CreateTextMessage("Hello"));

                // Consume the message...
                IMessageConsumer consumer = session.CreateConsumer(queue);
                IMessage msg = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNotNull(msg);
                // Don't ack the message.

                // Reset the session.  This should cause the unacknowledged message to be re-delivered.
                session.Close();
                session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);

                // Attempt to Consume the message...
                consumer = session.CreateConsumer(queue);
                msg = consumer.Receive(TimeSpan.FromMilliseconds(2000));
                Assert.IsNotNull(msg);
                msg.Acknowledge();

                session.Close();
            }
        }

        //[Test]
        public void TestAsyncAckedMessageAreConsumed()
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                IQueue queue = session.GetQueue(Guid.NewGuid().ToString());
                IMessageProducer producer = session.CreateProducer(queue);
                producer.Send(session.CreateTextMessage("Hello"));

                // Consume the message...
                IMessageConsumer consumer = session.CreateConsumer(queue);
                consumer.Listener += new MessageListener(OnMessage);

                Thread.Sleep(5000);

                // Reset the session.
                session.Close();

                session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);

                // Attempt to Consume the message...
                consumer = session.CreateConsumer(queue);
                IMessage msg = consumer.Receive(TimeSpan.FromMilliseconds(1000));
                Assert.IsNull(msg);

                session.Close();
            }
        }

        [Test]
        public void TestAckOfExpired()
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                IQueue queue = session.GetQueue(Guid.NewGuid().ToString());
                IMessageConsumer consumer = session.CreateConsumer(queue);

                ISession sendSession = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
                IMessageProducer producer = sendSession.CreateProducer(queue);
                producer.TimeToLive = TimeSpan.FromMilliseconds(1000);
                const int count = 4;
                for(int i = 0; i < count; i++)
                {
                    ITextMessage message = sendSession.CreateTextMessage("" + i);
                    producer.Send(message);
                }

                // let first bunch in queue expire
                Thread.Sleep(2000);

                producer.TimeToLive = TimeSpan.Zero;
                for(int i = 0; i < count; i++)
                {
                    ITextMessage message = sendSession.CreateTextMessage("no expiry" + i);
                    producer.Send(message);
                }

                for(int i=0; i < count; i++)
                {
                    ITextMessage msg = consumer.Receive(TimeSpan.FromMilliseconds(5000)) as ITextMessage;
                    Assert.IsNotNull(msg);
                    Assert.IsTrue(msg.Text.Contains("no expiry"), "message has \"no expiry\" text: " + msg.Text);
                }
            }
        }

        [Test]
        public void TestAckOfExpiredWithoutTimeStamps()
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                IQueue queue = session.GetQueue(Guid.NewGuid().ToString());
                IMessageConsumer consumer = session.CreateConsumer(queue);

                ISession sendSession = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
                IMessageProducer producer = sendSession.CreateProducer(queue);
                producer.DisableMessageTimestamp = true;
                producer.TimeToLive = TimeSpan.FromMilliseconds(1000);
                const int count = 4;
                for(int i = 0; i < count; i++)
                {
                    ITextMessage message = sendSession.CreateTextMessage("" + i);
                    producer.Send(message);
                }

                // let first bunch in queue expire
                Thread.Sleep(2000);

                producer.TimeToLive = TimeSpan.Zero;
                for(int i = 0; i < count; i++)
                {
                    ITextMessage message = sendSession.CreateTextMessage("no expiry" + i);
                    producer.Send(message);
                }

                for(int i=0; i < count; i++)
                {
                    ITextMessage msg = consumer.Receive(TimeSpan.FromMilliseconds(5000)) as ITextMessage;
                    Assert.IsNotNull(msg);
                    Assert.IsTrue(msg.Text.Contains("no expiry"), "message has \"no expiry\" text: " + msg.Text);
                }
            }
        }

        //[Test]
        public void TestAsyncUnAckedMessageAreNotConsumedOnSessionClose()
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();
                // don't aknowledge message on onMessage() call
                dontAck = true;
                ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                IQueue queue = session.GetQueue("Guid.NewGuid().ToString()");
                IMessageProducer producer = session.CreateProducer(queue);
                producer.Send(session.CreateTextMessage("Hello"));

                // Consume the message...
                IMessageConsumer consumer = session.CreateConsumer(queue);
                consumer.Listener += new MessageListener(OnMessage);
                // Don't ack the message.

                // Reset the session. This should cause the Unacked message to be
                // redelivered.
                session.Close();

                Thread.Sleep(5000);
                session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                // Attempt to Consume the message...
                consumer = session.CreateConsumer(queue);
                IMessage msg = consumer.Receive(TimeSpan.FromMilliseconds(2000));
                Assert.IsNotNull(msg);
                msg.Acknowledge();

                session.Close();
            }
        }

        //[Test]
        public void TestAddRemoveAsnycMessageListener()
        {
            using(IConnection connection = CreateConnection(TEST_CLIENT_ID + ":" + new Random().Next()))
            {
                connection.Start();

                ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
                ITemporaryTopic topic = session.CreateTemporaryTopic();
                IMessageConsumer consumer = session.CreateConsumer(topic);

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

                consumer.Close();
            }
        }

        public void OnMessage(IMessage message)
        {
            Assert.IsNotNull(message);

            if(!dontAck)
            {
                try
                {
                    message.Acknowledge();
                }
                catch(Exception)
                {
                }
            }
        }

#endif

    }
}