MSMQMonitorComponent

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Messaging;
using System.Threading;

namespace DbMon.NET {
    /// <summary>
    /// Monitors a msmq queue and outputs the contents via 'OutputQueue'
    /// </summary>
    public partial class MSMQMonitorComponent:Component {

        bool _enabled;
        bool _started;
        bool _cancelled;
        bool _paused;

        OutputQueue<DebugItem> _outputQueue;
        public bool PurgeOnStartup { get; set; }

        public bool Enabled {
            get {
                return _enabled;
            }
            set {
                _enabled = value;
            }
        }

        public bool Started {
            get {
                return _started;
            }
            protected set {
                _started = value;
            }
        }
        public bool Cancelled {
            get {
                return _cancelled;
            }
            protected set {
                _cancelled = value;
            }
        }
        public bool IsPaused {
            get { return _paused; }
            protected set { _paused = value; }
        }

        public string QueuePath {
            get {
                return _msmq.Path;
            }
        }
        public MSMQMonitorComponent() {
            _enabled = true;
            InitializeComponent();
        }
        public MSMQMonitorComponent(IContainer container) {
            container.Add(this);
            InitializeComponent();
        }
        public virtual void Pause() {
            IsPaused = true;
        }
        public virtual void Resume() {
            IsPaused = false;
            beginReceive(_msmq);
        }
        public virtual void Start(string queuePath,OutputQueue<DebugItem> outputQueue) {
            if(Started)
                throw new InvalidOperationException("Already Started");
            _outputQueue = outputQueue;

            Started = true;
            Cancelled = false;
            string path = queuePath ?? @".\private$\dbmon";
            if(!MessageQueue.Exists(path))
                MessageQueue.Create(path);

            _msmq.Path = path;
            _msmq.Formatter = new ActiveXMessageFormatter();
            _msmq.ReceiveCompleted += msmq_ReceiveCompleted;

            if(PurgeOnStartup) {
                _msmq.Purge();
                writePurgeMessage();
            }
            beginReceive(_msmq);
        }
        void writePurgeMessage() {
            var message = new Message("Queue Purged");
            message.Formatter = new ActiveXMessageFormatter();
            message.Priority = MessagePriority.Highest;
            _msmq.Send(message);
        }
        public virtual void Stop() {
            Cancelled = true;
            _msmq.ReceiveCompleted -= msmq_ReceiveCompleted;
            Started = false;
        }
        void msmq_ReceiveCompleted(object sender,System.Messaging.ReceiveCompletedEventArgs e) {
            if(Cancelled)
                return;

            int pid = -1;
            MessageQueue queue = e.AsyncResult.AsyncState as MessageQueue;
            DebugItem di = new DebugItem(pid,queue.QueueName);
            bool timedOut = false;

            try {
                var msg = queue.EndReceive(e.AsyncResult);
                var body = msg.Body;
                if(body == null)
                    body = string.Empty;
                if(checkPurgeQueue(queue,msg,body)) {
                    body += " PURGED @ " + DateTime.Now.ToUniversalTime() + " UTC";
                }
                di = new DebugItem(pid,body.ToString()) {
                    ProcessName = queue.Path
                };
            }
            catch(MessageQueueException mqerr) {
                var code = mqerr.MessageQueueErrorCode;
                if(code == MessageQueueErrorCode.IOTimeout) {
                    timedOut = true;
                    return;
                }
                else {
                    di = new DebugItem(pid,"MSMQ ERROR -> " + mqerr.Message + ";Code -> " + code.ToString());
                }
            }
            catch(Exception err) {
                di = new DebugItem(pid,"CRITICAL ERROR -> " + err.Message + ";Details -> " + err.ToString());
            }
            finally {
                try {
                    if(!timedOut)
                        _outputQueue.EnqueueOutput(di);
                }
                finally {
                    if(!Cancelled)
                        beginReceive(queue);
                }
            }
        }
        void beginReceive(MessageQueue queue) {
            if(!Enabled)
                return;
            if(IsPaused)
                return;
            var timeOut = TimeSpan.FromSeconds(10);
            queue.BeginReceive(timeOut,queue);
        }
        bool checkPurgeQueue(MessageQueue queue,Message msg,object body) {
            string path = queue.Path;
            string bodyMsg = body.ToString();
            if(bodyMsg == path + MSMQWriter.PURGE_TOKEN) {
                queue.Purge();
                return true;
            }

            return false;
        }
    }
}