//------------------------------------------------------------
// Copyright (c) Microsoft Corporation.     All rights    reserved.
//------------------------------------------------------------
using System;
using System.Collections;
using System.Collections.Generic;
using System.Text;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Globalization;
using System.Workflow.ComponentModel;
using System.Runtime.Serialization;
using System.Messaging;

namespace System.Workflow.Runtime
{
    [Obsolete("The System.Workflow.* types are deprecated.  Instead, please use the new types from System.Activities.*")]
    public class WorkflowQueuingService
    {
        Object syncRoot = new Object();
        IWorkflowCoreRuntime rootWorkflowExecutor;
        List<IComparable> dirtyQueues;
        EventQueueState pendingQueueState = new EventQueueState();
        Dictionary<IComparable, EventQueueState> persistedQueueStates;

        // event handler used by atomic execution context's Q service for message delivery 
        List<WorkflowQueuingService> messageArrivalEventHandlers;

        // set for inner queuing service
        WorkflowQueuingService rootQueuingService;

        // Runtime information visible to host, stored on the root activity
        [SuppressMessage("Microsoft.Security", "CA2104:DoNotDeclareReadOnlyMutableReferenceTypes",
            Justification = "Design has been approved.  This is a false positive. DependencyProperty is an immutable type.")]
        public readonly static DependencyProperty PendingMessagesProperty = DependencyProperty.RegisterAttached("PendingMessages", typeof(Queue), typeof(WorkflowQueuingService), new PropertyMetadata(DependencyPropertyOptions.NonSerialized));

        // Persisted state properties
        internal static DependencyProperty RootPersistedQueueStatesProperty = DependencyProperty.RegisterAttached("RootPersistedQueueStates", typeof(Dictionary<IComparable, EventQueueState>), typeof(WorkflowQueuingService));
        internal static DependencyProperty LocalPersistedQueueStatesProperty = DependencyProperty.RegisterAttached("LocalPersistedQueueStates", typeof(Dictionary<IComparable, EventQueueState>), typeof(WorkflowQueuingService));
        private const string pendingNotification = "*PendingNotifications";

        // Snapshots created during pre-persist and dumped during post-persist
        // If  persistence fails, changes made to queuing service during pre-persist must be undone
        //    in post-persist.
        // Created for ref. 20575.
        private Dictionary<IComparable, EventQueueState> persistedQueueStatesSnapshot = null;
        private EventQueueState pendingQueueStateSnapshot = null;

        // root Q service constructor
        internal WorkflowQueuingService(IWorkflowCoreRuntime rootWorkflowExecutor)
        {
            this.rootWorkflowExecutor = rootWorkflowExecutor;
            this.rootWorkflowExecutor.RootActivity.SetValue(WorkflowQueuingService.PendingMessagesProperty, this.pendingQueueState.Messages);
            this.persistedQueueStates = (Dictionary<IComparable, EventQueueState>)this.rootWorkflowExecutor.RootActivity.GetValue(WorkflowQueuingService.RootPersistedQueueStatesProperty);
            if (this.persistedQueueStates == null)
            {
                this.persistedQueueStates = new Dictionary<IComparable, EventQueueState>();
                this.rootWorkflowExecutor.RootActivity.SetValue(WorkflowQueuingService.RootPersistedQueueStatesProperty, this.persistedQueueStates);
            }
            if (!this.Exists(pendingNotification))
                this.CreateWorkflowQueue(pendingNotification, false);
        }

        // inner Q service constructor
        internal WorkflowQueuingService(WorkflowQueuingService copyFromQueuingService)
        {
            this.rootQueuingService = copyFromQueuingService;
            this.rootWorkflowExecutor = copyFromQueuingService.rootWorkflowExecutor;
            this.rootWorkflowExecutor.RootActivity.SetValue(WorkflowQueuingService.PendingMessagesProperty, this.pendingQueueState.Messages);
            this.persistedQueueStates = new Dictionary<IComparable, EventQueueState>();
            this.rootWorkflowExecutor.RootActivity.SetValue(WorkflowQueuingService.LocalPersistedQueueStatesProperty, this.persistedQueueStates);
            SubscribeForRootMessageDelivery();
        }

