blob: a98b4a4f03605bfbacd0e8d04403ed6fb7afd42a (
plain) (
blame)
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
|
#include "subscription.h"
namespace NThreading {
bool operator==(TSubscriptionId const& l, TSubscriptionId const& r) noexcept {
return l.StateId() == r.StateId() && l.SubId() == r.SubId();
}
bool operator!=(TSubscriptionId const& l, TSubscriptionId const& r) noexcept {
return !(l == r);
}
void TSubscriptionManager::TSubscription::operator()() {
Callback();
}
TSubscriptionManagerPtr TSubscriptionManager::NewInstance() {
return new TSubscriptionManager();
}
TSubscriptionManagerPtr TSubscriptionManager::Default() {
static auto instance = NewInstance();
return instance;
}
void TSubscriptionManager::Unsubscribe(TSubscriptionId id) {
with_lock(Lock) {
UnsubscribeImpl(id);
}
}
void TSubscriptionManager::Unsubscribe(TVector<TSubscriptionId> const& ids) {
with_lock(Lock) {
UnsubscribeImpl(ids);
}
}
void TSubscriptionManager::OnCallback(TFutureStateId stateId) noexcept {
THashMap<ui64, TSubscription> subscriptions;
with_lock(Lock) {
auto const it = Subscriptions.find(stateId);
Y_VERIFY(it != Subscriptions.end(), "The callback has been triggered more than once");
subscriptions.swap(it->second);
Subscriptions.erase(it);
}
for (auto& [_, subscription] : subscriptions) {
subscription();
}
}
void TSubscriptionManager::UnsubscribeImpl(TSubscriptionId id) {
auto const it = Subscriptions.find(id.StateId());
if (it == std::end(Subscriptions)) {
return;
}
it->second.erase(id.SubId());
}
void TSubscriptionManager::UnsubscribeImpl(TVector<TSubscriptionId> const& ids) {
for (auto const& id : ids) {
UnsubscribeImpl(id);
}
}
}
|
