//------------------------------------------------------------
// Copyright (c) Microsoft Corporation.  All rights reserved.
//------------------------------------------------------------
namespace System.ServiceModel.Channels
{
    using System.Runtime;
    using System.ServiceModel.Diagnostics;
    using System.Threading;
    using System.Transactions;
    using System.ComponentModel;
    using System.Runtime.Versioning;

    // PostRollbackErrorStrategy
    interface IPostRollbackErrorStrategy
    {
        bool AnotherTryNeeded();
    }

    // SimplePostRollbackErrorStrategy
    class SimplePostRollbackErrorStrategy : IPostRollbackErrorStrategy
    {
        const int Attempts = 50;
        const int MillisecondsToSleep = 100;

        int attemptsLeft = Attempts;
        long lookupId;

        internal SimplePostRollbackErrorStrategy(long lookupId)
        {
            this.lookupId = lookupId;
        }

        public bool AnotherTryNeeded()
        {
            if (--this.attemptsLeft > 0)
            {
                if (attemptsLeft == (Attempts - 1))
                    MsmqDiagnostics.MessageLockedUnderTheTransaction(lookupId);
                Thread.Sleep(TimeSpan.FromMilliseconds(MillisecondsToSleep));
                return true;
            }
            else
            {
                MsmqDiagnostics.MoveOrDeleteAttemptFailed(lookupId);
                return false;
            }
        }
    }

    sealed class MsmqReceiveHelper
    {
        IPoisonHandlingStrategy poisonHandler;
        string queueName;
        MsmqQueue queue;
        MsmqReceiveParameters receiveParameters;
        Uri uri;
        string instanceId;
        IMsmqMessagePool pool;
        MsmqInputChannelBase channel;
        MsmqChannelListenerBase listener;
        ServiceModelActivity activity;
        string msmqRuntimeNativeLibrary;

        internal MsmqReceiveHelper(MsmqReceiveParameters receiveParameters, Uri uri, IMsmqMessagePool messagePool, MsmqInputChannelBase channel, MsmqChannelListenerBase listener)
        {
            this.queueName = receiveParameters.AddressTranslator.UriToFormatName(uri);
            this.receiveParameters = receiveParameters;
            this.uri = uri;
            this.instanceId = uri.ToString().ToUpperInvariant();
            this.pool = messagePool;
            this.poisonHandler = Msmq.CreatePoisonHandler(this);
            this.channel = channel;
            this.listener = listener;
            this.queue = Msmq.CreateMsmqQueue(this);
        }

        internal ServiceModelActivity Activity
        {
            get { return this.activity; }
        }

        IPoisonHandlingStrategy PoisonHandler
        {
            get { return this.poisonHandler; }
        }

        internal MsmqReceiveParameters MsmqReceiveParameters
        {
            get { return this.receiveParameters; }
        }

        internal MsmqInputChannelBase Channel
        {
            get { return this.channel; }
        }

        internal MsmqChannelListenerBase ChannelListener
        {
            get { return this.listener; }
        }

        internal Uri ListenUri
        {
            get { return this.uri; }
        }

        internal string InstanceId
        {
            get { return this.instanceId; }
        }

        internal MsmqQueue Queue
        {
            get { return this.queue; }
        }

        internal bool Transactional
        {
            get { return this.receiveParameters.ExactlyOnce; }
        }

        internal string MsmqRuntimeNativeLibrary
        {
            get
            {
                if (this.msmqRuntimeNativeLibrary == null)
                {
                    this.msmqRuntimeNativeLibrary = Environment.SystemDirectory + "\\" + UnsafeNativeMethods.MQRT;
                }
                return this.msmqRuntimeNativeLibrary;
            }
        }

        internal void Open()
        {
            this.activity = MsmqDiagnostics.StartListenAtActivity(this);
            using (MsmqDiagnostics.BoundOpenOperation(this))
            {
                this.queue.EnsureOpen();
                this.poisonHandler.Open();
            }
        }

        internal void Close()
        {
            using (ServiceModelActivity.BoundOperation(this.Activity))
            {
                this.poisonHandler.Dispose();
                this.queue.Dispose();
            }
            ServiceModelActivity.Stop(this.activity);
        }