        public WorkflowQueue CreateWorkflowQueue(IComparable queueName, bool transactional)
        {
            if (queueName == null)
                throw new ArgumentNullException("queueName");

            lock (SyncRoot)
            {
                // if not transactional create one at the root 
                // so it is visible outside this transaction
                if (this.rootQueuingService != null && !transactional)
                {
                    return this.rootQueuingService.CreateWorkflowQueue(queueName, false);
                }

                NewQueue(queueName, true, transactional);

                return new WorkflowQueue(this, queueName);
            }
        }

        public void DeleteWorkflowQueue(IComparable queueName)
        {
            if (queueName == null)
                throw new ArgumentNullException("queueName");

            lock (SyncRoot)
            {
                // when we are deleting the queue from activity
                // message delivery should not happen.
                if (this.rootQueuingService != null && !IsTransactionalQueue(queueName))
                {
                    this.rootQueuingService.DeleteWorkflowQueue(queueName);
                    return;
                }

                EventQueueState queueState = GetEventQueueState(queueName);

                Queue queue = queueState.Messages;
                Queue pendingQueue = this.pendingQueueState.Messages;

                while (queue.Count != 0)
                {
                    pendingQueue.Enqueue(queue.Dequeue());
                }

                WorkflowTrace.Runtime.TraceInformation("Queuing Service: Deleting Queue with ID {0} for {1}", queueName.GetHashCode(), queueName);
                this.persistedQueueStates.Remove(queueName);
            }
        }

        public bool Exists(IComparable queueName)
        {
            if (queueName == null)
                throw new ArgumentNullException("queueName");

            lock (SyncRoot)
            {
                if (this.rootQueuingService != null && !IsTransactionalQueue(queueName))
                {
                    return this.rootQueuingService.Exists(queueName);
                }

                return this.persistedQueueStates.ContainsKey(queueName);
            }
        }

        public WorkflowQueue GetWorkflowQueue(IComparable queueName)
        {
            if (queueName == null)
                throw new ArgumentNullException("queueName");

            lock (SyncRoot)
            {
                if (this.rootQueuingService != null && !IsTransactionalQueue(queueName))
                {
                    return this.rootQueuingService.GetWorkflowQueue(queueName);
                }

                GetEventQueueState(queueName);

                return new WorkflowQueue(this, queueName);
            }
        }

        #region internal functions

        internal Object SyncRoot
        {
            get { return syncRoot; }
        }

