SafePq.h

//
// Created by pengwang5 on 2022/12/11.
//

#ifndef FADAS_SAFEPQ_H
#define FADAS_SAFEPQ_H

#include <unordered_set>
#include <condition_variable>
#include <type_traits>
#include <thread>
#include <chrono>
#include <deque>
#include <algorithm>

template <typename ConditionVarType,  typename LockType>
struct ConditionVarTraits;

template <>
struct ConditionVarTraits<std::condition_variable, std::mutex> {
    static void initialize(std::condition_variable& cond_var) {
    }

    static void signal(std::condition_variable& cond_var) {
        cond_var.notify_one();
    }

    static void broadcast(std::condition_variable& cond_var) {
        cond_var.notify_all();
    }

    static void wait(std::condition_variable& cond_var,
                     std::unique_lock<std::mutex>& lock) {
        cond_var.wait(lock);
    }

    template<class Rep, class Period> static bool wait_for(
            std::condition_variable& cond_var,
            std::unique_lock<std::mutex>& lock,
            const std::chrono::duration<Rep, Period>& rel_time) {
        return std::cv_status::timeout == cond_var.wait_for(lock, rel_time);
    }
};

/// @class ConditionVariable
/// The ConditionVariable class wraps a operating system ConditionVariable.
///
/// In addition, it implements support for attaching and detaching workers
/// to the condition variable.
/// @tparam ThreadContainerType The type of thread Container.
/// @tparam SignalStrategyType The type of signal policy.
/// @see NotEmptySignalStrategy
/// @see NotFullSignalStrategy
template<typename ThreadContainerType, typename SignalStrategyType,
        typename ConditionVarType, typename LockType>
class ConditionVariable {
public:
    /// Initializes a new instance of the ConditionVariable class without an
    /// upper-bound.
    ConditionVariable()
            : total_workers_(0), active_workers_(0), bounded_capacity_(SIZE_MAX),
              item_count_(0) {
        ConditionVarTraits<ConditionVarType, LockType>::initialize(
                condition_var_);
    }

    ~ConditionVariable() {
    }

    // "ConditionVariable" objects cannot be copied or assigned
    ConditionVariable(const ConditionVariable&) = delete;
    ConditionVariable& operator=(const ConditionVariable&) = delete;

    /// Gets the number of workers attached to this condition variable.
    /// @return The number of workers attached to this condition variable.
    /// @see Attach
    size_t total() const {
        return total_workers_;
    }

    /// Gets the number of active workers for this condition variable.
    /// active workers are workers that are currently NOT waiting on this
    /// condition variable.
    /// @return The number of active workers.
    size_t active() const {
        return active_workers_;
    }

    /// Gets the bounded capacity of this condition variable instance.
    /// @return The bounded capacity of this condition variable.
    size_t bounded_capacity() const {
        return bounded_capacity_;
    }

    /// Sets the bounded capacity of this condition variable instance.
    void bounded_capacity(size_t capacity) {
        bounded_capacity_ = capacity;
    }

    /// Gets the number of items contained in this condition variable
    /// instance.
    /// @return The number of items
    size_t size() const {
        return item_count_;
    }

    /// Set the number of items contained in this condition variable
    /// instance.
    void size(size_t count) {
        item_count_ = count;
    }

    /// Registers the a worker with this condition variable.
    /// If the worker is already registered then this method has no effect.
    /// @see Detach
    void attach() {
        if (container_.add()) {
            increment_total();
            increment_active();
        }
    }

    /// Unregisters the worker from this condition variable.
    /// If the worker was not previously registered then this method has
    /// no effect.
    /// @see Attach
    void detach() {
        if (container_.remove()) {
            decrement_total();
            decrement_active();
        }

        if (total_workers_ > 0 && active_workers_ == 0) {
            increment_active();
            ConditionVarTraits<ConditionVarType, LockType>::signal(
                    condition_var_);
        }
    }

    /// Wakes up a worker waiting on this condition variable.
    void signal() {
        // if no workers attached always signal!
        if (total_workers_ == 0) {
            ConditionVarTraits<ConditionVarType, LockType>::signal(
                    condition_var_);
            return;
        }
        // issue a signal only when there are no active workers, or when
        // the count starts to grow beyond a threshold level
        if (signal_.should_signal(active_workers_, total_workers_,
                                  item_count_, bounded_capacity_)) {
            increment_active();
            ConditionVarTraits<ConditionVarType, LockType>::signal(
                    condition_var_);
        }
    }

    /// Wakes up all workers waiting on this condition variable.
    void broadcast() {
        if (total_workers_ != 0) {
            // set active only if workers attached
            active(total_workers_);
        }
        ConditionVarTraits<ConditionVarType, LockType>::broadcast(
                condition_var_);
    }

    /// Waits indefinitely for this condition variable to become signaled.
    /// @param lock An object of type std::unique_lock which must be locked
    /// by the current thread.
    void wait(std::unique_lock<LockType>& lock) {
        decrement_active();
        ConditionVarTraits<ConditionVarType, LockType>::wait(
                condition_var_, lock);
    }

    /// Waits up to specified duration for this condition variable to become
    /// signaled.
    /// @param lock An object of type std::unique_lock which must be locked
    /// by the current thread.
    /// @param rel_time An object of type std::chrono::duration representing
    /// the maximum time to spend waiting.
    template<class Rep, class Period> bool wait_for(
            std::unique_lock<LockType>& lock,
            const std::chrono::duration<Rep, Period>& rel_time) {
        decrement_active();

        bool timed_out =
                ConditionVarTraits<ConditionVarType, LockType>::wait_for(
                        condition_var_, lock, rel_time);

        if (timed_out) {
            increment_active();
        }

        return timed_out;
    }

private:
    /// Sets the number of active workers for this condition variable.
    /// @param active The number of active workers.
    void active(size_t active) {
        active_workers_ = active > total_workers_ ? total_workers_ : active;
    }

    /// Increments the total worker count for this condition variable by 1.
    void increment_total() {
        total_workers_ += 1;
    }

    /// Decrements the total worker count for this condition variable by 1.
    void decrement_total() {
        total_workers_ = total_workers_ > 0 ? total_workers_ - 1 : 0;
    }

    /// Increments the active worker count for this condition variable by 1.
    void increment_active() {
        if (++active_workers_ > total_workers_)
            active_workers_ = total_workers_;
    }

    /// Decrements the active worker count for this condition variable by 1.
    void decrement_active() {
        active_workers_ = active_workers_ > 0 ? active_workers_ - 1 : 0;
    }

    size_t total_workers_;
    size_t active_workers_;
    size_t bounded_capacity_;
    size_t item_count_;

    ConditionVarType condition_var_;
    ThreadContainerType container_;
    SignalStrategyType signal_;
};