        internal MsmqInputMessage TakeMessage()
        {
            return this.pool.TakeMessage();
        }

        internal void ReturnMessage(MsmqInputMessage message)
        {
            this.pool.ReturnMessage(message);
        }

        internal static void TryAbortTransactionCurrent()
        {
            if (null != Transaction.Current)
            {
                try
                {
                    Transaction.Current.Rollback();
                }
                catch (TransactionAbortedException ex)
                {
                    MsmqDiagnostics.ExpectedException(ex);
                }
                catch (ObjectDisposedException ex)
                {
                    MsmqDiagnostics.ExpectedException(ex);
                }
            }
        }

        internal void DropOrRejectReceivedMessage(MsmqMessageProperty messageProperty, bool reject)
        {
            this.DropOrRejectReceivedMessage(this.Queue, messageProperty, reject);
        }

        internal void DropOrRejectReceivedMessage(MsmqQueue queue, MsmqMessageProperty messageProperty, bool reject)
        {
            if (this.Transactional)
            {
                TryAbortTransactionCurrent();
                IPostRollbackErrorStrategy postRollback = new SimplePostRollbackErrorStrategy(messageProperty.LookupId);
                MsmqQueue.MoveReceiveResult result = MsmqQueue.MoveReceiveResult.Unknown;
                do
                {
                    using (MsmqEmptyMessage emptyMessage = new MsmqEmptyMessage())
                    {
                        using (TransactionScope scope = new TransactionScope(TransactionScopeOption.RequiresNew))
                        {
                            result = queue.TryReceiveByLookupId(messageProperty.LookupId, emptyMessage, MsmqTransactionMode.CurrentOrThrow);
                            if (MsmqQueue.MoveReceiveResult.Succeeded == result && reject)
                                queue.MarkMessageRejected(messageProperty.LookupId);
                            scope.Complete();
                        }
                    }

                    if (result == MsmqQueue.MoveReceiveResult.Succeeded)
                        // If 'Reject' supported and 'Reject' requested, put reject in the trace, otherwise put 'Drop'
                        MsmqDiagnostics.MessageConsumed(instanceId, messageProperty.MessageId, (Msmq.IsRejectMessageSupported && reject));

                    if (result != MsmqQueue.MoveReceiveResult.MessageLockedUnderTransaction)
                        break;
                }
                while (postRollback.AnotherTryNeeded());
            }
            else
            {
                MsmqDiagnostics.MessageConsumed(instanceId, messageProperty.MessageId, false);
            }
        }

        //
        internal static void MoveReceivedMessage(MsmqQueue queueFrom, MsmqQueue queueTo, long lookupId)
        {
            TryAbortTransactionCurrent();

            IPostRollbackErrorStrategy postRollback = new SimplePostRollbackErrorStrategy(lookupId);
            MsmqQueue.MoveReceiveResult result = MsmqQueue.MoveReceiveResult.Unknown;
            do
            {
                result = queueFrom.TryMoveMessage(lookupId, queueTo, MsmqTransactionMode.Single);

                if (result != MsmqQueue.MoveReceiveResult.MessageLockedUnderTransaction)
                    break;
            }
            while (postRollback.AnotherTryNeeded());
        }

        internal void FinalDisposition(MsmqMessageProperty messageProperty)
        {
            this.poisonHandler.FinalDisposition(messageProperty);
        }

        // WaitForMessage
        internal bool WaitForMessage(TimeSpan timeout)
        {
            using (MsmqEmptyMessage message = new MsmqEmptyMessage())
            {
                return (MsmqQueue.ReceiveResult.Timeout != this.queue.TryPeek(message, timeout));
            }
        }
        //
        internal IAsyncResult BeginWaitForMessage(TimeSpan timeout, AsyncCallback callback, object state)
        {
            return new WaitForMessageAsyncResult(this.queue, timeout, callback, state);
        }
        //
        public bool EndWaitForMessage(IAsyncResult result)
        {
            return WaitForMessageAsyncResult.End(result);
        }