        internal void EnqueueEvent(IComparable queueName, Object item)
        {
            if (queueName == null)
                throw new ArgumentNullException("queueName");

            lock (SyncRoot)
            {
                if (this.rootQueuingService != null && !IsTransactionalQueue(queueName))
                {
                    this.rootQueuingService.EnqueueEvent(queueName, item);
                    return;
                }

                EventQueueState qState = GetQueue(queueName);
                if (!qState.Enabled)
                {
                    throw new QueueException(String.Format(CultureInfo.CurrentCulture, ExecutionStringManager.QueueNotEnabled, queueName), MessageQueueErrorCode.QueueNotAvailable);
                }

                // note enqueue allowed irrespective of dirty flag since it is delivered through
                qState.Messages.Enqueue(item);

                WorkflowTrace.Runtime.TraceInformation("Queuing Service: Enqueue item Queue ID {0} for {1}", queueName.GetHashCode(), queueName);

                // notify message arrived subscribers
                for (int i = 0; messageArrivalEventHandlers != null && i < messageArrivalEventHandlers.Count; ++i)
                {
                    this.messageArrivalEventHandlers[i].OnItemEnqueued(queueName, item);
                }

                NotifyExternalSubscribers(queueName, qState, item);
            }
        }
        internal bool SafeEnqueueEvent(IComparable queueName, Object item)
        {
            if (queueName == null)
                throw new ArgumentNullException("queueName");

            lock (SyncRoot)
            {
                if (this.rootQueuingService != null && !IsTransactionalQueue(queueName))
                {
                    return this.rootQueuingService.SafeEnqueueEvent(queueName, item);
                }

                EventQueueState qState = GetQueue(queueName);
                if (!qState.Enabled)
                {
                    throw new QueueException(String.Format(CultureInfo.CurrentCulture, ExecutionStringManager.QueueNotEnabled, queueName), MessageQueueErrorCode.QueueNotAvailable);
                }

                // note enqueue allowed irrespective of dirty flag since it is delivered through
                qState.Messages.Enqueue(item);

                WorkflowTrace.Runtime.TraceInformation("Queuing Service: Enqueue item Queue ID {0} for {1}", queueName.GetHashCode(), queueName);

                // notify message arrived subscribers
                for (int i = 0; messageArrivalEventHandlers != null && i < messageArrivalEventHandlers.Count; ++i)
                {
                    this.messageArrivalEventHandlers[i].OnItemSafeEnqueued(queueName, item);
                }

                NotifySynchronousSubscribers(queueName, qState, item);
                return QueueAsynchronousEvent(queueName, qState);
            }
        }


        internal object Peek(IComparable queueName)
        {
            if (queueName == null)
                throw new ArgumentNullException("queueName");

            lock (SyncRoot)
            {
                if (this.rootQueuingService != null && !IsTransactionalQueue(queueName))
                {
                    return this.rootQueuingService.Peek(queueName);
                }

                EventQueueState queueState = GetEventQueueState(queueName);
                if (queueState.Messages.Count != 0)
                    return queueState.Messages.Peek();

                object[] args = new object[] { System.Messaging.MessageQueueErrorCode.MessageNotFound, queueName };
                string message = string.Format(CultureInfo.CurrentCulture, ExecutionStringManager.EventQueueException, args);

                throw new QueueException(message, MessageQueueErrorCode.MessageNotFound);
            }
        }

        internal Object DequeueEvent(IComparable queueName)
        {
            if (queueName == null)
                throw new ArgumentNullException("queueName");

            lock (SyncRoot)
            {
                if (this.rootQueuingService != null && !IsTransactionalQueue(queueName))
                {
                    return this.rootQueuingService.DequeueEvent(queueName);
                }

                EventQueueState queueState = GetEventQueueState(queueName);
                if (queueState.Messages.Count != 0)
                    return queueState.Messages.Dequeue();

                object[] args = new object[] { System.Messaging.MessageQueueErrorCode.MessageNotFound, queueName };
                string message = string.Format(CultureInfo.CurrentCulture, ExecutionStringManager.EventQueueException, args);

                throw new QueueException(message, MessageQueueErrorCode.MessageNotFound);
            }
        }

        internal EventQueueState GetQueueState(IComparable eventType)
        {
            lock (SyncRoot)
            {
                return GetQueue(eventType);
            }
        }

        Activity caller;

        internal Activity CallingActivity
        {
            get
            {
                if (this.rootQueuingService != null)
                    return this.rootQueuingService.CallingActivity;
                return this.caller;
            }
            set
            {
                if (this.rootQueuingService != null)
                    this.rootQueuingService.CallingActivity = value;

                this.caller = value;
            }
        }

        private bool QueueAsynchronousEvent(IComparable queueName, EventQueueState qState)
        {
            if (qState.AsynchronousListeners.Count != 0 || IsNestedListenersExist(queueName))
            {
                Queue q = GetQueue(pendingNotification).Messages;
                q.Enqueue(new KeyValuePair<IComparable, EventQueueState>(queueName, qState));
                WorkflowTrace.Runtime.TraceInformation("Queuing Service: Queued delayed message notification for '{0}'", queueName.ToString());
                return q.Count == 1;
            }
            return false;
        }

