safe_queue.hpp

// SPDX-License-Identifier: Apache-2.0
// Copyright 2020 - 2024 Pionix GmbH and Contributors to EVerest
 
#pragma once
 
#include <condition_variable>
#include <mutex>
#include <queue>
 
namespace ocpp {
 
template <typename T> class SafeQueue {
public:
    inline bool empty() const {
        std::lock_guard lock(mutex);
        return queue.empty();
    }
 
    inline T front() {
        std::lock_guard lock(mutex);
        return queue.front();
    }
 
    inline T pop() {
        std::unique_lock<std::mutex> lock(mutex);
 
        T front = std::move(queue.front());
        queue.pop();
 
        // Unlock here and notify
        lock.unlock();
 
        notify_waiting_thread();
 
        return front;
    }
 
    inline void push(T&& value) {
        {
            std::lock_guard<std::mutex> lock(mutex);
            queue.push(value);
        }
 
        notify_waiting_thread();
    }
 
    inline void push(const T& value) {
        {
            std::lock_guard<std::mutex> lock(mutex);
            queue.push(value);
        }
 
        notify_waiting_thread();
    }
 
    inline void clear() {
        {
            std::lock_guard<std::mutex> lock(mutex);
 
            std::queue<T> empty;
            empty.swap(queue);
        }
 
        notify_waiting_thread();
    }
 
    inline void wait_on_queue_element(std::chrono::milliseconds timeout = std::chrono::milliseconds(0)) {
        wait_on_queue_element_or_predicate([]() { return false; }, timeout);
    }
 
    template <class Predicate>
    inline void wait_on_queue_element_or_predicate(Predicate pred,
                                                   std::chrono::milliseconds timeout = std::chrono::milliseconds(0)) {
        std::unique_lock<std::mutex> lock(mutex);
 
        if (timeout.count() > 0) {
            cv.wait_for(lock, timeout, [&]() { return (false == queue.empty()) or pred(); });
        } else {
            cv.wait(lock, [&]() { return (false == queue.empty()) or pred(); });
        }
    }
 
    template <class Predicate>
    inline void wait_on_custom_event(Predicate pred, std::chrono::milliseconds timeout = std::chrono::milliseconds(0)) {
        std::unique_lock<std::mutex> lock(mutex);
 
        if (timeout.count() > 0) {
            cv.wait_for(lock, timeout, [&]() { return pred(); });
        } else {
            cv.wait(lock, [&]() { return pred(); });
        }
    }
 
    inline void notify_waiting_thread() {
        cv.notify_one();
    }
 
private:
    std::queue<T> queue;
 
    mutable std::mutex mutex;
    std::condition_variable cv;
};
 
} // namespace ocpp