46 lines
1.6 KiB
C++
46 lines
1.6 KiB
C++
#pragma once
|
|
#include <condition_variable>
|
|
#include <deque>
|
|
#include <mutex>
|
|
#include <optional>
|
|
#include <stdexcept>
|
|
|
|
namespace cms {
|
|
// FIFO with bounded ownership. Normal close drains; cancellation may discard.
|
|
template<class T> class BoundedQueue {
|
|
public:
|
|
explicit BoundedQueue(std::size_t capacity):capacity_(capacity) {
|
|
if(!capacity)throw std::invalid_argument("Queue capacity must be positive");
|
|
}
|
|
bool push(T item) {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
changed_.wait(lock,[&]{return closed_||items_.size()<capacity_;});
|
|
if(closed_)return false;
|
|
items_.push_back(std::move(item));changed_.notify_all();return true;
|
|
}
|
|
bool pushLatest(T item) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
if(closed_)return false;
|
|
if(items_.size()==capacity_)items_.pop_front();
|
|
items_.push_back(std::move(item));changed_.notify_all();return true;
|
|
}
|
|
void clear() {
|
|
std::lock_guard<std::mutex> lock(mutex_);items_.clear();changed_.notify_all();
|
|
}
|
|
std::optional<T> pop() {
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
changed_.wait(lock,[&]{return closed_||!items_.empty();});
|
|
if(items_.empty())return {};
|
|
T item=std::move(items_.front());items_.pop_front();changed_.notify_all();return item;
|
|
}
|
|
void close(bool discard=false) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
closed_=true;if(discard)items_.clear();changed_.notify_all();
|
|
}
|
|
private:
|
|
const std::size_t capacity_;
|
|
std::mutex mutex_;std::condition_variable changed_;
|
|
std::deque<T> items_;bool closed_{};
|
|
};
|
|
}
|