// This file is part of Desktop App Toolkit, // a set of libraries for developing nice desktop applications. // // For license and copyright information please follow this link: // https://github.com/desktop-app/legal/blob/master/LEGAL // #pragma once #include #include #include #include "base/type_traits.h" namespace base { namespace internal { using ObservableCallHandlers = Fn; void RegisterPendingObservable(ObservableCallHandlers *handlers); void UnregisterActiveObservable(ObservableCallHandlers *handlers); void UnregisterObservable(ObservableCallHandlers *handlers); template struct SubscriptionHandlerHelper { using type = Fn)>; }; template <> struct SubscriptionHandlerHelper { using type = Fn; }; template using SubscriptionHandler = typename SubscriptionHandlerHelper::type; class BaseObservableData { }; template class CommonObservableData; template class ObservableData; } // namespace internal class Subscription { public: Subscription() = default; Subscription(const Subscription &) = delete; Subscription &operator=(const Subscription &) = delete; Subscription(Subscription &&other) : _node(base::take(other._node)), _removeAndDestroyMethod(other._removeAndDestroyMethod) { } Subscription &operator=(Subscription &&other) { qSwap(_node, other._node); qSwap(_removeAndDestroyMethod, other._removeAndDestroyMethod); return *this; } explicit operator bool() const { return (_node != nullptr); } void destroy() { if (_node) { (*_removeAndDestroyMethod)(base::take(_node)); } } ~Subscription() { destroy(); } private: struct Node { Node(const std::shared_ptr &observable) : observable(observable) { } Node *next = nullptr; Node *prev = nullptr; std::weak_ptr observable; }; using RemoveAndDestroyMethod = void(*)(Node*); Subscription(Node *node, RemoveAndDestroyMethod removeAndDestroyMethod) : _node(node) , _removeAndDestroyMethod(removeAndDestroyMethod) { } Node *_node = nullptr; RemoveAndDestroyMethod _removeAndDestroyMethod; template friend class internal::CommonObservableData; template friend class internal::ObservableData; }; namespace internal { template class BaseObservable; template class CommonObservable { public: Subscription add_subscription(Handler &&handler) { if (!_data) { _data = std::make_shared>(this); } return _data->append(std::move(handler)); } private: std::shared_ptr> _data; friend class CommonObservableData; friend class BaseObservable::is_fast_copy_type::value>; }; template class BaseObservable : public internal::CommonObservable { public: void notify(EventType event, bool sync = false) { if (this->_data) { this->_data->notify(std::move(event), sync); } } }; template class BaseObservable : public internal::CommonObservable { public: void notify(EventType &&event, bool sync = false) { if (this->_data) { this->_data->notify(std::move(event), sync); } } void notify(const EventType &event, bool sync = false) { if (this->_data) { auto event_copy = event; this->_data->notify(std::move(event_copy), sync); } } }; } // namespace internal namespace internal { template class CommonObservableData : public BaseObservableData { public: CommonObservableData(CommonObservable *observable) : _observable(observable) { } Subscription append(Handler &&handler) { auto node = new Node(_observable->_data, std::move(handler)); if (_begin) { _end->next = node; node->prev = _end; _end = node; } else { _begin = _end = node; } return { _end, &CommonObservableData::removeAndDestroyNode }; } bool empty() const { return !_begin; } private: struct Node : public Subscription::Node { Node( const std::shared_ptr &observer, Handler &&handler) : Subscription::Node(observer) , handler(std::move(handler)) { } Handler handler; }; void remove(Subscription::Node *node) { if (node->prev) { node->prev->next = node->next; } if (node->next) { node->next->prev = node->prev; } if (_begin == node) { _begin = static_cast(node->next); } if (_end == node) { _end = static_cast(node->prev); } if (_current == node) { _current = static_cast(node->prev); } else if (!_begin) { _observable->_data.reset(); } } static void removeAndDestroyNode(Subscription::Node *node) { if (const auto that = node->observable.lock()) { static_cast(that.get())->remove(node); } delete static_cast(node); } template void notifyEnumerate(CallCurrent callCurrent) { _current = _begin; do { callCurrent(); if (_current) { _current = static_cast(_current->next); } else if (_begin) { _current = _begin; } else { break; } } while (_current); } bool destroyMeIfEmpty() const { if (empty()) { _observable->_data.reset(); return true; } return false; } CommonObservable *_observable = nullptr; Node *_begin = nullptr; Node *_current = nullptr; Node *_end = nullptr; ObservableCallHandlers _callHandlers; friend class ObservableData; }; template class ObservableData : public CommonObservableData { public: using CommonObservableData::CommonObservableData; void notify(EventType &&event, bool sync) { if (_handling) { sync = false; } if (sync) { _events.push_back(std::move(event)); callHandlers(); } else { if (!this->_callHandlers) { this->_callHandlers = [this]() { callHandlers(); }; } if (_events.empty()) { RegisterPendingObservable(&this->_callHandlers); } _events.push_back(std::move(event)); } } ~ObservableData() { UnregisterObservable(&this->_callHandlers); } private: void callHandlers() { _handling = true; auto events = base::take(_events); for (auto &event : events) { this->notifyEnumerate([this, &event]() { this->_current->handler(event); }); if (this->destroyMeIfEmpty()) { return; } } _handling = false; UnregisterActiveObservable(&this->_callHandlers); } std::deque _events; bool _handling = false; }; template class ObservableData : public CommonObservableData { public: using CommonObservableData::CommonObservableData; void notify(bool sync) { if (_handling) { sync = false; } if (sync) { ++_eventsCount; callHandlers(); } else { if (!this->_callHandlers) { this->_callHandlers = [this]() { callHandlers(); }; } if (!_eventsCount) { RegisterPendingObservable(&this->_callHandlers); } ++_eventsCount; } } ~ObservableData() { UnregisterObservable(&this->_callHandlers); } private: void callHandlers() { _handling = true; auto eventsCount = base::take(_eventsCount); for (int i = 0; i != eventsCount; ++i) { this->notifyEnumerate([this]() { this->_current->handler(); }); if (this->destroyMeIfEmpty()) { return; } } _handling = false; UnregisterActiveObservable(&this->_callHandlers); } int _eventsCount = 0; bool _handling = false; }; template class BaseObservable::is_fast_copy_type::value> : public internal::CommonObservable { public: void notify(bool sync = false) { if (this->_data) { this->_data->notify(sync); } } }; } // namespace internal template > class Observable : public internal::BaseObservable::is_fast_copy_type::value> { public: Observable() = default; Observable(const Observable &other) = delete; Observable(Observable &&other) = default; Observable &operator=(const Observable &other) = delete; Observable &operator=(Observable &&other) = default; }; template class Variable { public: Variable(parameter_type startValue = Type()) : _value(startValue) { } Variable(Variable &&other) = default; Variable &operator=(Variable &&other) = default; parameter_type value() const { return _value; } void setForced(parameter_type newValue, bool sync = false) { _value = newValue; changed().notify(_value, sync); } void set(parameter_type newValue, bool sync = false) { if (_value != newValue) { setForced(newValue, sync); } } template void process(Callback callback, bool sync = false) { callback(_value); changed().notify(_value, sync); } Observable &changed() const { return _changed; } private: Type _value; mutable Observable _changed; }; class Subscriber { protected: template int subscribe(base::Observable &observable, Lambda &&handler) { _subscriptions.push_back(observable.add_subscription(std::forward(handler))); return _subscriptions.size(); } template int subscribe(base::Observable *observable, Lambda &&handler) { return subscribe(*observable, std::forward(handler)); } template int subscribe(const base::Variable &variable, Lambda &&handler) { return subscribe(variable.changed(), std::forward(handler)); } template int subscribe(const base::Variable *variable, Lambda &&handler) { return subscribe(variable->changed(), std::forward(handler)); } void unsubscribe(int index) { if (!index) return; auto count = static_cast(_subscriptions.size()); Assert(index > 0 && index <= count); _subscriptions[index - 1].destroy(); if (index == count) { while (index > 0 && !_subscriptions[--index]) { _subscriptions.pop_back(); } } } ~Subscriber() { auto subscriptions = base::take(_subscriptions); for (auto &subscription : subscriptions) { subscription.destroy(); } } private: std::vector _subscriptions; }; void InitObservables(void(*HandleDelayed)()); void HandleObservables(); template < typename Type, typename = std::enable_if_t>> inline auto ObservableViewer(base::Observable &observable) { return rpl::make_producer([&observable]( const auto &consumer) { auto lifetime = rpl::lifetime(); lifetime.make_state( observable.add_subscription([consumer](auto &&update) { consumer.put_next_forward( std::forward(update)); })); return lifetime; }); } rpl::producer<> ObservableViewer(base::Observable &observable); } // namespace base