AsyncWorkerCollection

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

namespace  Core {

    /// <summary>
    /// A class to process a collection of items in a threaded controlled manner
    /// </summary>
    /// <typeparam name="T"></typeparam>
    [DebuggerDisplay("Count = {Count}, Workers = {_numberOfWorkers}, Type = {typeof(T)}")]
    public class AsyncWorkerCollection<T>:IDisposable{

        readonly int _numberOfWorkers;
        readonly Action<T> _processAction;
        readonly BlockingCollection<T> _collection;
        readonly CancellationTokenSource _cancelSource;
        readonly Task[] _workerTasks;

        /// <summary>
        /// ctor
        /// </summary>
        /// <param name="numberOfWorkers">The number of worker threads to process the collection</param>
        /// <param name="processAction">The action delegate to process each item</param>
        public AsyncWorkerCollection(int numberOfWorkers,Action<T> processAction) {

            Argument.IsNotNegativeOrZero(numberOfWorkers,"numberOfThreads");
            Argument.IsNotNull(processAction,"processAction");

            _numberOfWorkers = numberOfWorkers;
            _processAction = processAction;
            _cancelSource = new CancellationTokenSource();
            _collection = new BlockingCollection<T>();
            _workerTasks = new Task[_numberOfWorkers];

            StartWorkers();
        }

        /// <summary>
        /// Attempts to add the specified item to the System.Collections.Concurrent.BlockingCollection<T>.
        /// </summary>
        /// <param name="item">The item to be added to the collection and processed</param>
        /// <returns>True if item could be added; otherwise false because <see cref="IsAddingCompleted"/> is true.</returns>
        public bool Add(T item) {

            this.EnsureNotDisposed();
            if(_collection.TryAdd(item)) {
                return true;
            }
            return false;
        }
        /// <summary>
        /// Add the specified items to the System.Collections.Concurrent.BlockingCollection<T>.
        /// </summary>
        public void AddRange(IEnumerable<T> items) {

            this.EnsureNotDisposed();

            foreach(T item in items)
                this.Add(item);
        }

        public int Count {
            get {
                return _collection.Count;
            }
        }

        /// <summary>
        ///  Gets whether this System.Collections.Concurrent.BlockingCollection<T> has been marked as complete for adding.
        /// </summary>
        public bool IsAddingCompleted {
            get {
                this.EnsureNotDisposed();
                return _collection.IsAddingCompleted;
            }
        }

        /// <summary>
        /// Waits for all of the provided System.Threading.Tasks.Task objects to complete
        /// </summary>
        public void Wait() {
            this.CompleteAdding();
            Task.WaitAll(_workerTasks,_cancelSource.Token);
        }

        /// <summary>
        /// Waits for all of the provided System.Threading.Tasks.Task objects to complete
        /// </summary>
        /// <param name="timeout">A System.TimeSpan that represents the number of milliseconds to wait, or a System.TimeSpan that represents -1 milliseconds to wait indefinitely.</param>
        /// <returns>true if the System.Threading.Tasks.Task completed execution within the allotted time; otherwise, false.</returns>
        public bool Wait(TimeSpan timeout) {
            Argument.IsNotNull(timeout,"timeout");
            this.CompleteAdding();
            return Task.WaitAll(_workerTasks,(int)timeout.TotalMilliseconds,_cancelSource.Token);
        }
        /// <summary>
        ///
        /// </summary>
        /// <param name="milliseconds">The number of milliseconds to wait, or System.Threading.Timeout.Infinite (-1) to wait indefinitely.</param>
        /// <returns>true if the System.Threading.Tasks.Task completed execution within the allotted time; otherwise, false.</returns>
        public bool Wait(int milliseconds) {
            Argument.IsNotOutOfRange(milliseconds,-1,int.MaxValue,"milliseconds");
            this.CompleteAdding();
            return Task.WaitAll(_workerTasks,milliseconds,_cancelSource.Token);
        }

        /// <summary>
        ///  Marks the collection as not accepting any more additions.
        /// </summary>
        public void CompleteAdding() {

            this.EnsureNotDisposed();

            if(!_collection.IsAddingCompleted)
                _collection.CompleteAdding();
        }
        /// <summary>
        /// True if collection has been cancelled
        /// </summary>
        public bool IsCancellationRequested {
            get {
                this.EnsureNotDisposed();
                return _cancelSource.IsCancellationRequested;
            }
        }

        /// <summary>
        /// Communicates a request for cancellation.
        /// </summary>
        public void Cancel() {
            this.EnsureNotDisposed();

            if(!_cancelSource.IsCancellationRequested)
                _cancelSource.Cancel();
        }

        /// <summary>
        /// Starts the worker threads on ctor
        /// </summary>
        void StartWorkers() {
            this.EnsureNotDisposed();

            for(int i = 0;i < _numberOfWorkers;i++) {

                _workerTasks[i] = Task.Factory.StartNew((state) =>
                {
                    Tuple<int> parms = (Tuple<int>)state;
                    int workerId = parms.Item1;

                    foreach(T item in _collection.GetConsumingEnumerable(_cancelSource.Token)) {
                        try {
                            //process action on each item
                            _processAction(item);
                        }
                        catch { }
                    }

                },Tuple.Create<int>(i));

            }
        }

        #region IDisposable Pattern

        protected bool IsDisposed { get; private set; }

        [System.Diagnostics.DebuggerStepThrough]
        protected void EnsureNotDisposed() {
            if(IsDisposed)
                throw new ObjectDisposedException(GetType().Name);
        }

        [System.Diagnostics.DebuggerStepThrough]
        public void Dispose() {
            if(IsDisposed)
                throw new ObjectDisposedException(this.GetType().Name);
            try {
                this.Dispose(true);
            }
            finally {
                GC.SuppressFinalize(this);
            }
        }

        [System.Diagnostics.DebuggerStepThrough]
        protected virtual void Dispose(bool disposing) {
            if(IsDisposed)
                return;
            if(!disposing)
                return;

            try {
                try {

                    if(!this._collection.IsAddingCompleted) {
                        this._collection.CompleteAdding();
                    }

                }
                finally {

                    try {
                        this._collection.Dispose();
                    }
                    finally {

                        try {
                            _workerTasks.ForEach(t => t.Dispose());
                        }
                        finally {
                            this._cancelSource.Dispose();
                        }
                    }
                }
            }
            finally {
                this.IsDisposed = true;
            }
        }
        #endregion
    }
}