/*
 * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
 * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
*/

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

namespace QuantConnect.Util
{
    /// <summary>
    /// A small wrapper around <see cref="BlockingCollection{T}"/> used to communicate busy state of the items
    /// being processed
    /// </summary>
    /// <typeparam name="T">The item type being processed</typeparam>
    public class BusyBlockingCollection<T> : IDisposable
    {
        private readonly BlockingCollection<T> _collection;
        private readonly ManualResetEventSlim _processingCompletedEvent;
        private readonly object _lock = new object();

        /// <summary>
        /// Gets true if an item is currently being processed or needs to be processed,
        /// gets false when the collection is empty and the consumer is blocked, waiting
        /// for more items to process
        /// </summary>
        public bool IsBusy
        {
            get { return !_processingCompletedEvent.IsSet; }
        }

        /// <summary>
        /// Gets the number of items held within this collection
        /// </summary>
        public int Count
        {
            get { return _collection.Count; }
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="BusyBlockingCollection{T}"/> class
        /// with a bounded capacity of <see cref="int.MaxValue"/>
        /// </summary>
        public BusyBlockingCollection()
            : this(int.MaxValue)
        {
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="BusyBlockingCollection{T}"/> class
        /// with the specified <paramref name="boundedCapacity"/>
        /// </summary>
        /// <param name="boundedCapacity">The maximum number of items allowed in the collection</param>
        public BusyBlockingCollection(int boundedCapacity)
        {
            _collection = new BlockingCollection<T>(boundedCapacity);

            // initialize as not busy
            _processingCompletedEvent = new ManualResetEventSlim(true);
        }

        /// <summary>
        /// Adds the items to this collection
        /// </summary>
        /// <param name="item">The item to be added</param>
        public void Add(T item)
        {
            Add(item, CancellationToken.None);
        }

        /// <summary>
        /// Adds the items to this collection
        /// </summary>
        /// <param name="item">The item to be added</param>
        /// <param name="cancellationToken">A cancellation token to observer</param>
        public void Add(T item, CancellationToken cancellationToken)
        {
            bool added;
            lock (_lock)
            {
                // we're adding work to be done, mark us as busy
                _processingCompletedEvent.Reset();
                added = _collection.TryAdd(item, 0, cancellationToken);
            }

            if (!added)
            {
                _collection.Add(item, cancellationToken);
            }
        }

        /// <summary>
        /// Blocks the current thread until a consumer is blocked inside <see cref="GetConsumingEnumerable()"/>
        /// This effectively waits until the consumer has 'caught up' with the producer and finished processing
        /// everything in the collection
        /// </summary>
        /// <param name="milliseconds">The number of milliseconds to wait, or <see cref="Timeout.Infinite"/> (-1), 
        /// to wait indefinitely</param>
        /// <returns>True if the event was set, false if the timeout was reached</returns>
        public bool Wait(int milliseconds)
        {
            return _processingCompletedEvent.Wait(milliseconds);
        }

        /// <summary>
        /// Blocks the current thread until a consumer is blocked inside <see cref="GetConsumingEnumerable()"/>
        /// This effectively waits until the consumer has 'caught up' with the producer and finished processing
        /// everything in the collection
        /// </summary>
        /// <param name="milliseconds">The number of milliseconds to wait, or <see cref="Timeout.Infinite"/> (-1), 
        /// to wait indefinitely</param>
        /// <param name="cancellationToken">A cancellation token to observe</param>
        /// <returns>True if the event was set, false if the timeout was reached</returns>
        public bool Wait(int milliseconds, CancellationToken cancellationToken)
        {
            return _processingCompletedEvent.Wait(milliseconds, cancellationToken);
        }

        /// <summary>
        /// Blocks the current thread until a consumer is blocked inside <see cref="GetConsumingEnumerable()"/>
        /// This effectively waits until the consumer has 'caught up' with the producer and finished processing
        /// everything in the collection
        /// </summary>
        /// <param name="timeout">The duration of time to wait, or -1 milliseconds to wait indefinitely</param>
        /// <returns>True if the event was set, false if the timeout was reached</returns>
        public bool Wait(TimeSpan timeout)
        {
            return _processingCompletedEvent.Wait(timeout);
        }

        /// <summary>
        /// Blocks the current thread until a consumer is blocked inside <see cref="GetConsumingEnumerable()"/>
        /// This effectively waits until the consumer has 'caught up' with the producer and finished processing
        /// everything in the collection
        /// </summary>
        /// <param name="timeout">The duration of time to wait, or -1 milliseconds to wait indefinitely</param>
        /// <param name="cancellationToken">A cancellation token to observe</param>
        /// <returns>True if the event was set, false if the timeout was reached</returns>
        public bool Wait(TimeSpan timeout, CancellationToken cancellationToken)
        {
            return _processingCompletedEvent.Wait(timeout, cancellationToken);
        }

        /// <summary>
        /// Marks the <see cref="BusyBlockingCollection{T}"/> as not accepting any more additions
        /// </summary>
        public void CompleteAdding()
        {
            _collection.CompleteAdding();
        }

        /// <summary>
        /// Provides a consuming enumerable for items in this collection.
        /// </summary>
        /// <returns>An enumerable that removes and returns items from the collection</returns>
        public IEnumerable<T> GetConsumingEnumerable()
        {
            return GetConsumingEnumerable(CancellationToken.None);
        }

        /// <summary>
        /// Provides a consuming enumerable for items in this collection.
        /// </summary>
        /// <param name="cancellationToken">A cancellation token to observer</param>
        /// <returns>An enumerable that removes and returns items from the collection</returns>
        public IEnumerable<T> GetConsumingEnumerable(CancellationToken cancellationToken)
        {
            while (!_collection.IsCompleted)
            {
                T item;

                // check to see if something is immediately available
                bool tookItem;

                try
                {
                    tookItem = _collection.TryTake(out item, 0, cancellationToken);
                }
                catch (OperationCanceledException)
                {
                    // if the operation was canceled, just bail on the enumeration
                    yield break;
                }

                if (tookItem)
                {
                    // something was immediately available, emit it
                    yield return item;
                    continue;
                }


                // we need to lock this with the Add method since we need to model the act of
                // taking/flipping the switch and adding/flipping the switch as one operation
                lock (_lock)
                {
                    // double check that there's nothing in the collection within a lock, it's possible
                    // that between the TryTake above and this statement, the Add method was called, so we
                    // don't want to flip the switch if there's something in the collection
                    if (_collection.Count == 0)
                    {
                        // nothing was immediately available, mark us as idle
                        _processingCompletedEvent.Set();
                    }
                }

                try
                {
                    // now block until something is available
                    tookItem = _collection.TryTake(out item, Timeout.Infinite, cancellationToken);
                }
                catch (OperationCanceledException)
                {
                    // if the operation was canceled, just bail on the enumeration
                    yield break;
                }

                if (tookItem)
                {
                    // emit the item we found
                    yield return item;
                }
            }

            // no more items to process
            _processingCompletedEvent.Set();
        }

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        /// <filterpriority>2</filterpriority>
        public void Dispose()
        {
            _collection.Dispose();
            _processingCompletedEvent.Dispose();
        }
    }
}