        internal bool TryReceive(MsmqInputMessage msmqMessage, TimeSpan timeout, MsmqTransactionMode transactionMode, out MsmqMessageProperty property)
        {
            property = null;

            MsmqQueue.ReceiveResult receiveResult = this.Queue.TryReceive(msmqMessage, timeout, transactionMode);
            if (MsmqQueue.ReceiveResult.OperationCancelled == receiveResult)
                return true;
            if (MsmqQueue.ReceiveResult.Timeout == receiveResult)
                return false;
            else
            {
                property = new MsmqMessageProperty(msmqMessage);
                if (this.Transactional)
                {
                    if (this.PoisonHandler.CheckAndHandlePoisonMessage(property))
                    {
                        long lookupId = property.LookupId;
                        property = null;
                        throw DiagnosticUtility.ExceptionUtility.ThrowHelperCritical(new MsmqPoisonMessageException(lookupId));
                    }
                }
                return true;
            }
        }
        //
        internal IAsyncResult BeginTryReceive(MsmqInputMessage msmqMessage, TimeSpan timeout, MsmqTransactionMode transactionMode, AsyncCallback callback, object state)
        {
            if (this.receiveParameters.ExactlyOnce || this.queue is ILockingQueue)
                return new TryTransactedReceiveAsyncResult(this, msmqMessage, timeout, transactionMode, callback, state);
            else
                return new TryNonTransactedReceiveAsyncResult(this, msmqMessage, timeout, callback, state);
        }
        //
        internal bool EndTryReceive(IAsyncResult result, out MsmqInputMessage msmqMessage, out MsmqMessageProperty msmqProperty)
        {
            msmqMessage = null;
            msmqProperty = null;

            if (null == result)
                throw DiagnosticUtility.ExceptionUtility.ThrowHelperArgumentNull("result");

            if (this.receiveParameters.ExactlyOnce)
            {
                TryTransactedReceiveAsyncResult receiveResult = result as TryTransactedReceiveAsyncResult;
                if (null == receiveResult)
                    throw DiagnosticUtility.ExceptionUtility.ThrowHelperArgument(SR.GetString(SR.InvalidAsyncResult));
                return TryTransactedReceiveAsyncResult.End(receiveResult, out msmqMessage, out msmqProperty);
            }
            else
            {
                TryNonTransactedReceiveAsyncResult receiveResult = result as TryNonTransactedReceiveAsyncResult;
                if (null == receiveResult)
                    throw DiagnosticUtility.ExceptionUtility.ThrowHelperArgument(SR.GetString(SR.InvalidAsyncResult));
                return TryNonTransactedReceiveAsyncResult.End(receiveResult, out msmqMessage, out msmqProperty);
            }
        }

        // TryReceiveAsyncResult (tx version)
        class TryTransactedReceiveAsyncResult : AsyncResult
        {
            bool expired;
            MsmqReceiveHelper receiver;
            TimeoutHelper timeoutHelper;
            Transaction txCurrent;
            MsmqInputMessage msmqMessage;
            MsmqMessageProperty messageProperty;
            MsmqTransactionMode transactionMode;
            static Action<object> onComplete = new Action<object>(OnComplete);

            internal TryTransactedReceiveAsyncResult(MsmqReceiveHelper receiver, MsmqInputMessage msmqMessage,
                TimeSpan timeout, MsmqTransactionMode transactionMode, AsyncCallback callback, object state)
                : base(callback, state)
            {
                this.timeoutHelper = new TimeoutHelper(timeout);
                this.txCurrent = Transaction.Current;
                this.receiver = receiver;
                this.msmqMessage = msmqMessage;
                this.transactionMode = transactionMode;
                ActionItem.Schedule(onComplete, this);
            }

            static void OnComplete(object parameter)
            {
                TryTransactedReceiveAsyncResult result = parameter as TryTransactedReceiveAsyncResult;
                Transaction savedTransaction = Transaction.Current;
                Transaction.Current = result.txCurrent;
                try
                {
                    Exception ex = null;
                    try
                    {
                        result.expired = !result.receiver.TryReceive(result.msmqMessage, result.timeoutHelper.RemainingTime(), result.transactionMode, out result.messageProperty);
                    }
                    catch (Exception e)
                    {
                        if (Fx.IsFatal(e))
                            throw;
                        ex = e;
                    }
                    result.Complete(false, ex);
                }
                finally
                {
                    Transaction.Current = savedTransaction;
                }
            }

