add get msg timeout

zhangmeng

2022-12-27 cf0a3209b51babf72469d962914db0dac2e5f52c

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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
#ifndef _c_fixed_q_h_
#define _c_fixed_q_h_
 
#include <deque>
#include <condition_variable>
#include <functional>
 
#define notify notify_one
// #define notify notify_all
 
constexpr int du = 6;
 
template <class T>
class fixed_q{
public:
    fixed_q(const size_t qsize, std::function<bool()> fn)
    :qsize(qsize),pred(move(fn)){}
    fixed_q(const fixed_q&)=delete;
    fixed_q& operator=(const fixed_q&)=delete;
 
    const size_t capacity()const{ return qsize; }
    const size_t size(){
        std::lock_guard<std::mutex> l{mtx_q};
        return q.size();
    }
    int push(const T& m){
        std::unique_lock<std::mutex> l{mtx_q};
        auto d = std::chrono::milliseconds{du};
        while(!cond_spare.wait_for(l, d, [this]{ return q.size() < qsize; })){
            if (pred()) return -1;
        }
 
        q.push_back(m);
        cond_node.notify();
        return q.size();
    }
    int emplace(T&& m){
        std::unique_lock<std::mutex> l{mtx_q};
        auto d = std::chrono::milliseconds{du};
        while(!cond_spare.wait_for(l, d, [this]{ return q.size() < qsize; })){
            if (pred()) {
                return -1;
            }
        }
        q.emplace_back(move(m));
        cond_node.notify();
        return q.size();
    }
    int pop(T& m, const size_t ms){
        std::unique_lock<std::mutex> l{mtx_q};
        auto d = std::chrono::milliseconds{ms};
        if(!cond_node.wait_for(l, d, [this]{ return !q.empty(); })){
            return -1;
        }
        m = q.front();
        q.pop_front();
        cond_spare.notify();
        return q.size();
    }
    int pop(T& m){
        std::unique_lock<std::mutex> l{mtx_q};
        auto d = std::chrono::milliseconds{du};
        while(!cond_node.wait_for(l, d, [this]{ return !q.empty(); })){
            if (pred()) return -1;
        }
        m = q.front();
        q.pop_front();
        cond_spare.notify();
        return q.size();
    }
    T pop(const size_t ms){
        std::unique_lock<std::mutex> l{mtx_q};
        auto d = std::chrono::milliseconds{ms};
        T t{};
        if(!cond_node.wait_for(l, d, [this]{ return !q.empty(); })){
            return t;
        }
        t = q.front();
        q.pop_front();
        cond_spare.notify();
        return t;
    }
    T pop(){
        std::unique_lock<std::mutex> l{mtx_q};
        auto d = std::chrono::milliseconds{du};
        T t{};
        while(!cond_node.wait_for(l, d, [this]{ return !q.empty(); })){
            if (pred()) return t;
        }
        t = q.front();
        q.pop_front();
        cond_spare.notify();
        return t;
    }
 
    template<typename F> void clear(F&& f){
        std::lock_guard<std::mutex> l{mtx_q};
        while(!q.empty()){
            auto& msg = q.front();
            f(msg);
            q.pop_front();
        }
    }
 
private:
    const size_t qsize;
    std::function<bool()> pred;
    std::deque<T> q;
    std::mutex mtx_q;
    std::condition_variable cond_node;
    std::condition_variable cond_spare;
};
 
#endif