/// @class NotEmptySignalStrategy
///
/// A strategy object for determining whether or not a "not empty" condition
/// variable should issue a signal.
///
/// This strategy will only return true if there are no active workers
/// (i.e. all workers are waiting
/// on empty BlockingCollection). Or when the BlockingCollection's element
/// count starts to grow beyond a
/// threshold level.
///
/// This approach minimizes condition variable sleeps, wakes and lock
/// contention. Which in turn,
/// improves performance and makes it more predictable.
/// @tparam ItemsPerThread The number of items to allow per thread.
/// @see ConditionVariable
/// @see NotFullSignalStrategy
template<size_t ItemsPerThread = 16> struct NotEmptySignalStrategy {
    bool should_signal(size_t active_workers, size_t total_workers,
                       size_t item_count, size_t /*capacity*/) const {
        return active_workers == 0 || (active_workers < total_workers &&
                                       item_count / active_workers > ItemsPerThread);
    }
};

/// @class NotFullSignalStrategy
///
/// A strategy object for determining whether or not a "not full" condition
///  variable should issue a signal.
///
/// This strategy will only return true if there are no active workers
/// (i.e. all workers are
/// waiting on a full BlockingCollection). Or when the BlockingCollection's
/// available capacity
/// starts to grow beyond a threshold level.
///
/// This approach minimizes condition variable sleeps, wakes and lock
/// contention. Which in turn,
/// improves performance and makes it more predictable.
/// @tparam ItemsPerThread The number of items to allow per thread.
/// @see ConditionVariable
/// @see NotEmptySignalStrategy
template<size_t ItemsPerThread = 16> struct NotFullSignalStrategy {
    bool should_signal(size_t active_workers, size_t total_workers,
                       size_t item_count, size_t capacity) const {
        return (active_workers == 0 || (active_workers < total_workers &&
                                        (capacity - item_count) / active_workers > ItemsPerThread));
    }
};

/// @class ConditionVariableGenerator
///
/// Generates the "not full" and "not empty" condition variables for
/// the specified ThreadContainerType.
///
/// @tparam ThreadContainerType The thread Container policy to use when
/// generating the condition variables.
template<typename ThreadContainerType, typename NotFullSignalStrategy,
        typename NotEmptySignalStrategy, typename ConditionVarType,
        typename LockType> struct ConditionVariableGenerator {
    using NotFullType = ConditionVariable<ThreadContainerType,
            NotFullSignalStrategy, ConditionVarType, LockType>;
    using NotEmptyType = ConditionVariable<ThreadContainerType,
            NotEmptySignalStrategy, ConditionVarType, LockType>;

    using lock_type = LockType;
};

template <typename T>
struct ThreadContainerTraits;

template <>
struct ThreadContainerTraits<std::thread::id> {
    static std::thread::id get_thread_id() {
        return std::this_thread::get_id();
    }
};

/// @class ThreadContainer
/// This class adds and removes the specified thread type from the
/// Container.
/// @tparam T The thread type.
template<typename T> class ThreadContainer {
public:
    ThreadContainer() {
    }

    /// Adds the calling thread to the Container.
    /// @returns True if the calling thread was added to Container.
    /// Otherwise false.
    bool add() {
        T id = ThreadContainerTraits<T>::get_thread_id();

        typename std::unordered_set<T>::iterator itr = thread_id_.find(id);

        if (itr != thread_id_.end()) {
            return false;
        }

        thread_id_.insert(id);
        return true;
    }

    /// Removes the calling thread from the Container.
    /// @returns True if the calling thread was removed from Container.
    /// Otherwise false.
    bool remove() {
        if (thread_id_.erase(ThreadContainerTraits<T>::get_thread_id())
            > 0) {
            return true;
        }
        return false;
    }

private:
    std::unordered_set<T> thread_id_;
};

namespace detail {
    struct QueueType {};
    struct StackType {};

    template< typename T >
    struct is_queue : std::false_type { };

    template<>
    struct is_queue<QueueType> : std::true_type {};

    /// @class Container
    ///
    /// Represents a first in-first out (FIFO) or a last in-first out
    /// (LIFO) collection depending on
    /// the ContainerType template parameter value.
    ///
    /// Implements the implicitly defined IProducerConsumerCollection<T>
    /// policy.
    /// @tparam T The type of items in the Container.
    /// @tparam ContainerType The type of Container (i.e. Queue or Stack).
    template<typename T, typename ContainerType>
    class Container {
    public:
        using container_type = std::deque<T>;
        using value_type = typename container_type::value_type;
        using size_type = typename container_type::size_type;

        /// Initializes a new instance of the Container<T> class.
        Container()
                : bounded_capacity_(SIZE_MAX)  {
        }

        /// Sets the max number of elements this container can hold.
        /// @param bounded_capacity The max number of elements this
        /// container can hold.
        void bounded_capacity(size_t bounded_capacity) {
            bounded_capacity_ = bounded_capacity;
        }

        /// Gets the max number of elements this container can hold.
        /// @returns The max number of elements this container can hold.
        size_t bounded_capacity() {
            return bounded_capacity_;
        }

        /// Gets the number of elements contained in the collection.
        /// @returns The number of elements contained in the collection.
        size_type size() {
            return container_.size();
        }

        /// Attempts to add an element to the collection.
        /// @param item The element to add to the collection.
        /// @returns True if the element was added successfully; otherwise,
        /// false.
        bool try_add(const value_type& item) {
            if (container_.size() == bounded_capacity_)
                return false;
            container_.push_back(item);
            return true;
        }

        /// Attempts to add an element to the collection.
        /// @param item The element to add to the collection.
        /// @returns True if the element was added successfully; otherwise,
        /// false.
        bool try_add(value_type&& item) {
            if (container_.size() == bounded_capacity_)
                return false;
            container_.push_back(std::forward<value_type>(item));
            return true;
        }

        /// Attempts to remove and return an element from the collection.
        /// @param [out] item When this method returns, if the element was
        /// removed and returned successfully, item
        /// contains the removed element. If no element was available to be
        /// removed, the value is unspecified.
        /// @returns True if an element was removed and returned
        /// successfully; otherwise, false.
        bool try_take(value_type& item) {
            if (container_.empty())
                return false;
            return try_take_i(item, is_queue<ContainerType>());
        }

        /// Attempts to add an element to the collection.
        /// This new element is constructed in place using args as the
        /// arguments for its construction.
        /// @param args Arguments forwarded to construct the new element.
        template <typename... Args> bool try_emplace(Args&&... args) {
            if (container_.size() == bounded_capacity_)
                return false;
            return try_emplace_i<Args...>(std::forward<Args>(args)...,
                                          is_queue<ContainerType>());
        }

    private:
        size_t bounded_capacity_;
        container_type container_;

        bool try_take_i(value_type& item, std::false_type) {
            item = container_.back();
            container_.pop_back();
            return true;
        }

        bool try_take_i(value_type& item, std::true_type) {
            item = container_.front();
            container_.pop_front();
            return true;
        }

        template <typename... Args> bool try_emplace_i(Args&&... args,
                                                       std::false_type) {
            container_.emplace_front(std::forward<Args>(args)...);
            return true;
        }

        template <typename... Args> bool try_emplace_i(Args&&... args,
                                                       std::true_type) {
            container_.emplace_back(std::forward<Args>(args)...);
            return true;
        }
    };
} // namespace detail