        bool IsNestedListenersExist(IComparable queueName)
        {
            for (int i = 0; messageArrivalEventHandlers != null && i < messageArrivalEventHandlers.Count; ++i)
            {
                WorkflowQueuingService qService = messageArrivalEventHandlers[i];
                EventQueueState queueState = null;

                if (qService.persistedQueueStates.TryGetValue(queueName, out queueState) &&
                    queueState.AsynchronousListeners.Count != 0)
                    return true;
            }
            return false;
        }
        internal void ProcessesQueuedAsynchronousEvents()
        {
            Queue q = GetQueue(pendingNotification).Messages;
            while (q.Count > 0)
            {
                KeyValuePair<IComparable, EventQueueState> pair = (KeyValuePair<IComparable, EventQueueState>)q.Dequeue();
                // notify message arrived subscribers
                WorkflowTrace.Runtime.TraceInformation("Queuing Service: Processing delayed message notification '{0}'", pair.Key.ToString());
                for (int i = 0; messageArrivalEventHandlers != null && i < messageArrivalEventHandlers.Count; ++i)
                {
                    WorkflowQueuingService service = this.messageArrivalEventHandlers[i];
                    if (service.persistedQueueStates.ContainsKey(pair.Key))
                    {
                        EventQueueState qState = service.GetQueue(pair.Key);
                        if (qState.Enabled)
                        {
                            service.NotifyAsynchronousSubscribers(pair.Key, qState, 1);
                        }
                    }
                }
                NotifyAsynchronousSubscribers(pair.Key, pair.Value, 1);
            }
        }

        internal void NotifyAsynchronousSubscribers(IComparable queueName, EventQueueState qState, int numberOfNotification)
        {
            for (int i = 0; i < numberOfNotification; ++i)
            {
                QueueEventArgs args = new QueueEventArgs(queueName);
                lock (SyncRoot)
                {
                    foreach (ActivityExecutorDelegateInfo<QueueEventArgs> subscriber in qState.AsynchronousListeners)
                    {
                        Activity contextActivity = rootWorkflowExecutor.GetContextActivityForId(subscriber.ContextId);
                        Debug.Assert(contextActivity != null);
                        subscriber.InvokeDelegate(contextActivity, args, false);
                        WorkflowTrace.Runtime.TraceInformation("Queuing Service: Notifying async subscriber on queue:'{0}' activity:{1}", queueName.ToString(), subscriber.ActivityQualifiedName);
                    }
                }
            }
        }

        /// <summary>
        /// At termination/completion point, need to move messages from all queues to the pending queue
        /// </summary>
        internal void MoveAllMessagesToPendingQueue()
        {
            lock (SyncRoot)
            {
                Queue pendingQueue = this.pendingQueueState.Messages;
                foreach (EventQueueState queueState in this.persistedQueueStates.Values)
                {
                    Queue queue = queueState.Messages;
                    while (queue.Count != 0)
                    {
                        pendingQueue.Enqueue(queue.Dequeue());
                    }
                }
            }
        }

        #endregion

        #region private root q service helpers

        private EventQueueState GetEventQueueState(IComparable queueName)
        {
            EventQueueState queueState = GetQueue(queueName);
            if (queueState.Dirty)
            {
                string message =
                    string.Format(CultureInfo.CurrentCulture, ExecutionStringManager.QueueBusyException, new object[] { queueName });

                throw new QueueException(message, MessageQueueErrorCode.QueueNotAvailable);
            }

            return queueState;
        }

