library/cpp/threading/future/subscription/subscription.cpp


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); 
    } 
} 
 
}