template<typename T>
using QueueContainer = detail::Container<T, detail::QueueType>;

template<typename T>
using StackContainer = detail::Container<T, detail::StackType>;

using StdConditionVariableGenerator = ConditionVariableGenerator<
        ThreadContainer<std::thread::id>, NotFullSignalStrategy<16>,
        NotEmptySignalStrategy<16>, std::condition_variable, std::mutex>;

/// @enum BlockingCollectionState
/// The BlockCollection states.
enum class BlockingCollectionState {
    // BlockingCollection is active and processing normally.
    Activated = 1,
    // BlockingCollection is deactivated; no add or take operations allowed.
    Deactivated = 2,
    // BlockingCollection was pulsed; add and take may proceed normally.
    Pulsed = 3
};

/// @enum BlockingCollectionStatus
/// The BlockCollection status codes.
/// These are the status codes returned by all of BlockingCollection's Add
/// and Take operations.
enum class BlockingCollectionStatus {
    /// Operation succeeded
    Ok = 0,
    /// Operation failed due to CompleteAdding() having been invoked
    AddingCompleted = -1,
    /// Operation failed due to time out
    TimedOut = -2,
    /// Operation failed due to BlockingCollection not being activated
    NotActivated = -3,
    /// Operation failed due to BlockingCollection being completed
    Completed = -4,
    /// Operation failed due to invalid iterators
    InvalidIterators = -5,
    /// Operation failed due to concurrent Add and CompleteAdding
    CompleteAddingConcurrent = -6,
    /// Operation failed due to BlockingCollection Container error
    InternalError = -8
};

template <typename T, typename ContainerType = QueueContainer<T>,
        typename ConditionVariableGenerator = StdConditionVariableGenerator>
class BlockingCollection {
public:
    using LockType = typename ConditionVariableGenerator::lock_type;

    /// Initializes a new instance of the BlockingCollection<T> class
    /// without an upper-bound.
    BlockingCollection()
            : BlockingCollection(SIZE_MAX) {
    }

    /// Initializes a new instance of the BlockingCollection<T> class
    /// with the specified upper-bound.
    /// @param capacity The bounded size of the collection.
    explicit BlockingCollection(size_t capacity)
            : state_(BlockingCollectionState::Activated),
              bounded_capacity_(capacity),
              is_adding_completed_(false) {
        not_empty_condition_var_.bounded_capacity(capacity);
        not_full_condition_var_.bounded_capacity(capacity);
        container_.bounded_capacity(capacity);
    }

    // "BlockingCollection" objects cannot be copied or assigned
    BlockingCollection(const BlockingCollection&) = delete;
    BlockingCollection& operator=(const BlockingCollection&) = delete;

    ~BlockingCollection() {
    }

    /// Gets the bounded capacity of this BlockingCollection<T> instance.
    /// @return The bounded capacity of the collection.
    size_t bounded_capacity() {
        std::lock_guard<LockType> guard(lock_);
        return bounded_capacity_;
    }

    /// Gets the current state of this BlockingCollection<T> instance.
    /// @return The current state of the collection.
    /// @see BlockingCollectionState
    BlockingCollectionState state() {
        std::lock_guard<LockType> guard(lock_);
        return state_;
    }

    /// Gets whether this BlockingCollection<T> instance is full.
    /// @return True if the collection is full; otherwise false.
    bool is_full() {
        std::lock_guard<LockType> guard(lock_);
        return is_full_i();
    }

    /// Gets whether this BlockingCollection<T> instance is empty.
    /// @return True if the collection is empty; otherwise false.
    bool is_empty() {
        std::lock_guard<LockType> guard(lock_);
        return is_empty_i();
    }

    /// Gets the number of items contained in the BlockingCollection<T>
    /// instance.
    /// If any method in BlockingCollection is executing while the size
    /// property is being accessd, the return value
    /// is approximate. size may reflect a number that is either greater
    /// than or less than the actual number of
    /// items in the BlockingCollection.
    /// @return The number of item in the collection.
    size_t size() {
        std::lock_guard<LockType> guard(lock_);
        return container_.size();
    }

    /// Gets whether this BlockingCollection<T> instance has been
    /// deactivated.
    /// @return True is this collection has been deactivated.
    /// Otherwise false.
    bool is_deactivated() {
        std::lock_guard<LockType> guard(lock_);
        return state_ == BlockingCollectionState::Deactivated;
    }

    /// Gets whether this BlockingCollection<T> instance has been marked
    /// as complete for adding and is empty.
    /// @return True if this collection has been marked as complete for
    /// adding and is empty. Otherwise false.
    bool is_completed() {
        std::lock_guard<LockType> guard(lock_);
        return is_completed_i();
    }

    /// Gets whether this BlockingCollection<T> instance has been marked
    /// as complete for adding.
    /// @return True if this collection has been marked as complete for
    /// adding. Otherwise false.
    bool is_adding_completed() {
        std::lock_guard<LockType> guard(lock_);
        return is_adding_completed_i();
    }

    /// Pulses this BlockingCollection<T> instance to wake up any waiting
    /// threads.
    /// Changes the collection's state to Pulsed. Future Add and Take
    /// operations proceed
    /// as in the Activated state.
    /// @return The BlockingCollection's state before this call.
    /// @see BlockingCollectionState
    BlockingCollectionState pulse() {
        std::lock_guard<LockType> guard(lock_);
        return deactivate_i(true);
    }

    /// Deactivate this BlockingCollection<T> instance and wakeup all
    /// threads waiting
    /// on the collection so they can continue. No items are removed from
    /// the collection,
    /// however. Any other operations called until the collection is
    /// activated again will immediately return
    /// BlockingCollectionStatus::NotActivated.
    /// @return The BlockingCollection's state before this call.
    /// @see BlockingCollectionState
    /// @see BlockingCollectionStatus
    BlockingCollectionState deactivate() {
        std::lock_guard<LockType> guard(lock_);
        return deactivate_i(false);
    }

    /// Reactivate this BlockingCollection<T> instance so that threads
    /// can Add and Take
    /// items again.
    /// @return The BlockingCollection's state before this call.
    /// @see BlockingCollectionState
    BlockingCollectionState activate() {
        std::lock_guard<LockType> guard(lock_);
        return activate_i();
    }

    /// Releases all items from this BlockingCollection<T> instance
    /// but does not mark it deactivated.
    /// @return The number of items flushed.
    size_t flush() {
        std::lock_guard<LockType> guard(lock_);

        auto itemsFlushed = container_.size();

        T item;

        while (container_.size() > 0) {
            container_.try_take(item);
        }

        not_empty_condition_var_.size(0);
        not_full_condition_var_.size(0);

        return itemsFlushed;
    }

    /// Marks the BlockingCollection<T> instances as not accepting any more
    /// additions.
    /// After a collection has been marked as complete for adding, adding
    /// to the collection
    /// is not permitted and attempts to remove from the collection will
    /// not wait when the collection is empty.
    void complete_adding() {
        std::lock_guard<LockType> guard(lock_);

        if (is_adding_completed_)
            return;

        is_adding_completed_ = true;

        not_empty_condition_var_.broadcast();
        not_full_condition_var_.broadcast();
    }