        private void NewQueue(IComparable queueID, bool enabled, bool transactional)
        {
            WorkflowTrace.Runtime.TraceInformation("Queuing Service: Creating new Queue with ID {0} for {1}", queueID.GetHashCode(), queueID);

            if (this.persistedQueueStates.ContainsKey(queueID))
            {
                object[] args =
                    new object[] { System.Messaging.MessageQueueErrorCode.QueueExists, queueID };
                string message =
                    string.Format(CultureInfo.CurrentCulture, ExecutionStringManager.EventQueueException, args);

                throw new QueueException(message, MessageQueueErrorCode.QueueExists);
            }

            EventQueueState queueState = new EventQueueState();
            queueState.Enabled = enabled;
            queueState.queueName = queueID;
            queueState.Transactional = transactional;
            this.persistedQueueStates.Add(queueID, queueState);
        }

        internal EventQueueState GetQueue(IComparable queueID)
        {
            EventQueueState queue;
            if (this.persistedQueueStates.TryGetValue(queueID, out queue))
            {
                queue.queueName = queueID;
                return queue;
            }

            object[] args =
                new object[] { System.Messaging.MessageQueueErrorCode.QueueNotFound, queueID };
            string message =
                string.Format(CultureInfo.CurrentCulture, ExecutionStringManager.EventQueueException, args);

            throw new QueueException(message, MessageQueueErrorCode.QueueNotFound);
        }

        internal IEnumerable<IComparable> QueueNames
        {
            get
            {
                List<IComparable> list = new List<IComparable>(this.persistedQueueStates.Keys);
                foreach (IComparable name in list)
                {
                    if (name is String && (String)name == pendingNotification)
                    {
                        list.Remove(name);
                        break;
                    }
                }
                return list;
            }
        }

        private void ApplyChangesFrom(EventQueueState srcPendingQueueState, Dictionary<IComparable, EventQueueState> srcPersistedQueueStates)
        {
            lock (SyncRoot)
            {
                Dictionary<IComparable, EventQueueState> modifiedItems = new Dictionary<IComparable, EventQueueState>();

                foreach (KeyValuePair<IComparable, EventQueueState> mergeItem in srcPersistedQueueStates)
                {
                    Debug.Assert(mergeItem.Value.Transactional, "Queue inside a transactional context is not transactional!");

                    if (mergeItem.Value.Transactional)
                    {
                        if (this.persistedQueueStates.ContainsKey(mergeItem.Key))
                        {
                            EventQueueState oldvalue = this.persistedQueueStates[mergeItem.Key];
                            if (!oldvalue.Dirty)
                            {
                                // we could get here when there
                                // are conflicting create Qs
                                string message =
                                    string.Format(CultureInfo.CurrentCulture, ExecutionStringManager.QueueBusyException, new object[] { mergeItem.Key });

                                throw new QueueException(message, MessageQueueErrorCode.QueueNotAvailable);
                            }
                        }
                        modifiedItems.Add(mergeItem.Key, mergeItem.Value);
                    }
                }

                // no conflicts detected now make the updates visible
                foreach (KeyValuePair<IComparable, EventQueueState> modifiedItem in modifiedItems)
                {
                    // shared queue in the root, swap out to new value 
                    // or add new item
                    this.persistedQueueStates[modifiedItem.Key] = modifiedItem.Value;
                }

                this.pendingQueueState.CopyFrom(srcPendingQueueState);
            }
        }

        // message arrival async notification
        private void NotifyExternalSubscribers(IComparable queueName, EventQueueState qState, Object eventInstance)
        {
            NotifySynchronousSubscribers(queueName, qState, eventInstance);
            NotifyAsynchronousSubscribers(queueName, qState, 1);
        }

        private void NotifySynchronousSubscribers(IComparable queueName, EventQueueState qState, Object eventInstance)
        {
            QueueEventArgs args = new QueueEventArgs(queueName);

            for (int i = 0; i < qState.SynchronousListeners.Count; ++i)
            {
                if (qState.SynchronousListeners[i].HandlerDelegate != null)
                    qState.SynchronousListeners[i].HandlerDelegate(new WorkflowQueue(this, queueName), args);
                else
                    qState.SynchronousListeners[i].EventListener.OnEvent(new WorkflowQueue(this, queueName), args);
            }
        }

