blob: 6fcb651dc4c9951c533c00a53a205b689f5bc6be (
plain)
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
|
/*
* Copyright (C) 2016-2018 Team Kodi
* This file is part of Kodi - https://kodi.tv
*
* SPDX-License-Identifier: GPL-2.0-or-later
* See LICENSES/README.md for more information.
*/
#pragma once
#include "EventStreamDetail.h"
#include "JobManager.h"
#include "threads/CriticalSection.h"
#include <algorithm>
#include <memory>
#include <mutex>
#include <vector>
template<typename Event>
class CEventStream
{
public:
template<typename A>
void Subscribe(A* owner, void (A::*fn)(const Event&))
{
auto subscription = std::make_shared<detail::CSubscription<Event, A>>(owner, fn);
std::unique_lock<CCriticalSection> lock(m_criticalSection);
m_subscriptions.emplace_back(std::move(subscription));
}
template<typename A>
void Unsubscribe(A* obj)
{
std::vector<std::shared_ptr<detail::ISubscription<Event>>> toCancel;
{
std::unique_lock<CCriticalSection> lock(m_criticalSection);
auto it = m_subscriptions.begin();
while (it != m_subscriptions.end())
{
if ((*it)->IsOwnedBy(obj))
{
toCancel.push_back(*it);
it = m_subscriptions.erase(it);
}
else
{
++it;
}
}
}
for (auto& subscription : toCancel)
subscription->Cancel();
}
protected:
std::vector<std::shared_ptr<detail::ISubscription<Event>>> m_subscriptions;
CCriticalSection m_criticalSection;
};
template<typename Event>
class CEventSource : public CEventStream<Event>
{
public:
explicit CEventSource() : m_queue(false, 1, CJob::PRIORITY_HIGH) {}
template<typename A>
void Publish(A event)
{
std::unique_lock<CCriticalSection> lock(this->m_criticalSection);
auto& subscriptions = this->m_subscriptions;
auto task = [subscriptions, event](){
for (auto& s: subscriptions)
s->HandleEvent(event);
};
lock.unlock();
m_queue.Submit(std::move(task));
}
private:
CJobQueue m_queue;
};
template<typename Event>
class CBlockingEventSource : public CEventStream<Event>
{
public:
template<typename A>
void HandleEvent(A event)
{
std::unique_lock<CCriticalSection> lock(this->m_criticalSection);
for (const auto& subscription : this->m_subscriptions)
{
subscription->HandleEvent(event);
}
}
};
|