    /// Gets the number of consumer threads that are actively taking items
    /// from this BlockingCollection<T> instance.
    /// @return The number of active consumer threads.
    /// @see AttachConsumer
    size_t active_consumers() {
        std::lock_guard<LockType> guard(lock_);
        return not_empty_condition_var_.active();
    }

    /// Gets the number of producer threads that are actively adding items
    /// to this BlockingCollection<T> instance.
    /// @return The number of active producer threads.
    /// @see AttachProducer
    size_t active_producers() {
        std::lock_guard<LockType> guard(lock_);
        return not_full_condition_var_.active();
    }

    /// Gets the total number of consumer threads that can take items
    /// from this BlockingCollection<T> instance.
    /// @return The total number of consumer threads.
    /// @see AttachConsumer
    size_t total_consumers() {
        std::lock_guard<LockType> guard(lock_);
        return not_empty_condition_var_.total();
    }

    /// Gets the total number of producer threads that can add items
    /// to this BlockingCollection<T> instance.
    /// @return The total number of producer threads.
    /// @see AttachProducer
    size_t total_producers() {
        std::lock_guard<LockType> guard(lock_);
        return not_full_condition_var_.total();
    }

    /// Registers a consumer thread with this BlockingCollection<T>
    /// instance.
    /// @see TotalConsumers
    void attach_consumer() {
        std::lock_guard<LockType> guard(lock_);
        not_empty_condition_var_.attach();
    }

    /// Unregisters a consumer thread with this BlockingCollection<T>
    /// instance.
    /// @see TotalConsumers
    void detach_consumer() {
        std::lock_guard<LockType> guard(lock_);
        not_empty_condition_var_.detach();
    }

    /// Registers a producer thread with this BlockingCollection<T>
    /// instance.
    /// @see TotalProducers
    void attach_producer() {
        std::lock_guard<LockType> guard(lock_);
        not_full_condition_var_.attach();
    }

    /// Unregisters a producer thread with this BlockingCollection<T>
    /// instance.
    /// @see TotalProducers
    void detach_producer() {
        std::lock_guard<LockType> guard(lock_);
        not_full_condition_var_.detach();
    }

    /// Adds the given element value to the BlockingCollection<T>.
    /// The new element is initialized as a copy of value.
    /// If a bounded capacity was specified when this instance of
    /// BlockingCollection<T> was initialized,
    /// a call to Add may block until space is available to store the
    /// provided item.
    /// @param value the value of the element to add
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    BlockingCollectionStatus add(const T& value) {
        return try_emplace_timed(std::chrono::milliseconds(-1), value);
    }

    /// Adds the given element value to the BlockingCollection<T>.
    /// Value is moved into the new element.
    /// If a bounded capacity was specified when this instance of
    /// BlockingCollection<T> was initialized,
    /// a call to Add may block until space is available to store the
    /// provided item.
    /// @param value the value of the element to add
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    BlockingCollectionStatus add(T&& value) {
        return try_emplace_timed(std::chrono::milliseconds(-1),
                                 std::forward<T>(value));
    }

    /// Tries to add the given element value to the BlockingCollection<T>.
    /// The new element is initialized as a copy of value.
    /// If the collection is a bounded collection, and is full, this method
    /// immediately returns without adding the item.
    /// @param value the value of the element to try to add
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    BlockingCollectionStatus try_add(const T& value) {
        return try_emplace_timed(std::chrono::milliseconds::zero(), value);
    }

    /// Tries to add the given element value to the BlockingCollection<T>.
    /// Value is moved into the new element.
    /// If the collection is a bounded collection, and is full, this
    /// method immediately returns without adding the item.
    /// @param value the value of the element to try to add
    BlockingCollectionStatus try_add(T&& value) {
        return try_emplace_timed(std::chrono::milliseconds::zero(),
                                 std::forward<T>(value));
    }

    /// Tries to add the given element value to the BlockingCollection<T>
    /// within the specified time period.
    /// Value is moved into the new element.
    /// @param value the value of the element to try to add
    /// @param rel_time An object of type std::chrono::duration
    /// representing the maximum time to spend waiting.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/duration
    template<typename U, class Rep, class Period>
    BlockingCollectionStatus try_add_timed(U&& value,
                                           const std::chrono::duration<Rep, Period>& rel_time) {
        return try_emplace_timed(rel_time, std::forward<U>(value));
    }

    /// Adds new element to the BlockingCollection<T>.
    /// The arguments args... are forwarded to the constructor as
    /// std::forward<Args>(args)....If a bounded capacity was specified
    /// when this instance of BlockingCollection<T> was initialized,
    /// a call to Emplace may block until space is available to store the
    /// provided item.
    /// @param args arguments to forward to the constructor of the element
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    template<typename... Args>
    BlockingCollectionStatus emplace(Args&&... args) {
        return try_emplace_timed(std::chrono::milliseconds(-1),
                                 std::forward<Args>(args)...);
    }

    /// Tries to add new element to the BlockingCollection<T>.
    /// The arguments args... are forwarded to the constructor as
    /// std::forward<Args>(args)....
    /// If the collection is a bounded collection, and is full, this method
    /// immediately
    /// returns without adding the item.
    /// @param args arguments to forward to the constructor of the element
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    template<typename... Args>
    BlockingCollectionStatus try_emplace(Args&&... args) {
        return try_emplace_timed(std::chrono::milliseconds::zero(),
                                 std::forward<Args>(args)...);
    }

    /// Tries to add the given element value to the BlockingCollection<T>
    /// within the specified time period.
    /// The arguments args... are forwarded to the constructor as
    /// std::forward<Args>(args)....
    /// If the collection is a bounded collection, and is full, this
    /// method immediately returns without adding the item.
    /// @param args arguments to forward to the constructor of the element
    /// @param rel_time An object of type std::chrono::duration
    /// representing the maximum time to spend waiting.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/duration
    template<class Rep, class Period, typename... Args>
    BlockingCollectionStatus try_emplace_timed(
            const std::chrono::duration<Rep, Period>& rel_time,
            Args&&... args) {
        {
            std::unique_lock<LockType> guard(lock_);

            auto status = wait_not_full_condition(guard, rel_time);

            if (BlockingCollectionStatus::Ok != status)
                return status;

            if (!container_.try_emplace(std::forward<Args>(args)...))
                return BlockingCollectionStatus::InternalError;

            signal(container_.size(), false);
        }
        return BlockingCollectionStatus::Ok;
    }

