Skip to content

Commit da1738b

Browse files
forgotthepenforgotthepen
committed
* custom class for a callable consumer
* lock on the mutex before sending a thread kill signal * allow setting the item when poping
1 parent 76b246c commit da1738b

3 files changed

Lines changed: 178 additions & 23 deletions

File tree

‎CMakeLists.txt‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,19 +6,15 @@ project(${project_name})
66

77
file(GLOB_RECURSE src_files
88
"test/*.cpp"
9-
)
10-
11-
file(GLOB_RECURSE inc_files
12-
"include/*.hpp"
9+
"observable/*.hpp"
1310
)
1411

1512
add_executable(${project_name}
1613
${src_files}
17-
${inc_files}
1814
)
1915

2016
target_include_directories(${project_name} PUBLIC
21-
"include/"
17+
"./"
2218
)
2319

2420
set_target_properties(${project_name} PROPERTIES

include/observable/queue.hpp renamed to observable/queue.hpp

Lines changed: 96 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,10 @@ SOFTWARE.
2424

2525
#pragma once
2626

27+
#include <type_traits>
28+
#include <typeinfo>
2729
#include <functional>
30+
#include <memory>
2831
#include <thread>
2932
#include <condition_variable>
3033
#include <mutex>
@@ -36,8 +39,70 @@ SOFTWARE.
3639
namespace obs {
3740
template<typename Ty, bool IsThreaded = true>
3841
class queue {
42+
private:
43+
template<typename Tany>
44+
struct fn_type {
45+
template<typename Tignore>
46+
inline constexpr static const std::type_info& type(Tignore &&) {
47+
return typeid(Tany);
48+
}
49+
};
50+
51+
template<typename Tany>
52+
struct fn_type<std::function<Tany>> {
53+
inline static const std::type_info& type(const std::function<Tany> &fn) noexcept {
54+
return fn.target_type();
55+
}
56+
};
57+
3958
public:
40-
using t_consumer = std::function<void(Ty &)>;
59+
class t_consumer {
60+
private:
61+
struct ICallableConsumer {
62+
virtual ~ICallableConsumer() { } // in case the child is a type with a dtor
63+
virtual void operator ()(Ty& item) = 0;
64+
virtual const std::type_info& type() const noexcept = 0;
65+
};
66+
67+
template<typename Tfn>
68+
class CallableConsumerImpl : public ICallableConsumer {
69+
private:
70+
using TfnObj = typename std::remove_cv<typename std::remove_reference<Tfn>::type>::type;
71+
72+
TfnObj fn_;
73+
74+
public:
75+
CallableConsumerImpl(TfnObj fn):
76+
fn_( std::move(fn) )
77+
{ }
78+
79+
void operator ()(Ty& item) override {
80+
fn_( item );
81+
}
82+
83+
const std::type_info& type() const noexcept override {
84+
return fn_type<Tfn>::type(fn_);
85+
}
86+
};
87+
88+
std::unique_ptr<ICallableConsumer> callable_consumer_;
89+
90+
public:
91+
template<typename Tfn>
92+
t_consumer(Tfn &&fn):
93+
callable_consumer_(new CallableConsumerImpl<Tfn>(
94+
std::forward<Tfn>(fn)
95+
))
96+
{ }
97+
98+
inline void operator ()(Ty& item) const {
99+
(*callable_consumer_)(item);
100+
}
101+
102+
inline const std::type_info& type() const {
103+
return callable_consumer_->type();
104+
}
105+
};
41106

42107
private:
43108
std::thread thread_;
@@ -51,14 +116,21 @@ namespace obs {
51116
std::list<t_consumer> consumers_{};
52117

53118
void kill() {
54-
if (kill_) {
55-
return;
56-
}
119+
{
120+
std::lock_guard<std::mutex> lock(mtx_);
57121

58-
kill_ = true;
122+
if (kill_) {
123+
return;
124+
}
125+
126+
kill_ = true;
127+
128+
if (IsThreaded) {
129+
cv_.notify_one();
130+
}
131+
}
59132

60133
if (IsThreaded) {
61-
cv_.notify_one();
62134
thread_.join();
63135
}
64136
}
@@ -139,6 +211,7 @@ namespace obs {
139211
template<typename ...Args>
140212
Ty& emplace_back(Args ...args) {
141213
Ty *item;
214+
142215
{
143216
std::lock_guard<std::mutex> lock(mtx_);
144217

@@ -152,34 +225,42 @@ namespace obs {
152225
return *item;
153226
}
154227

155-
bool try_pop_front() {
228+
bool try_pop_front(Ty *obj = nullptr) {
156229
std::lock_guard<std::mutex> lock(mtx_);
157230

158231
if (queue_.empty()) {
159232
return false;
160233
}
161234

235+
if (obj) {
236+
*obj = std::move( queue_.front() );
237+
}
238+
162239
queue_.pop_front();
163240
return true;
164241
}
165242

166-
queue<Ty>& operator +=(const t_consumer &consumer) {
243+
template<typename Tfn>
244+
queue<Ty>& operator +=(Tfn &&consumer) {
167245
std::lock_guard<std::mutex> lock(mtx_consumers_);
168246

169-
bool exists = std::any_of(consumers_.begin(), consumers_.end(), [&consumer](const t_consumer &item){
170-
return item.target_type() == consumer.target_type();
247+
const auto &consumer_type = fn_type<Tfn>::type(std::forward<Tfn>(consumer));
248+
bool exists = std::any_of(consumers_.begin(), consumers_.end(), [&consumer_type](const t_consumer &item){
249+
return item.type() == consumer_type;
171250
});
172251
if (!exists) {
173-
consumers_.emplace_back(consumer);
252+
consumers_.emplace_back(std::forward<Tfn>(consumer));
174253
}
175254
return *this;
176255
}
177256

178-
queue<Ty>& operator -=(const t_consumer &consumer) {
257+
template<typename Tfn>
258+
queue<Ty>& operator -=(Tfn &&consumer) {
179259
std::lock_guard<std::mutex> lock(mtx_consumers_);
180260

181-
consumers_.remove_if([&consumer](const t_consumer &item){
182-
return item.target_type() == consumer.target_type();
261+
const auto &consumer_type = fn_type<Tfn>::type(std::forward<Tfn>(consumer));
262+
consumers_.remove_if([&consumer_type](const t_consumer &item){
263+
return item.type() == consumer_type;
183264
});
184265
return *this;
185266
}
@@ -196,7 +277,7 @@ namespace obs {
196277
return consumers_.size();
197278
}
198279

199-
void poll() {
280+
inline void poll() {
200281
if (!IsThreaded) {
201282
worker_fn();
202283
}

‎test/main.cpp‎

Lines changed: 80 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
#include <chrono>
44
#include <iostream>
55
#include <string>
6+
#include <functional>
67

78

89
struct MyStr {
@@ -26,17 +27,46 @@ struct MyStr {
2627
std::cout << "** ~MyStr()" << '\n';
2728
}
2829
MyStr& operator =(const MyStr &other) {
29-
std::cout << "** MyStr +(copy)" << '\n';
30+
std::cout << "** MyStr =(copy)" << '\n';
3031
str = other.str;
3132
return *this;
3233
}
3334
MyStr& operator =(MyStr &&other) {
34-
std::cout << "** MyStr +(move)" << '\n';
35+
std::cout << "** MyStr =(move)" << '\n';
3536
str = std::move(other.str);
3637
return *this;
3738
}
3839
};
3940

41+
struct MyCallable {
42+
MyCallable() {
43+
std::cout << "** MyCallable()" << '\n';
44+
}
45+
MyCallable(const MyCallable &other) {
46+
std::cout << "** MyCallable(copy)" << '\n';
47+
}
48+
MyCallable(MyCallable &&other) {
49+
std::cout << "** MyCallable(move)" << '\n';
50+
}
51+
MyCallable(const std::string &str_other) {
52+
std::cout << "** MyCallable(str)" << '\n';
53+
}
54+
~MyCallable() {
55+
std::cout << "** ~MyCallable()" << '\n';
56+
}
57+
MyCallable& operator =(const MyCallable &other) {
58+
std::cout << "** MyCallable +(copy)" << '\n';
59+
return *this;
60+
}
61+
MyCallable& operator =(MyCallable &&other) {
62+
std::cout << "** MyCallable +(move)" << '\n';
63+
return *this;
64+
}
65+
66+
void operator () (MyStr &str) const {
67+
std::cout << " (( MyStr ()(call)" << '\n';
68+
}
69+
};
4070

4171
int main(int argc, char* *argv) {
4272
obs::queue<MyStr> queue{};
@@ -54,12 +84,60 @@ int main(int argc, char* *argv) {
5484
}
5585
};
5686

87+
auto bound_lval = std::bind([](MyStr &str){
88+
std::cout << " >> bound l-val: " << str.str << '\n';
89+
}, std::placeholders::_1);
90+
91+
queue += bound_lval;
92+
queue += bound_lval;
93+
94+
{
95+
queue += std::bind([](MyStr &str){
96+
std::cout << " >> bound r-val: " << str.str << '\n';
97+
}, std::placeholders::_1);
98+
}
99+
100+
auto func_lval = std::function<void(MyStr &)>([](MyStr &str){
101+
std::cout << " >> func l-val: " << str.str << '\n';
102+
});
103+
104+
queue += func_lval;
105+
queue += func_lval;
106+
107+
{
108+
queue += std::function<void(MyStr &)>([](MyStr &str){
109+
std::cout << " >> func r-val: " << str.str << '\n';
110+
});
111+
}
112+
113+
{
114+
queue += MyCallable{};
115+
}
116+
117+
auto mycallable_lval = MyCallable{};
118+
queue += mycallable_lval;
119+
queue += mycallable_lval;
120+
57121
while (true) {
58122
std::cout << "Type anything and press Enter: ";
59123
std::string ss{};
60124
std::getline(std::cin, ss);
61125
if ("exit" == ss || "quit" == ss || "x" == ss || "q" == ss) {
62126
break;
127+
} else if ("remove bound l-val" == ss) {
128+
queue -= bound_lval;
129+
queue -= bound_lval;
130+
} else if ("add bound l-val" == ss) {
131+
queue += bound_lval;
132+
queue += bound_lval;
133+
} else if ("remove func l-val" == ss) {
134+
queue -= func_lval;
135+
queue -= func_lval;
136+
} else if ("add func l-val" == ss) {
137+
queue += func_lval;
138+
queue += func_lval;
139+
} else if ("try pop" == ss) {
140+
queue.try_pop_front();
63141
}
64142

65143
queue.emplace_back(ss);

0 commit comments

Comments
 (0)