        // returns a valid state only if transactional and entry exists
        private EventQueueState MarkQueueDirtyIfTransactional(IComparable queueName)
        {
            lock (SyncRoot)
            {
                Debug.Assert(this.rootQueuingService == null, "MarkQueueDirty should be done at root");

                if (!this.persistedQueueStates.ContainsKey(queueName))
                    return null;

                EventQueueState queueState = GetQueue(queueName);

                if (!queueState.Transactional)
                    return null;

                if (queueState.Dirty)
                    return queueState; // already marked

                queueState.Dirty = true;

                if (this.dirtyQueues == null)
                    this.dirtyQueues = new List<IComparable>();

                // add to the list of dirty queues
                this.dirtyQueues.Add(queueName);

                return queueState;
            }
        }

        private void AddMessageArrivedEventHandler(WorkflowQueuingService handler)
        {
            lock (SyncRoot)
            {
                if (this.messageArrivalEventHandlers == null)
                    this.messageArrivalEventHandlers = new List<WorkflowQueuingService>();
                this.messageArrivalEventHandlers.Add(handler);
            }
        }

        private void RemoveMessageArrivedEventHandler(WorkflowQueuingService handler)
        {
            lock (SyncRoot)
            {
                if (this.messageArrivalEventHandlers != null)
                    this.messageArrivalEventHandlers.Remove(handler);

                if (this.dirtyQueues != null)
                {
                    foreach (IComparable queueName in this.dirtyQueues)
                    {
                        EventQueueState qState = GetQueue(queueName);
                        qState.Dirty = false;
                    }
                }
            }
        }
        #endregion

        #region inner QueuingService functions
        private bool IsTransactionalQueue(IComparable queueName)
        {
            // check inner service for existense
            if (!this.persistedQueueStates.ContainsKey(queueName))
            {
                EventQueueState queueState = this.rootQueuingService.MarkQueueDirtyIfTransactional(queueName);

                if (queueState != null)
                {
                    // if transactional proceed to the inner queue service 
                    // for this operation after adding the state                    
                    EventQueueState snapshotState = new EventQueueState();
                    snapshotState.CopyFrom(queueState);
                    this.persistedQueueStates.Add(queueName, snapshotState);
                    return true;
                }

                return false;
            }

            return true; // if entry exits, it must be transactional
        }

        private void SubscribeForRootMessageDelivery()
        {
            if (this.rootQueuingService == null)
                return;
            this.rootQueuingService.AddMessageArrivedEventHandler(this);
        }

        private void UnSubscribeFromRootMessageDelivery()
        {
            if (this.rootQueuingService == null)
                return;
            this.rootQueuingService.RemoveMessageArrivedEventHandler(this);
        }

        // listen on its internal(parent) queuing service 
        // messages and pull messages. There is one parent queuing service visible to the external
        // host environment. A queueing service snapshot exists per atomic scope and external messages
        // for existing queues need to be pushed through
        private void OnItemEnqueued(IComparable queueName, object item)
        {
            if (this.persistedQueueStates.ContainsKey(queueName))
            {
                // make the message visible to inner queueing service
                EventQueueState qState = GetQueue(queueName);
                if (!qState.Enabled)
                {
                    object[] msgArgs = new object[] { System.Messaging.MessageQueueErrorCode.QueueNotFound, queueName };
                    string message = string.Format(CultureInfo.CurrentCulture, ExecutionStringManager.EventQueueException, msgArgs);
                    throw new QueueException(message, MessageQueueErrorCode.QueueNotAvailable);
                }
                qState.Messages.Enqueue(item);
                NotifyExternalSubscribers(queueName, qState, item);
            }
        }