    /// Removes an item from the BlockingCollection<T>.
    /// A call to Take may block until an item is available to be removed.
    /// @param[out] item The item removed from the collection.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    BlockingCollectionStatus take(T& item) {
        return try_take(item, std::chrono::milliseconds(-1));
    }

    /// Tries to remove an item from the BlockingCollection<T>.
    /// If the collection is empty, this method immediately returns without
    /// taking an item.
    /// @param[out] item The item removed from the collection.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    BlockingCollectionStatus try_take(T& item) {
        return try_take(item, std::chrono::milliseconds::zero());
    }

    /// Tries to remove an item from the BlockingCollection<T> in the
    /// specified time period.
    /// @param[out] item The item removed from the collection.
    /// @param rel_time An object of type std::chrono::duration
    /// representing the maximum time to spend waiting.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/duration
    template<class Rep, class Period> BlockingCollectionStatus
    try_take(T& item, const std::chrono::duration<Rep, Period>& rel_time) {
        {
            std::unique_lock<LockType> guard(lock_);

            auto status = wait_not_empty_condition(guard, rel_time);

            if (BlockingCollectionStatus::Ok != status)
                return status;

            if (!container_.try_take(item))
                return BlockingCollectionStatus::InternalError;

            signal(container_.size(), true);
        }
        return BlockingCollectionStatus::Ok;
    }

    /// Adds the items from range [first, last] to the
    /// BlockingCollection<T>.
    /// If a bounded capacity was specified when this instance of
    /// BlockingCollection<T> was initialized,
    /// a call to Add may block until space is available to store the
    /// provided items.
    /// Use std::make_move_iterator if the elements should be moved
    /// instead of copied.
    /// @param first The start range of elements to insert.
    /// @param last The end range of elements to insert.
    /// @param [out] added The actual number of elements added.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    template <typename Iterator> BlockingCollectionStatus
    add_bulk(Iterator first, Iterator last, size_t& added) {
        return try_add_bulk(first, last, added,
                            std::chrono::milliseconds(-1));
    }

    /// Tries to add the items from range [first, last] to the
    /// BlockingCollection<T>.
    /// If the collection is a bounded collection, and is full, this method
    /// immediately returns without adding the items.
    /// Use std::make_move_iterator if the elements should be moved
    /// instead of copied.
    /// @param first The start range of elements to insert.
    /// @param last The end range of elements to insert.
    /// @param [out] added The actual number of elements added.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    template <typename Iterator> BlockingCollectionStatus
    try_add_bulk(Iterator first, Iterator last, size_t& added) {
        return try_add_bulk(first, last, added,
                            std::chrono::milliseconds::zero());
    }

    /// Tries to add the specified items from the range [first, last] to
    /// the BlockingCollection<T> within
    /// the specified time period.
    /// Use std::make_move_iterator if the elements should be moved
    /// instead of copied.
    /// @param first The start range of elements to insert.
    /// @param last The end range of elements to insert.
    /// @param [out] added The actual number of elements added.
    /// @param rel_time An object of type std::chrono::duration representing
    ///  the maximum time to spend waiting.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/durations
    template<typename Iterator, class Rep, class Period>
    BlockingCollectionStatus try_add_bulk(Iterator first, Iterator last,
                                          size_t& added, const std::chrono::duration<Rep, Period>& rel_time) {
        {
            added = 0;

            std::unique_lock<LockType> guard(lock_);

            auto status = wait_not_full_condition(guard, rel_time);

            if (BlockingCollectionStatus::Ok != status)
                return status;

            if (first == last)
                return BlockingCollectionStatus::InvalidIterators;

            for (; first != last; ++first) {
                if (!container_.try_add((*first)))
                    break;
                ++added;
            }

            signal(container_.size(), false);
        }
        return BlockingCollectionStatus::Ok;
    }

    /// Takes up to count elements from the BlockingCollection<T>.
    /// A call to take_bulk may block until an element is available to be
    /// removed.
    /// Use std::make_move_iterator if the elements should be moved instead
    /// of copied.
    /// @param[out] first Contains first item taken.
    /// @param count The number of elements to take from collection.
    /// @param[out] taken The actual number of elements taken.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    template <typename Iterator> BlockingCollectionStatus
    take_bulk(Iterator first, size_t count, size_t& taken) {
        return try_take_bulk(first, count, taken,
                             std::chrono::milliseconds(-1));
    }

    /// Takes up to count elements from the BlockingCollection<T>.
    /// If the collection is empty, this method immediately returns without
    /// taking any items.
    /// Use std::make_move_iterator if the elements should be moved instead
    /// of copied.
    /// @param[out] first Contains first item taken.
    /// @param count The number of elements to take from collection.
    /// @param[out] taken The actual number of elements taken.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    template <typename Iterator> BlockingCollectionStatus
    try_take_bulk(Iterator first, size_t count, size_t& taken) {
        return try_take_bulk(first, count, taken,
                             std::chrono::milliseconds::zero());
    }

    /// Tries to take up to count elements from the BlockingCollection<T>
    /// within the specified time period.
    /// If the collection is empty, this method immediately returns without
    /// taking any items.
    /// Use std::make_move_iterator if the elements should be moved instead
    /// of copied.
    /// @param[out] first Contains first item taken.
    /// @param count The number of elements to take from collection.
    /// @param[out] taken The actual number of elements taken.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/durations
    template <typename Iterator, class Rep, class Period>
    BlockingCollectionStatus try_take_bulk(Iterator first, size_t count,
                                           size_t& taken, const std::chrono::duration<Rep, Period>& rel_time) {
        {
            taken = 0;

            if (count == 0)
                return BlockingCollectionStatus::Ok;

            std::unique_lock<LockType> guard(lock_);

            auto status = wait_not_empty_condition(guard, rel_time);

            if (BlockingCollectionStatus::Ok != status)
                return status;

            auto end = first + count;

            for (; first != end; ++first) {
                if (!container_.try_take((*first)))
                    break;

                if (++taken == count)
                    break;
            }

            signal(container_.size(), true);
        }
        return BlockingCollectionStatus::Ok;
    }

private:
    class Iterator {
    public:
        Iterator(BlockingCollection<T, ContainerType,
                ConditionVariableGenerator> &collection)
                : collection_(collection), status_(BlockingCollectionStatus::Ok),
                  wait_for_first_item(true) {
        }

        Iterator(BlockingCollection<T,
                ContainerType, ConditionVariableGenerator> &collection,
                 BlockingCollectionStatus status)
                : collection_(collection), status_(status), wait_for_first_item(false) {
        }

        // "Iterator" objects cannot be copied or assigned
        Iterator(const BlockingCollection&) = delete;
        Iterator& operator=(const Iterator&) = delete;

        bool operator!=(const Iterator& it)  {
            if (wait_for_first_item) {
                wait_for_first_item = false;
                status_ = collection_.try_take(item_,
                        // -1 forces TryTake to wait
                                               std::chrono::milliseconds(-1));
            }

            return !(status_ != BlockingCollectionStatus::Ok);
        }

        Iterator& operator++() {
            status_ = collection_.try_take(item_,
                                           std::chrono::milliseconds(-1));
            return *this;
        }

        T& operator*() {
            return item_;
        }

    private:
        BlockingCollection<T, ContainerType, ConditionVariableGenerator>
                &collection_;
        BlockingCollectionStatus status_;
        bool wait_for_first_item;
        T item_;
    };

public:
    Iterator begin() { return { *this }; }
    Iterator end()   { return { *this }; }

