OutputQueue

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ComponentModel;
using System.Diagnostics;
using System.Threading;
using System.Threading.Parallel;

namespace DbMon.NET {

    public partial class OutputQueue<T>:Component {
        /// <summary>
        /// Max size of queue
        /// </summary>
        const int MAX_QUEUE_SIZE = 5000;

        //internal queue
        readonly ConcurrentQueue<T> _outputQueue = new ConcurrentQueue<T>();
        readonly object _syncRoot = new object();

        AsyncOperation _outputOperation { get; set; }
        AutoResetEvent _stopResetEvent = new AutoResetEvent(false);

        long _enabled = 1;
        long _state;

        public Action<OutputQueue<T>,OutputArgs<T>> Output;

        #region ctr & Properties
        /// <summary>
        /// Gets or Sets the maximum queue size, when the maximum is reached pause notifications are send to the DebugOutputQueue subscribers
        /// </summary>
        public int MaxQueueSize { get; set; }

        public object SyncRoot {
            get {
                return _syncRoot;
            }
        }
        public bool IsEnabled {
            get {
                return ParallelTask.GetRefBool(ref _enabled);
            }
            set {
                ParallelTask.SetRefBool(value,ref _enabled);
            }
        }
        public OutputQueueState State {
            get {
                return (OutputQueueState)Interlocked.Read(ref _state);
            }
            set {
                Interlocked.Exchange(ref _state,(long)value);
            }
        }

        public bool IsStarted {
            get {
                if(State == OutputQueueState.Started)
                    return true;
                return false;
            }

        }
        public bool IsPaused {
            get {
                if(State == OutputQueueState.Paused)
                    return true;
                return false;
            }

        }
        public bool IsStoppedOrPending {
            get {
                if(State == OutputQueueState.Stopped || State == OutputQueueState.StopPending)
                    return true;
                return false;
            }

        }
        public OutputQueue() {
            MaxQueueSize = MAX_QUEUE_SIZE;
            State = OutputQueueState.None;
            InitializeComponent();
        }
        public OutputQueue(IContainer container)
            : this() {
            container.Add(this);
        }
        #endregion

        #region Stop, Pause, & Resume
        volatile bool IsPurging;

        public bool CanStart(bool throwIfFalse) {
            try {
                if(IsStarted || State == OutputQueueState.StartPending)
                    throw new ArgumentException("Alread started");
                if(State == OutputQueueState.StopPending)
                    throw new ArgumentException("Stop pending");

                return true;
            }
            catch {
                if(throwIfFalse)
                    throw;
                return false;
            }
        }
        public bool CanStop(bool throwIfFalse) {
            try {
                if(IsStoppedOrPending)
                    throw new ArgumentException("Already stopped");
                return true;
            }
            catch {
                if(throwIfFalse)
                    throw;
                return false;
            }
        }
        public bool CanPause(bool throwIfFalse) {
            try {
                if(IsPaused)
                    throw new ArgumentException("Already paused");
                if(!IsStarted)
                    throw new ArgumentException("Not started");
                if(State == OutputQueueState.StartPending)
                    throw new ArgumentException("Not started yet");
                if(IsStoppedOrPending)
                    throw new ArgumentException("Already stopped or is stop pending");
                return true;
            }
            catch {
                if(throwIfFalse)
                    throw;
                return false;
            }
        }
        public bool CanResume(bool throwIfFalse) {
            try {
                if(!IsPaused)
                    throw new ArgumentException("Not paused");
                if(!IsStarted)
                    throw new ArgumentException("Not started");
                if(State == OutputQueueState.StartPending)
                    throw new ArgumentException("Not started yet");
                if(IsStoppedOrPending)
                    throw new ArgumentException("Already stopped or is stop pending");
                return true;
            }
            catch {
                if(throwIfFalse)
                    throw;
                return false;
            }
        }

        public void Stop() {
            Stop(false);
        }
        public void Stop(bool waitOnStopped) {
            if(!CanStop(true))
                return;

            State = OutputQueueState.StopPending;
            if(waitOnStopped) {
                _stopResetEvent.WaitOne();
            }
        }
        public void Pause() {
            if(!CanPause(true))
                return;
            State = OutputQueueState.Paused;
        }
        public void Resume() {
            if(!CanResume(true))
                return;
            State = OutputQueueState.Started;
        }
        #endregion

        public void Start() {
            if(!CanStart(true))
                return;
            State = OutputQueueState.StartPending;

            _outputOperation = AsyncOperationManager.CreateOperation(null);

            ThreadPool.QueueUserWorkItem((s) => {
                runningThread();
            });
        }
        public void EnqueueOutput(T item) {
            _outputQueue.Enqueue(item);
        }
        public void EnqueueOutput(T[] items) {
            foreach(var item in items) {
                _outputQueue.Enqueue(item);
            }
        }
        public void PurgeAll() {

            try {
                IsPurging = true;

                lock(this.SyncRoot) {
                    while(_outputQueue.Count > 0) {
                        try {
                            T qitem = default(T);
                            if(!_outputQueue.TryDequeue(out qitem))
                                break;
                        }
                        catch { }
                    }
                }
            }
            finally {
                IsPurging = false;
            }
        }

        void runningThread() {

            State = OutputQueueState.Started;
            _stopResetEvent.Reset();

            try {
                while(!IsStoppedOrPending) {
                    try {
                        if(IsPurging) {
                            Thread.Sleep(10);
                            continue;
                        }

                        if(IsPaused) {
                            SleepSpin(200);
                            continue;
                        }
                        //dequeue and broadcast items
                        dequeueQueue();
                    }
                    catch(Exception err) {
                        Trace.WriteLine("ERROR: runningThread() -> dequeueQueue() -> " + err.ToString());
                        SleepSpin(500);
                    }
                }
            }
            finally {
                State = OutputQueueState.Stopped;
                _stopResetEvent.Set();
            }
        }

        void dequeueQueue() {
            if(!IsEnabled)
                return;

            List<T> list = new List<T>();
            try {

                while(_outputQueue.Count > 0 && list.Count < this.MaxQueueSize) {

                    if(IsPurging) {
                        list.Clear();
                        return;
                    }

                    if(IsStoppedOrPending)
                        return;

                    T qitem = default(T);

                    if(_outputQueue.TryDequeue(out qitem))
                        list.Add(qitem);

                }

            }
            finally {

                if(list.Count > 0)
                    OnOutput(new OutputArgs<T>(list.ToArray()));
            }
            Thread.Sleep(1);
        }
        protected void OnOutput(OutputArgs<T> args) {
            if(IsPurging)
                return;

            if(Output != null)
                Output(this,args);
        }
        protected void SleepSpin(int millisecondsTimeout) {
            int spinby = 10;
            int iterations = (millisecondsTimeout / spinby);
            for(int i = 0;i < iterations;i++) {
                if(IsStoppedOrPending)
                    return;
                Thread.Sleep(spinby);
            }
        }
    }
    public class OutputArgs<T>:EventArgs {
        public T[] Items { get; set; }
        public OutputArgs(T[] items) {
            this.Items = items;
        }
    }
    public enum OutputQueueState:long {
        None,
        StartPending,
        Started,
        StopPending,
        Stopped,
        Paused
    }
}