UVW源码漫谈（番外篇）—— Emitter

  1 #pragma once
  2 
  3 
  4 #include <type_traits>
  5 #include <functional>
  6 #include <algorithm>
  7 #include <utility>
  8 #include <cstddef>
  9 #include <vector>
 10 #include <memory>
 11 #include <list>
 12 #include <queue>
 13 #include <thread>
 14 #include <mutex>
 15 #include <condition_variable>
 16 #include <chrono>
 17 
 18 
 19 template<typename E>
 20 using event_ptr = std::unique_ptr<E>;
 21 
 22 
 23 template<typename E, typename... Args>
 24 event_ptr<E> make_event(Args&&... args) {
 25     return std::make_unique<E>(std::forward<Args>(args)...);
 26 }
 27 
 28 
 29 /**
 30  * @brief Event emitter base class.
 31  */
 32 template<typename T>
 33 class Emitter : public std::enable_shared_from_this<T> {
 34     struct BaseHandler {
 35         virtual ~BaseHandler() noexcept = default;
 36         virtual bool empty() const noexcept = 0;
 37         virtual void clear() noexcept = 0;
 38         virtual void join() noexcept = 0;
 39         virtual void exit() noexcept = 0;
 40     };
 41 
 42     template<typename E>
 43     struct Handler final: BaseHandler {
 44         using Listener = std::function<void(E &, std::shared_ptr<T>&)>;
 45         using Element = std::pair<bool, Listener>;
 46         using ListenerList = std::list<Element>;
 47         using Connection = typename ListenerList::iterator;
 48 
 49         bool empty() const noexcept override {
 50             auto pred = [](auto &&element){ return element.first; };
 51 
 52             return std::all_of(onceL.cbegin(), onceL.cend(), pred) &&
 53                     std::all_of(onL.cbegin(), onL.cend(), pred);
 54         }
 55 
 56         void clear() noexcept override {
 57             if(!publishing.try_lock()) {
 58                 auto func = [](auto &&element){ element.first = true; };
 59                 std::for_each(onceL.begin(), onceL.end(), func);
 60                 std::for_each(onL.begin(), onL.end(), func);
 61             } else {
 62                 onceL.clear();
 63                 onL.clear();
 64             }
 65             publishing.unlock();
 66         }
 67 
 68         Connection once(Listener f) {
 69             return onceL.emplace(onceL.cend(), false, std::move(f));
 70         }
 71 
 72         Connection on(Listener f) {
 73             return onL.emplace(onL.cend(), false, std::move(f));
 74         }
 75 
 76         void erase(Connection conn) noexcept {
 77             conn->first = true;
 78 
 79             if(publishing.try_lock()) {
 80                 auto pred = [](auto &&element){ return element.first; };
 81                 onceL.remove_if(pred);
 82                 onL.remove_if(pred);
 83             }
 84             publishing.unlock();
 85         }
 86 
 87         void run(E event, std::shared_ptr<T> ptr) {
 88             ListenerList currentL;
 89             onceL.swap(currentL);
 90 
 91             auto func = [&event, &ptr](auto &&element) {
 92                 return element.first ? void() : element.second(event, ptr);
 93             };
 94 
 95             publishing.lock();
 96 
 97             std::for_each(onL.rbegin(), onL.rend(), func);
 98             std::for_each(currentL.rbegin(), currentL.rend(), func);
 99 
100             publishing.unlock();
101 
102             onL.remove_if([](auto &&element){ return element.first; });
103         }
104 
105         void thread_fun(std::shared_ptr<T> ptr)
106         {
107             while(true) {
108                 std::unique_lock<std::mutex> lk(emutex);
109                 econd.wait_for(lk, std::chrono::milliseconds(60000), [this](){return !events.empty();});
110 
111                 if(events.size() > 0) {
112                     E event = std::move(events.front());
113                     events.pop();
114                     run(std::move(event), std::move(ptr));
115                 } else {
116                     break;
117                 }
118             }
119         }
120 
121         void publish(E event, std::shared_ptr<T> ptr, bool asyn) {
122             if(asyn) {
123                 {
124                     std::lock_guard<std::mutex> lk(emutex);
125                     events.push(std::move(event));
126                 }
127                 econd.notify_all();
128 
129                 if(!ethread.joinable()) {
130                     ethread = std::thread(&Handler<E>::thread_fun, this, std::move(ptr));
131                 }
132             } else {
133                 run(std::move(event), ptr);
134             }
135         }
136 
137         void join() noexcept override {
138             if(ethread.joinable()) {
139                 ethread.join();
140             }
141         }
142 
143         void exit() noexcept override {
144             if(ethread.joinable()) {
145                 econd.notify_all();
146                 ethread.join();
147             }
148         }
149 
150     private:
151         std::mutex publishing;
152         ListenerList onceL{};
153         ListenerList onL{};
154 
155         std::thread ethread;
156         std::queue<E> events;
157         std::mutex emutex;
158         std::condition_variable econd;
159     };
160 
161     static std::size_t next_type() noexcept {
162         static std::size_t counter = 0;
163         return counter++;
164     }
165 
166     template<typename>
167     static std::size_t event_type() noexcept {
168         static std::size_t value = next_type();
169         return value;
170     }
171 
172     template<typename E>
173     Handler<E> & handler() noexcept {
174         std::size_t type = event_type<E>();
175 
176         if(!(type < handlers.size())) {
177             handlers.resize(type+1);
178         }
179 
180         if(!handlers[type]) {
181            handlers[type] = std::make_unique<Handler<E>>();
182         }
183 
184         return static_cast<Handler<E>&>(*handlers[type]);
185     }
186 
187 protected:
188     template<typename E>
189     void publish(E event, bool asyn = false) {
190 //        handler<E>().publish(std::move(event), *static_cast<T*>(this), asyn);
191         handler<E>().publish(std::move(event), this->shared_from_this(), asyn);
192     }
193 
194 public:
195     template<typename E>
196     using Listener = typename Handler<E>::Listener;
197 
198     /**
199      * @brief Connection type for a given event type.
200      *
201      * Given an event type `E`, `Connection<E>` is the type of the connection
202      * object returned by the event emitter whenever a listener for the given
203      * type is registered.
204      */
205     template<typename E>
206     struct Connection: private Handler<E>::Connection {
207         template<typename> friend class Emitter;
208 
209         Connection() = default;
210         Connection(const Connection &) = default;
211         Connection(Connection &&) = default;
212 
213         Connection(typename Handler<E>::Connection conn)
214             : Handler<E>::Connection{std::move(conn)}
215         {}
216 
217         Connection & operator=(const Connection &) = default;
218         Connection & operator=(Connection &&) = default;
219     };
220 
221     virtual ~Emitter() noexcept {
222         static_assert(std::is_base_of<Emitter<T>, T>::value, "!");
223     }
224 
225     /**
226      * @brief Registers a long-lived listener with the event emitter.
227      *
228      * This method can be used to register a listener that is meant to be
229      * invoked more than once for the given event type.<br/>
230      * The Connection object returned by the method can be freely discarded. It
231      * can be used later to disconnect the listener, if needed.
232      *
233      * Listener is usually defined as a callable object assignable to a
234      * `std::function<void(const E &, T &)`, where `E` is the type of the event
235      * and `T` is the type of the resource.
236      *
237      * @param f A valid listener to be registered.
238      * @return Connection object to be used later to disconnect the listener.
239      */
240     template<typename E>
241     Connection<E> on(Listener<E> f) {
242         return handler<E>().on(std::move(f));
243     }
244 
245     /**
246      * @brief Registers a short-lived listener with the event emitter.
247      *
248      * This method can be used to register a listener that is meant to be
249      * invoked only once for the given event type.<br/>
250      * The Connection object returned by the method can be freely discarded. It
251      * can be used later to disconnect the listener, if needed.
252      *
253      * Listener is usually defined as a callable object assignable to a
254      * `std::function<void(const E &, T &)`, where `E` is the type of the event
255      * and `T` is the type of the resource.
256      *
257      * @param f Avalid listener to be registered.
258      * @return Connection object to be used later to disconnect the listener.
259      */
260     template<typename E>
261     Connection<E> once(Listener<E> f) {
262         return handler<E>().once(std::move(f));
263     }
264 
265     /**
266      * @brief Disconnects a listener from the event emitter.
267      * @param conn A valid Connection object
268      */
269     template<typename E>
270     void erase(Connection<E> conn) noexcept {
271         handler<E>().erase(std::move(conn));
272     }
273 
274     /**
275      * @brief Disconnects all the listeners for the given event type.
276      */
277     template<typename E>
278     void clear() noexcept {
279         handler<E>().clear();
280     }
281 
282     /**
283      * @brief Disconnects all the listeners.
284      */
285     void clear() noexcept {
286         std::for_each(handlers.begin(), handlers.end(),
287                       [](auto &&hdlr){ if(hdlr) { hdlr->clear(); } });
288     }
289 
290     /**
291      * @brief Checks if there are listeners registered for the specific event.
292      * @return True if there are no listeners registered for the specific event,
293      * false otherwise.
294      */
295     template<typename E>
296     bool empty() const noexcept {
297         std::size_t type = event_type<E>();
298 
299         return (!(type < handlers.size()) ||
300                 !handlers[type] ||
301                 static_cast<Handler<E>&>(*handlers[type]).empty());
302     }
303 
304     /**
305      * @brief Checks if there are listeners registered with the event emitter.
306      * @return True if there are no listeners registered with the event emitter,
307      * false otherwise.
308      */
309     bool empty() const noexcept {
310         return std::all_of(handlers.cbegin(), handlers.cend(),
311                            [](auto &&hdlr){ return !hdlr || hdlr->empty(); });
312     }
313 
314     void thread_join() const noexcept {
315         std::for_each(handlers.begin(), handlers.end(),
316                       [](auto &&hdlr){ if(hdlr) { hdlr->join(); } });
317     }
318 
319     void thread_exit() const noexcept {
320         std::for_each(handlers.begin(), handlers.end(),
321                       [](auto &&hdlr){ if(hdlr) { hdlr->exit(); } });
322     }
323 
324 private:
325     std::vector<std::unique_ptr<BaseHandler>> handlers{};
326 };

 1 #include <iostream>
 2 #include <memory>
 3 #include <thread>
 4 
 5 using namespace std;
 6 #include "emitter.h"
 7 
 8 
 9 struct StringEvent