private:
    // the member functions below assume lock is held!

    /// The implementation for the Deactivate method.
    /// This method is not thread safe.
    /// @see Deactivate
    BlockingCollectionState deactivate_i(bool pulse) {
        auto previous_state = state_;

        if (previous_state != BlockingCollectionState::Deactivated) {
            if (pulse)
                state_ = BlockingCollectionState::Pulsed;
            else
                state_ = BlockingCollectionState::Deactivated;

            not_empty_condition_var_.broadcast();
            not_full_condition_var_.broadcast();
        }

        return previous_state;
    }

    /// The implementation for the Activate method.
    /// This method is not thread safe.
    /// @see Activate
    BlockingCollectionState activate_i() {
        auto previous_state = state_;

        state_ = BlockingCollectionState::Activated;

        return previous_state;
    }

    /// The implementation for the is_full method.
    /// This method is not thread safe.
    /// @see is_full
    bool is_full_i() {
        return bounded_capacity_ != SIZE_MAX &&
               container_.size() >= bounded_capacity_;
    }

    /// The implementation for the is_empty method.
    /// This method is not thread safe.
    /// @see is_empty
    bool is_empty_i() {
        return container_.size() == 0;
    }

    /// The implementation for the is_completed method.
    /// This method is not thread safe.
    /// @see is_completed
    bool is_completed_i() {
        return is_adding_completed_ && is_empty_i();
    }

    /// The implementation for the is_adding_completed method.
    /// This method is not thread safe.
    /// @see is_adding_completed
    bool is_adding_completed_i() {
        return is_adding_completed_;
    }

protected:
    /// Wraps the condition variable signal methods.
    /// This method updates the size property on both
    /// condition variables before invoking the signal
    /// method on the specified condition variable.
    void signal(size_t itemCount, bool signal_not_full) {
        not_empty_condition_var_.size(itemCount);
        not_full_condition_var_.size(itemCount);

        if (signal_not_full) {
            // signal only if capacity is bounded
            if (bounded_capacity_ != SIZE_MAX) {
                not_full_condition_var_.signal();
            }
        } else {
            not_empty_condition_var_.signal();
        }
    }

    /// The method waits on the "not full" condition variable whenever
    /// the collection becomes full.
    /// It atomically releases lock, blocks the current executing thread,
    /// and adds it to the
    /// list of threads waiting on the "not full" condition variable. The
    /// thread will be unblocked
    /// when notify_all() or notify_one() is executed, or when the relative
    /// timeout rel_time expires.
    /// It may also be unblocked spuriously. When unblocked, regardless of
    /// the reason, lock is reacquired
    /// and wait_not_full_condition() exits.
    /// @param lock An object of type std::unique_lock which must be locked
    /// by the current thread.
    /// @param rel_time An object of type std::chrono::duration representing
    /// the maximum time to spend waiting.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/duration
    template<class Rep, class Period> BlockingCollectionStatus
    wait_not_full_condition(std::unique_lock<LockType>& lock,
                            const std::chrono::duration<Rep, Period>& rel_time) {
        if (state_ == BlockingCollectionState::Deactivated)
            return BlockingCollectionStatus::NotActivated;

        if (is_adding_completed_i())
            return BlockingCollectionStatus::AddingCompleted;

        auto status = BlockingCollectionStatus::Ok;

        while (is_full_i()) {
            if (rel_time == std::chrono::duration<Rep, Period>::zero()) {
                status = BlockingCollectionStatus::TimedOut;
                break;
            }

            if (is_adding_completed_i()) {
                status = BlockingCollectionStatus::AddingCompleted;
                break;
            }

            if (rel_time.count() < 0) {
                not_full_condition_var_.wait(lock);
            } else {
                if (not_full_condition_var_.wait_for(lock, rel_time)) {
                    status = BlockingCollectionStatus::TimedOut;
                    break;
                }
            }

            // Add/TryAdd methods and CompleteAdding should not
            // be called concurrently - invalid operation

            if (is_adding_completed_i()) {
                status = BlockingCollectionStatus::CompleteAddingConcurrent;
                break;
            }

            if (state_ != BlockingCollectionState::Activated) {
                status = BlockingCollectionStatus::NotActivated;
                break;
            }
        }
        return status;
    }

    /// The method waits on the "not empty" condition variable whenever the
    /// collection becomes empty.
    /// It atomically releases lock, blocks the current executing thread,
    /// and adds it to the
    /// list of threads waiting on the "not empty" condition variable. The
    /// thread will be unblocked
    /// when notify_all() or notify_one() is executed, or when the relative
    /// timeout rel_time expires.
    /// It may also be unblocked spuriously. When unblocked, regardless of
    /// the reason, lock is reacquired
    /// and wait_not_empty_condition() exits.
    /// @param lock An object of type std::unique_lock which must be locked
    /// by the current thread.
    /// @param rel_time An object of type std::chrono::duration representing
    /// the maximum time to spend waiting.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/duration
    template<class Rep, class Period> BlockingCollectionStatus
    wait_not_empty_condition(std::unique_lock<LockType>& lock,
                             const std::chrono::duration<Rep, Period>& rel_time) {
        if (state_ == BlockingCollectionState::Deactivated)
            return BlockingCollectionStatus::NotActivated;

        if (is_completed_i())
            return BlockingCollectionStatus::Completed;

        auto status = BlockingCollectionStatus::Ok;

        while (is_empty_i()) {
            if (rel_time == std::chrono::duration<Rep, Period>::zero()) {
                status = BlockingCollectionStatus::TimedOut;
                break;
            }

            if (is_adding_completed_i()) {
                status = BlockingCollectionStatus::AddingCompleted;
                break;
            }

            if (rel_time.count() < 0) {
                not_empty_condition_var_.wait(lock);
            } else {
                if (not_empty_condition_var_.wait_for(lock, rel_time)) {
                    status = BlockingCollectionStatus::TimedOut;
                    break;
                }
            }

            if (state_ != BlockingCollectionState::Activated) {
                status = BlockingCollectionStatus::NotActivated;
                break;
            }
        }

        return status;
    }

    ContainerType& container() {
        return container_;
    }

    LockType& lock() {
        return lock_;
    }

private:
    BlockingCollectionState state_;

    size_t bounded_capacity_;
    bool is_adding_completed_;

    typename ConditionVariableGenerator::NotEmptyType not_empty_condition_var_;
    typename ConditionVariableGenerator::NotFullType not_full_condition_var_;

    // Synchronizes access to the BlockCollection.
    LockType lock_;
    // The underlying Container (e.g. Queue, Stack).
    ContainerType container_;
};

/// @class PriorityContainer
/// Represents a priority based collection. Items with the highest priority
/// will be at the head of the collection.
/// Implements the implicitly defined IProducerConsumerCollection<T> policy.
/// @tparam T The type of items in the collection.
/// @tparam ComparerType The type of comparer to use when comparing items.
template<typename T, typename ComparerType>
class PriorityContainer {
public:
    using container_type = std::deque<T>;
    using size_type = typename container_type::size_type;
    using value_type = typename container_type::value_type;
    using value_comparer = ComparerType;

