/* * 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.Threading; using System.Collections; using System.Collections.Generic; namespace QuantConnect.Lean.Engine.DataFeeds.Enumerators { /// /// An implementation of that relies on the /// method being called and only ends when /// is called /// /// The item type yielded by the enumerator public class EnqueueableEnumerator : IEnumerator { private T _current; private bool _end; private readonly bool _isBlocking; private long _consumerCount; private Queue _consumer = new(); private Queue _producer = new(); private readonly object _lock = new object(); private readonly ManualResetEventSlim _resetEvent = new(false); /// /// Gets the current number of items held in the internal queue /// public int Count { get { lock (_lock) { if (_end) return 0; return _producer.Count + (int)Interlocked.Read(ref _consumerCount); } } } /// /// Returns true if the enumerator has finished and will not accept any more data /// public bool HasFinished { get { return _end; } } /// /// Initializes a new instance of the class /// /// Specifies whether or not to use the blocking behavior public EnqueueableEnumerator(bool blocking = false) { _isBlocking = blocking; } /// /// Enqueues the new data into this enumerator /// /// The data to be enqueued public void Enqueue(T data) { lock (_lock) { _producer.Enqueue(data); // most of the time this will be set if(!_resetEvent.IsSet) { _resetEvent.Set(); } } } /// /// Signals the enumerator to stop enumerating when the items currently /// held inside are gone. No more items will be added to this enumerator. /// public void Stop() { lock (_lock) { if (_end) return; _end = true; // no more items can be added, so no need to wait anymore _resetEvent.Set(); _resetEvent.Dispose(); } } /// /// Advances the enumerator to the next element of the collection. /// /// /// true if the enumerator was successfully advanced to the next element; false if the enumerator has passed the end of the collection. /// /// The collection was modified after the enumerator was created. 2 public bool MoveNext() { // we read with no lock most of the time if (_consumer.TryDequeue(out _current)) { Interlocked.Decrement(ref _consumerCount); return true; } bool ended; do { var producer = _producer; lock (_lock) { // swap queues ended = _end; _producer = _consumer; } _consumer = producer; if(_consumer.Count > 0) { _current = _consumer.Dequeue(); Interlocked.Exchange(ref _consumerCount, _consumer.Count); break; } // if we are here no queue has data if (ended) { return false; } if (_isBlocking) { try { _resetEvent.Wait(Timeout.Infinite); _resetEvent.Reset(); } catch (ObjectDisposedException) { // can happen if disposed } } else { break; } } while (!ended); // even if we don't have data to return, we haven't technically // passed the end of the collection, so always return true until // the enumerator is explicitly disposed or ended return true; } /// /// Sets the enumerator to its initial position, which is before the first element in the collection. /// /// The collection was modified after the enumerator was created. 2 public void Reset() { throw new NotImplementedException("EnqueableEnumerator.Reset() has not been implemented yet."); } /// /// Gets the element in the collection at the current position of the enumerator. /// /// /// The element in the collection at the current position of the enumerator. /// public T Current { get { return _current; } } /// /// Gets the current element in the collection. /// /// /// The current element in the collection. /// /// 2 object IEnumerator.Current { get { return Current; } } /// /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// /// 2 public void Dispose() { Stop(); } } }