10 {
11     StringEvent(std::string str):i_str(str)
12     {
13         cout << "StringEvent" << std::endl;
14     }
15 
16     void print()
17     {
18         std::cout << "string event:" << i_str << std::endl;
19     }
20 
21     std::string i_str;
22 
23     ~StringEvent()
24     {
25         cout << "~StringEvent" << std::endl;
26     }
27 };
28 
29 struct IntEvent
30 {
31     IntEvent(int t) : i_t(t)
32     {
33         cout << "IntEvent" << std::endl;
34     }
35     void print()
36     {
37         std::cout << "int event:" << i_t << std::endl;
38     }
39     ~IntEvent()
40     {
41         cout << "~IntEvent" << std::endl;
42     }
43 
44     int i_t{1};
45 };
46 
47 
48 class A : public Emitter<A>
49 {
50 public:
51     A()
52     {
53         cout << "A" << endl;
54     }
55 
56     void print()
57     {
58         publish(StringEvent("Hello"), false);
59         publish(make_event<StringEvent>("Hello"), true);
60         publish(make_unique<StringEvent>("World"), true);
61 
62         this_thread::sleep_for(1000ms);
63         publish(make_unique<IntEvent>(2), true);
64         publish(make_unique<IntEvent>(3), true);
65 
66     }
67 
68     ~A()
69     {
70         cout << "~A" << endl;
71     }
72 };
73 
74 
75 int main()
76 {
77     shared_ptr<A> em = make_shared<A>();
78 
79     em->on<StringEvent>([](StringEvent& ev, shared_ptr<A>& a){
80         ev.print();
81     });
82 
83     em->on<event_ptr<StringEvent>>([](event_ptr<StringEvent>& ev, shared_ptr<A>& a){
84         ev->print();
85     });
86 
87     em->on<event_ptr<IntEvent>>([](event_ptr<IntEvent>& ev, shared_ptr<A>& a){
88         ev->print();
89     });
90 
91     em->print();
92 
93     em->thread_join();
94 
95     return 0;
96 }