            internal static bool End(IAsyncResult result, out MsmqInputMessage msmqMessage, out MsmqMessageProperty property)
            {
                TryTransactedReceiveAsyncResult receiveResult = AsyncResult.End<TryTransactedReceiveAsyncResult>(result);
                msmqMessage = receiveResult.msmqMessage;
                property = receiveResult.messageProperty;
                return !receiveResult.expired;
            }
        }

        // TryReceiveAsyncResult (non-tx version)
        class TryNonTransactedReceiveAsyncResult : AsyncResult
        {
            MsmqQueue.ReceiveResult receiveResult;
            MsmqReceiveHelper receiver;
            MsmqInputMessage msmqMessage;
            static AsyncCallback onCompleteStatic = Fx.ThunkCallback(new AsyncCallback(OnCompleteStatic));


            internal TryNonTransactedReceiveAsyncResult(MsmqReceiveHelper receiver, MsmqInputMessage msmqMessage, TimeSpan timeout, AsyncCallback callback, object state)
                : base(callback, state)
            {
                this.receiver = receiver;
                this.msmqMessage = msmqMessage;
                receiver.Queue.BeginTryReceive(msmqMessage, timeout, onCompleteStatic, this);
            }

            static void OnCompleteStatic(IAsyncResult result)
            {
                (result.AsyncState as TryNonTransactedReceiveAsyncResult).OnComplete(result);
            }

            void OnComplete(IAsyncResult result)
            {
                Exception ex = null;
                try
                {
                    receiveResult = receiver.Queue.EndTryReceive(result);
                }
                catch (Exception e)
                {
                    if (Fx.IsFatal(e))
                        throw;
                    ex = e;
                }
                Complete(result.CompletedSynchronously, ex);
            }

            internal static bool End(IAsyncResult result, out MsmqInputMessage msmqMessage, out MsmqMessageProperty property)
            {
                TryNonTransactedReceiveAsyncResult asyncResult = AsyncResult.End<TryNonTransactedReceiveAsyncResult>(result);
                msmqMessage = asyncResult.msmqMessage;
                property = null;
                if (MsmqQueue.ReceiveResult.Timeout == asyncResult.receiveResult)
                    return false;
                else if (MsmqQueue.ReceiveResult.OperationCancelled == asyncResult.receiveResult)
                    return true;
                else
                {
                    property = new MsmqMessageProperty(msmqMessage);
                    return true;
                }
            }
        }

        // WaitForMessageAsyncResult
        class WaitForMessageAsyncResult : AsyncResult
        {
            MsmqQueue msmqQueue;
            MsmqEmptyMessage msmqMessage;
            bool successResult;
            static AsyncCallback onCompleteStatic = Fx.ThunkCallback(new AsyncCallback(OnCompleteStatic));

            public WaitForMessageAsyncResult(MsmqQueue msmqQueue, TimeSpan timeout, AsyncCallback callback, object state)
                : base(callback, state)
            {
                this.msmqMessage = new MsmqEmptyMessage();
                this.msmqQueue = msmqQueue;
                this.msmqQueue.BeginPeek(this.msmqMessage, timeout, onCompleteStatic, this);
            }

            static void OnCompleteStatic(IAsyncResult result)
            {
                ((WaitForMessageAsyncResult)result.AsyncState).OnComplete(result);
            }

            void OnComplete(IAsyncResult result)
            {
                this.msmqMessage.Dispose();
                MsmqQueue.ReceiveResult receiveResult = MsmqQueue.ReceiveResult.Unknown;
                Exception completionException = null;
                try
                {
                    receiveResult = this.msmqQueue.EndPeek(result);
                }
                catch (Exception e)
                {
                    if (Fx.IsFatal(e))
                        throw;
                    completionException = e;
                }

                this.successResult = receiveResult != MsmqQueue.ReceiveResult.Timeout;
                base.Complete(result.CompletedSynchronously, completionException);
            }

            public static bool End(IAsyncResult result)
            {
                WaitForMessageAsyncResult thisPtr = AsyncResult.End<WaitForMessageAsyncResult>(result);
                return thisPtr.successResult;
            }
        }
    }
}