        private void OnItemSafeEnqueued(IComparable queueName, object item)
        {
            if (this.persistedQueueStates.ContainsKey(queueName))
            {
                // make the message visible to inner queueing service
                EventQueueState qState = GetQueue(queueName);
                if (!qState.Enabled)
                {
                    object[] msgArgs = new object[] { System.Messaging.MessageQueueErrorCode.QueueNotFound, queueName };
                    string message = string.Format(CultureInfo.CurrentCulture, ExecutionStringManager.EventQueueException, msgArgs);
                    throw new QueueException(message, MessageQueueErrorCode.QueueNotAvailable);
                }
                qState.Messages.Enqueue(item);
                NotifySynchronousSubscribers(queueName, qState, item);
            }
        }

        internal void Complete(bool commitSucceeded)
        {
            if (commitSucceeded)
            {
                this.rootQueuingService.ApplyChangesFrom(this.pendingQueueState, this.persistedQueueStates);
            }

            UnSubscribeFromRootMessageDelivery();
        }
        #endregion

        #region Pre-persist and Post-persist helpers for queuing service states

        // Created for ref. 20575
        internal void PostPersist(bool isPersistSuccessful)
        {
            // If persist is unsuccessful, we'll undo the changes done
            //   because of the call to .Complete() in PrePresist
            if (!isPersistSuccessful)
            {
                Debug.Assert(rootWorkflowExecutor.CurrentAtomicActivity != null);
                Debug.Assert(pendingQueueStateSnapshot != null);
                Debug.Assert(persistedQueueStatesSnapshot != null);

                TransactionalProperties transactionalProperties = rootWorkflowExecutor.CurrentAtomicActivity.GetValue(WorkflowExecutor.TransactionalPropertiesProperty) as TransactionalProperties;
                Debug.Assert(transactionalProperties != null);

                // Restore queuing states and set root activity's dependency properties to the new values.
                pendingQueueState = pendingQueueStateSnapshot;
                persistedQueueStates = persistedQueueStatesSnapshot;
                rootWorkflowExecutor.RootActivity.SetValue(WorkflowQueuingService.RootPersistedQueueStatesProperty, persistedQueueStatesSnapshot);
                rootWorkflowExecutor.RootActivity.SetValue(WorkflowQueuingService.PendingMessagesProperty, pendingQueueStateSnapshot.Messages);

                // Also call Subscribe...() because the .Complete() call called Unsubscribe
                transactionalProperties.LocalQueuingService.SubscribeForRootMessageDelivery();
            }

            // The backups are no longer necessary.
            // The next call to PrePresistQueuingServiceState() will do a re-backup.
            persistedQueueStatesSnapshot = null;
            pendingQueueStateSnapshot = null;
        }

        // Created for ref. 20575
        internal void PrePersist()
        {
            if (rootWorkflowExecutor.CurrentAtomicActivity != null)
            {
                // Create transactionalProperties from currentAtomicActivity                
                TransactionalProperties transactionalProperties = this.rootWorkflowExecutor.CurrentAtomicActivity.GetValue(WorkflowExecutor.TransactionalPropertiesProperty) as TransactionalProperties;

                // Create backup snapshot of root queuing service's persistedQueuesStates
                // qService.persistedQueueStates is changed when LocalQueuingService.Complete is called later.
                persistedQueueStatesSnapshot = new Dictionary<IComparable, EventQueueState>();
                foreach (KeyValuePair<IComparable, EventQueueState> kv in persistedQueueStates)
                {
                    EventQueueState individualPersistedQueueStateValue = new EventQueueState();
                    individualPersistedQueueStateValue.CopyFrom(kv.Value);
                    persistedQueueStatesSnapshot.Add(kv.Key, individualPersistedQueueStateValue);
                }

                // Create backup snapshot of root queuing service's pendingQueueState
                // qService.pendingQueueState is changed when LocalQueuingService.Complete is called later.
                pendingQueueStateSnapshot = new EventQueueState();
                pendingQueueStateSnapshot.CopyFrom(pendingQueueState);

                // Reconcile differences between root and local queuing services.
                transactionalProperties.LocalQueuingService.Complete(true);
            }
        }


        #endregion Pre-persist and post-persist helpers for queuing service states
    }
}