LockedQueue.h

Go to the documentation of this file.

 
// SPDX-License-Identifier: LGPL-2.1-or-later
 
#ifndef TASKOLIB_LOCKEDQUEUE_H_
#define TASKOLIB_LOCKEDQUEUE_H_
 
#include <condition_variable>
#include <mutex>
#include <optional>
 
#include <gul17/SlidingBuffer.h>
 
namespace task {
 
template <typename MessageT>
class LockedQueue
{
public:
    using MessageType = MessageT;
    using message_type = MessageT;
    using SizeType = std::uint32_t;
    using size_type = SizeType;
 
    LockedQueue(SizeType capacity)
        : queue_(capacity)
    { }
 
    SizeType capacity() const
    {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.capacity();
    }
 
    bool empty() const
    {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.empty();
    }
 
    MessageType pop()
    {
        std::unique_lock<std::mutex> lock(mutex_);
 
        if (queue_.empty())
            cv_message_available_.wait(lock, [this] { return not queue_.empty(); });
 
        auto msg = std::move(queue_.front());
        queue_.pop_front();
        lock.unlock();
        cv_slot_available_.notify_one();
        return msg;
    }
 
    MessageType back() const
    {
        std::unique_lock<std::mutex> lock(mutex_);
 
        if (queue_.empty())
            cv_message_available_.wait(lock, [this] { return not queue_.empty(); });
 
        auto msg = queue_.back();
        lock.unlock();
        return msg;
    }
 
    template <typename MsgT,
              std::enable_if_t<std::is_convertible_v<MsgT, MessageType>, bool> = true>
    void push(MsgT&& msg)
    {
        std::unique_lock<std::mutex> lock(mutex_);
 
        if (queue_.filled())
            cv_slot_available_.wait(lock, [this] { return not queue_.filled(); });
 
        queue_.push_back(std::forward<MsgT>(msg));
        lock.unlock();
        cv_message_available_.notify_one();
    }
 
    SizeType size() const
    {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.size();
    }
 
    std::optional<MessageType> try_pop()
    {
        std::unique_lock<std::mutex> lock(mutex_);
 
        if (queue_.empty())
            return std::nullopt;
 
        auto msg = std::move(queue_.front());
        queue_.pop_front();
        lock.unlock();
        cv_slot_available_.notify_one();
        return msg;
    }
 
    template <typename MsgT,
              std::enable_if_t<std::is_convertible_v<MsgT, MessageType>, bool> = true>
    bool try_push(MsgT&& msg)
    {
        std::unique_lock<std::mutex> lock(mutex_);
 
        if (queue_.filled())
            return false;
 
        queue_.push_back(std::forward<MsgT>(msg));
        lock.unlock();
        cv_message_available_.notify_one();
        return true;
    }
 
private:
    mutable std::mutex mutex_;
 
    mutable std::condition_variable cv_message_available_;
 
    mutable std::condition_variable cv_slot_available_;
 
    gul17::SlidingBuffer<MessageType> queue_;
};
 
} // namespace task
 
#endif