Repository navigation
Expand file tree
/
Copy pathmessagequeue.h
More file actions
67 lines (57 loc) · 1.48 KB
/
Copy pathmessagequeue.h
File metadata and controls
67 lines (57 loc) · 1.48 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
#ifndef MESSAGEQUEUE_H
#define MESSAGEQUEUE_H
#include <queue>
#include <mutex>
#include <condition_variable>
#include <optional>
#include "message.h"
class MessageQueue {
public:
MessageQueue() = default;
~MessageQueue() = default;
//Push message in the queue thread-safe
void push(const Message &m) {
{
std::lock_guard<std::mutex> lk(mutex_);
q_.push(m);
}
cv_.notify_one();
}
std::optional<Message> tryPop() {
std::lock_guard<std::mutex> lk(mutex_);
if (q_.empty()) return std::nullopt;
Message m = q_.front();
q_.pop();
return m;
}
Message waitPop() {
std::unique_lock<std::mutex> lk(mutex_);
cv_.wait(lk, [this]{ return !q_.empty() || stop_; });
if (q_.empty()) return Message();
Message m = q_.front();
q_.pop();
return m;
}
void clear() {
std::lock_guard<std::mutex> lk(mutex_);
std::queue<Message> empty;
std::swap(q_, empty);
}
size_t size() {
std::lock_guard<std::mutex> lk(mutex_);
return q_.size();
}
void requestStop() {
{
std::lock_guard<std::mutex> lk(mutex_);
stop_ = true;
}
cv_.notify_all();
}
private:
std::queue<Message> q_;
std::mutex mutex_;
std::condition_variable cv_;
bool stop_ = false;
};
#endif // MESSAGEQUEUE_H