    /// Initializes a new instance of the PriorityContainer<T> class.
    PriorityContainer()
            : bounded_capacity_(SIZE_MAX) {
    }

    /// Sets the max number of elements this container can hold.
    /// @param bounded_capacity The max number of elements this
    /// container can hold.
    void bounded_capacity(size_t bounded_capacity) {
        bounded_capacity_ = bounded_capacity;
    }

    /// Gets the max number of elements this container can hold.
    /// @returns The max number of elements this container can hold.
    size_t bounded_capacity() {
        return bounded_capacity_;
    }

    /// Gets the number of elements contained in the collection.
    /// @returns The number of elements contained in the collection.
    size_type size() {
        return container_.size();
    }

    /// Attempts to add an object to the collection according to the item's
    /// priority.
    /// @param new_item The object to add to the collection.
    /// @returns True if the object was added successfully; otherwise,
    /// false.
    bool try_add(const value_type& new_item) {
        if (container_.size() == bounded_capacity_)
            return false;
        return try_emplace(new_item);
    }

    /// Attempts to add an object to the collection according to the item's
    /// priority.
    /// @param new_item The object to add to the collection.
    /// @returns True if the object was added successfully; otherwise,
    /// false.
    bool try_add(value_type&& new_item) {
        if (container_.size() == bounded_capacity_)
            return false;
        return try_emplace(std::forward<T>(new_item));
    }

    /// Attempts to add an element to the collection according to the
    /// element's priority.
    /// This new element is constructed in place using args as the
    /// arguments for its construction.
    /// @param args Arguments forwarded to construct the new element.
    template <typename... Args> bool try_emplace(Args&&... args) {
        if (container_.size() == bounded_capacity_)
            return false;
        if (container_.empty()) {
            container_.emplace_front(std::forward<Args>(args)...);
        } else {
            T new_item(args...);

            // search from back to front (i.e. from the lowest priority to
            // the highest priority) for
            // item with a priority greater than or equal to new_item's
            // priority

            typename container_type::reverse_iterator itr = std::find_if(
                    container_.rbegin(), container_.rend(),
                    [&new_item, this](value_type &item) {
                        return this->comparer_(item, new_item) >= 0;
                    });

            if (itr == container_.rend()) {
                // if at end then new_item's priority is now the highest
                container_.emplace_front(std::move(new_item));
            } else if (itr == container_.rbegin()) {
                // if at start then new_item's priority is now the lowest
                container_.emplace_back(std::move(new_item));
            } else {
                // insert the new item behind the item of greater or
                // equal priority. This ensures that FIFO order is
                // maintained when items of the same priority are
                // inserted consecutively.
                container_.emplace(itr.base(), std::move(new_item));
            }
        }
        return true;
    }

    /// Attempts to remove and return the highest priority object from the
    /// collection.
    /// @param [out] item When this method returns, if the object was
    /// removed and returned successfully, item contains
    /// the removed object. If no object was available to be removed, the
    /// value is unspecified.
    /// @returns True if an object was removed and returned successfully;
    /// otherwise, false.
    bool try_take(value_type& item) {
        if (container_.empty())
            return false;
        item = container_.front();
        container_.pop_front();
        return true;
    }

    /// Attempts to remove and return the lowest priority object from the
    /// collection.
    /// @param [out] item When this method returns, if the object was
    /// removed and returned successfully, item contains
    /// the removed object. If no object was available to be removed, the
    /// value is unspecified.
    /// @returns True if an object was removed and returned successfully;
    /// otherwise, false.
    bool try_take_prio(value_type& item) {
        if (container_.empty())
            return false;

        bool init_current_priority = true;
        value_type* current_priority;

        typename container_type::reverse_iterator itr = std::find_if_not(
                container_.rbegin(), container_.rend(),
                [&current_priority, &init_current_priority, this](value_type &item) -> bool {
                    // Find the first version of the earliest item (i.e.,
                    // preserve FIFO order for items at the same priority).

                    if (init_current_priority) {
                        current_priority = &item;
                        init_current_priority = false;
                        return true;
                    }
                    bool continue_search = this->comparer_(item, *current_priority) <= 0;
                    if (continue_search) {
                        current_priority = &item;
                    }
                    return continue_search;
                });

        if (itr == container_.rend()) {
            item = container_.front();
            container_.pop_front();
        } else {
            typename container_type::iterator base = itr.base();
            item = (*base);
            container_.erase(base);
        }

        return true;
    }

private:
    size_t bounded_capacity_;
    container_type container_;
    value_comparer comparer_;
};

/// @class PriorityComparer
/// This is the default PriorityContainer comparer.
/// It expects that the objects being compared have overloaded
/// < and > operators.
/// @tparam T The type of objects to compare.
template<typename T> class PriorityComparer {
public:
    /// Initializes a new instance of the PriorityComparer<T> class.
    PriorityComparer() {
    }
    /// Compares two objects and returns a value indicating whether one is
    /// less than, equal to, or greater than the other.
    /// Implement this method to provide a customized sort order comparison
    ///  for type T.
    /// @param x The first object to compare.
    /// @param y The second object to compare.
    /// @return A signed integer that indicates the relative values of
    /// x and y, as shown in the following table.
    ///
    /// Value             | Meaning
    /// ------------------|---------------------
    /// Less than zero    | x is less than y.
    /// Zero              | x equals y.
    /// Greater than zero | x is greater than y.
    ///
    int operator() (const T& x, const T& y) const {
        if (x < y)
            return -1;
        else if (x > y)
            return 1;
        else
            return 0;
    }
};

template<typename T,
        typename ContainerType = PriorityContainer<T, PriorityComparer<T>>,
        typename ConditionVariableGenerator = StdConditionVariableGenerator>
class PriorityBlockingCollection : public BlockingCollection<T, ContainerType, ConditionVariableGenerator> {
public:
    using base = BlockingCollection<T, ContainerType, ConditionVariableGenerator>;

    /// Initializes a new instance of the PriorityBlockingCollection<T>
    /// class without an upper-bound.
    PriorityBlockingCollection()
            : base() {
    }

    /// Initializes a new instance of the PriorityBlockingCollection<T>
    /// class with the specified upper-bound.
    /// @param capacity The bounded size of the collection.
    explicit PriorityBlockingCollection(size_t capacity)
            : base(capacity) {
    }

    // "PriorityBlockingCollection" objects cannot be copied or assigned
    PriorityBlockingCollection(const PriorityBlockingCollection&) = delete;
    PriorityBlockingCollection& operator=(const PriorityBlockingCollection&) = delete;

    /// Removes the lowest priority item from the
    /// PriorityBlockingCollection<T>.
    /// A call to TakePrio may block until an item is available to be
    /// removed.
    /// @param[out] item The lowest priority item removed from the
    /// collection.
    void take_prio(T& item) {
        try_take_prio(item, std::chrono::milliseconds(-1));
    }

    /// Tries to remove the lowest priority item from the
    /// PriorityBlockingCollection<T>.
    /// If the collection is empty, this method immediately returns
    /// immediately without taking an item.
    /// @param[out] item The lowest priority item removed from the
    /// collection.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    BlockingCollectionStatus try_take_prio(T& item) {
        return try_take_prio(item, std::chrono::milliseconds::zero());
    }

    /// Tries to remove the lowest priority item from the
    /// PriorityBlockingCollection<T> in the specified time period.
    /// @param[out] item The lowest priority item removed from the
    /// collection.
    /// @param rel_time An object of type std::chrono::duration
    /// representing the maximum time to spend waiting.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/duration
    template<class Rep, class Period> BlockingCollectionStatus
    try_take_prio(T& item,
                  const std::chrono::duration<Rep, Period>& rel_time) {
        {
            std::unique_lock<typename ConditionVariableGenerator::lock_type>
                    guard(base::lock());

            auto status = base::wait_not_empty_condition(guard, rel_time);

            if (BlockingCollectionStatus::Ok != status)
                return status;

            if (!base::container().try_take_prio(item))
                return BlockingCollectionStatus::InternalError;

            base::signal(base::container().size(), true);
        }
        return BlockingCollectionStatus::Ok;
    }

    /// Takes up to count low priority elements from the
    /// PriorityBlockingCollection<T>.
    /// A call to take_prio_bulk may block until an element is available
    /// to be removed.
    /// Use std::make_move_iterator if the elements should be moved instead
    /// of copied.
    /// @param[out] first Contains first item taken.
    /// @param count The number of elements to take from collection.
    /// @param[out] taken The actual number of elements taken.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    template<typename Iterator> BlockingCollectionStatus
    take_prio_bulk(Iterator first, size_t count, size_t& taken) {
        return try_take_prio_bulk(first, count, taken,
                                  std::chrono::milliseconds(-1));
    }

    /// Takes up to count low priority elements from the
    /// PriorityBlockingCollection<T>.
    /// If the collection is empty, this method immediately returns without
    /// taking any items.
    /// Use std::make_move_iterator if the elements should be moved instead
    /// of copied.
    /// @param[out] first Contains first item taken.
    /// @param count The number of elements to take from collection.
    /// @param[out] taken The actual number of elements taken.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    template<typename Iterator> BlockingCollectionStatus
    try_take_prio_bulk(Iterator first, size_t count, size_t& taken) {
        return try_take_prio_bulk(first, count, taken,
                                  std::chrono::milliseconds::zero());
    }

    /// Tries to take up to count low priority elements from the
    /// PriorityBlockingCollection<T> within
    /// the specified time period.
    /// If the collection is empty, this method immediately returns without
    /// taking any items.
    /// Use std::make_move_iterator if the elements should be moved instead
    /// of copied.
    /// @param[out] first Contains first item taken.
    /// @param count The number of elements to take from collection.
    /// @param[out] taken The actual number of elements taken.
    /// @return A BlockCollectionStatus code.
    /// @see BlockingCollectionStatus
    /// @see http://en.cppreference.com/w/cpp/chrono/durations
    template <typename Iterator, class Rep, class Period>
    BlockingCollectionStatus try_take_prio_bulk(Iterator first,
                                                size_t count, size_t& taken,
                                                const std::chrono::duration<Rep, Period>& rel_time) {
        {
            taken = 0;

            if (count == 0)
                return BlockingCollectionStatus::Ok;

            std::unique_lock<typename ConditionVariableGenerator::lock_type>
                    guard(base::lock());

            auto status = base::wait_not_empty_condition(guard, rel_time);

            if (BlockingCollectionStatus::Ok != status)
                return status;

            auto end = first + count;

            for (; first != end; ++first)  {
                if (!base::container().try_take_prio((*first)))
                    break;

                if (++taken == count)
                    break;
            }
        }
        return BlockingCollectionStatus::Ok;
    }
};

namespace detail {
    struct ProducerType {};
    struct ConsumerType {};

    template< typename T >
    struct is_producer : std::false_type { };

    template<>
    struct is_producer<ProducerType> : std::true_type { };

    /// @class Guard
    /// Implements a strictly scope-based BlockingCollection wrapper.
    /// The class Guard is a BlockingCollection wrapper that provides a
    /// convenient RAII-style
    /// mechanism for attaching the current thread as a producer or
    /// consumer to the BlockingCollection for the
    /// duration of the scoped block.
    ///
    /// When a Guard object is created, it attaches the current thread as a
    /// producer or consumer of the
    /// BlockingCollection it is given. When control leaves the scope in
    /// which the Guard object
    /// was created, the Guard is destructed and the current thread is
    /// detached from the BlockingCollection.
    ///
    /// The Guard class makes it simple for threads to register as producer
    /// or consumers with the BlockingCollection<T>
    /// instance. Plus it ensures the thread will be detached from the
    /// BlockingCollection<T> in an
    /// exception scenario.
    ///
    /// The Guard class is non-copyable.
    /// @tparam BlockingCollectionType The type of BlockingCollection to
    /// Guard.
    /// @tparam GuardType The type of Guard to create (i.e. ProducerType
    /// or ConsumerType).
    /// @see ProducerGuard
    /// @see ConsumerGuard
    /// http://en.wikipedia.com/wiki/Resource_Acquisition_Is_Initialization
    template<typename BlockingCollectionType, typename GuardType>
    class Guard {
    public:
        explicit Guard(BlockingCollectionType &collection)
                : collection_(collection) {
            attach_i(is_producer<GuardType>());
        }

        Guard(Guard const&) = delete;
        Guard& operator=(Guard const&) = delete;

        ~Guard() {
            detach_i(is_producer<GuardType>());
        }

    private:
        void attach_i(std::false_type) {
            collection_.attach_consumer();
        }

        void attach_i(std::true_type) {
            collection_.attach_producer();
        }

        void detach_i(std::false_type) {
            collection_.detach_consumer();
        }

        void detach_i(std::true_type) {
            collection_.detach_producer();
        }

        BlockingCollectionType& collection_;
    };
} // namespace detail

/// A type alias for Guard<T, ProducerType>
template<typename T>
using ProducerGuard = detail::Guard<T, detail::ProducerType>;

/// A type alias for Guard<T, ConsumerType>
template<typename T>
using ConsumerGuard = detail::Guard<T, detail::ConsumerType>;

/// A type alias for BlockingCollection<T, Stack> - a last in-first out
/// (LIFO) BlockingCollection.
template<typename T>
using BlockingStack = BlockingCollection<T, StackContainer<T>>;

/// A type alias for BlockingCollection<T, Queue> - a first in-first out
/// (FIFO) BlockingCollection.
template<typename T>
using BlockingQueue = BlockingCollection<T, QueueContainer<T>>;

/// A type alias for BlockingCollection<T, PriorityQueue> - a priority-based
/// BlockingCollection.
template<typename T>
using BlockingPriorityQueue = BlockingCollection<T,
        PriorityContainer<T, PriorityComparer<T>>>;

#endif //FADAS_SAFEPQ_H