Line data    Source code

       1             : /*
       2             :  *  Copyright (C) 2004-2024 Savoir-faire Linux Inc.
       3             :  *
       4             :  *  This program is free software: you can redistribute it and/or modify
       5             :  *  it under the terms of the GNU General Public License as published by
       6             :  *  the Free Software Foundation, either version 3 of the License, or
       7             :  *  (at your option) any later version.
       8             :  *
       9             :  *  This program is distributed in the hope that it will be useful,
      10             :  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
      11             :  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
      12             :  *  GNU General Public License for more details.
      13             :  *
      14             :  *  You should have received a copy of the GNU General Public License
      15             :  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
      16             :  */
      17             : #pragma once
      18             : 
      19             : #include "noncopyable.h"
      20             : 
      21             : #include <cstdlib>
      22             : #include <cstdint>
      23             : #include <memory>
      24             : #include <set>
      25             : #include <list>
      26             : #include <mutex>
      27             : #include <functional>
      28             : #include <ciso646> // fix windows compiler bug
      29             : #ifndef __DEBUG__ // this is only defined on plugins build for debugging
      30             : #include "logger.h"
      31             : #endif
      32             : 
      33             : namespace jami {
      34             : 
      35             : template<typename T>
      36             : class Observer;
      37             : template<typename T>
      38             : class Observable;
      39             : 
      40             : /*=== Observable =============================================================*/
      41             : 
      42             : template<typename T>
      43             : class Observable
      44             : {
      45             : public:
      46        1834 :     Observable()
      47        1834 :         : mutex_()
      48        1834 :         , observers_()
      49        1834 :     {}
      50             : 
      51             :     /**
      52             :      * @brief ~Observable
      53             :      * Detach all observers to avoid making them call this observable when
      54             :      * destroyed
      55             :      */
      56        1834 :     virtual ~Observable()
      57             :     {
      58        1834 :         std::lock_guard lk(mutex_);
      59             : 
      60        2591 :         for (auto& pobs : priority_observers_) {
      61        1175 :             if (auto so = pobs.lock()) {
      62         418 :                 so->detached(this);
      63             :             }
      64             :         }
      65             : 
      66        1840 :         for (auto& o : observers_)
      67           6 :             o->detached(this);
      68        3668 :     }
      69             : 
      70         616 :     bool attach(Observer<T>* o)
      71             :     {
      72         616 :         std::lock_guard lk(mutex_);
      73         616 :         if (o and observers_.insert(o).second) {
      74         616 :             o->attached(this);
      75         616 :             return true;
      76             :         }
      77           0 :         return false;
      78         616 :     }
      79             : 
      80         788 :     void attachPriorityObserver(std::shared_ptr<Observer<T>> o)
      81             :     {
      82         788 :         std::lock_guard lk(mutex_);
      83         788 :         priority_observers_.push_back(o);
      84         788 :         o->attached(this);
      85         788 :     }
      86             : 
      87             :     void detachPriorityObserver(Observer<T>* o)
      88             :     {
      89             :         std::lock_guard lk(mutex_);
      90             :         for (auto it = priority_observers_.begin(); it != priority_observers_.end(); it++) {
      91             :             if (auto so = it->lock()) {
      92             :                 if (so.get() == o) {
      93             :                     so->detached(this);
      94             :                     priority_observers_.erase(it);
      95             :                     return;
      96             :                 }
      97             :             }
      98             :         }
      99             :     }
     100             : 
     101         868 :     bool detach(Observer<T>* o)
     102             :     {
     103         868 :         std::lock_guard lk(mutex_);
     104         868 :         if (o and observers_.erase(o)) {
     105         610 :             o->detached(this);
     106         610 :             return true;
     107             :         }
     108         258 :         return false;
     109         868 :     }
     110             : 
     111        8524 :     size_t getObserversCount()
     112             :     {
     113        8524 :         std::lock_guard lk(mutex_);
     114       17048 :         return observers_.size() + priority_observers_.size();
     115        8524 :     }
     116             : 
     117             : protected:
     118       17013 :     void notify(T data)
     119             :     {
     120       17013 :         std::lock_guard lk(mutex_);
     121       25494 :         for (auto it = priority_observers_.begin(); it != priority_observers_.end();) {
     122       16962 :             if (auto so = it->lock()) {
     123        8450 :                 it++;
     124             :                 try {
     125        8450 :                     so->update(this, data);
     126           0 :                 } catch (std::exception& e) {
     127             : #ifndef __DEBUG__
     128           0 :                     JAMI_ERR() << e.what();
     129             : #endif
     130             :                 }
     131             :             } else {
     132          31 :                 it = priority_observers_.erase(it);
     133             :             }
     134             :         }
     135             : 
     136       35796 :         for (auto observer : observers_) {
     137       18783 :             observer->update(this, data);
     138             :         }
     139       17013 :     }
     140             : 
     141             : private:
     142             :     NON_COPYABLE(Observable);
     143             : 
     144             : protected:
     145             :     std::mutex mutex_; // lock observers_
     146             :     std::list<std::weak_ptr<Observer<T>>> priority_observers_;
     147             :     std::set<Observer<T>*> observers_;
     148             : };
     149             : 
     150             : template<typename T>
     151             : class PublishObservable : public Observable<T>
     152             : {
     153             : public:
     154           1 :     void publish(T data) { this->notify(data); }
     155             : };
     156             : 
     157             : /*=== Observer =============================================================*/
     158             : 
     159             : template<typename T>
     160             : class Observer
     161             : {
     162             : public:
     163        1556 :     virtual ~Observer() {}
     164             :     virtual void update(Observable<T>*, const T&) = 0;
     165         497 :     virtual void attached(Observable<T>*) {}
     166         497 :     virtual void detached(Observable<T>*) {}
     167             : };
     168             : 
     169             : template<typename T>
     170             : class FuncObserver : public Observer<T>
     171             : {
     172             : public:
     173             :     using F = std::function<void(const T&)>;
     174             :     FuncObserver(F f)
     175             :         : f_(f)
     176             :     {}
     177             :     virtual ~FuncObserver() {}
     178             :     void update(Observable<T>*, const T& t) override { f_(t); }
     179             : 
     180             : private:
     181             :     F f_;
     182             : };
     183             : 
     184             : /*=== PublishMapSubject ====================================================*/
     185             : 
     186             : template<typename T1, typename T2>
     187             : class PublishMapSubject : public Observer<T1>, public Observable<T2>
     188             : {
     189             : public:
     190             :     using F = std::function<T2(const T1&)>;
     191             : 
     192         892 :     PublishMapSubject(F f)
     193         892 :         : map_ {f}
     194         892 :     {}
     195             : 
     196        8450 :     void update(Observable<T1>*, const T1& t) override { this->notify(map_(t)); }
     197             : 
     198             :     /**
     199             :      * @brief attached
     200             :      * Here we just make sure that the PublishMapSubject is only attached to one
     201             :      * Observable at a time.
     202             :      * @param srcObs
     203             :      */
     204         788 :     virtual void attached(Observable<T1>* srcObs) override
     205             :     {
     206         788 :         if (obs_ != nullptr && obs_ != srcObs) {
     207           0 :             obs_->detach(this);
     208           0 :             obs_ = srcObs;
     209             :         }
     210         788 :     }
     211             : 
     212             :     /**
     213             :      * @brief detached
     214             :      * Since a MapSubject is only attached to one Observable, when detached
     215             :      * We should detach all of it observers
     216             :      */
     217        1310 :     virtual void detached(Observable<T1>*) override
     218             :     {
     219        1310 :         std::lock_guard lk(this->mutex_);
     220        1310 :         for (auto& pobs : this->priority_observers_) {
     221           0 :             if (auto so = pobs.lock()) {
     222           0 :                 so->detached(this);
     223             :             }
     224             :         }
     225        1310 :         for (auto& o : this->observers_)
     226           0 :             o->detached(this);
     227        1310 :     }
     228             : 
     229             :     /**
     230             :      * @brief ~PublishMapSubject()
     231             :      * Detach all observers to avoid making them call this observable when
     232             :      * destroyed
     233             :      **/
     234         892 :     ~PublishMapSubject() { detached(nullptr); }
     235             : 
     236             : private:
     237             :     F map_;
     238             :     Observable<T1>* obs_ = nullptr;
     239             : };
     240             : 
     241             : }; // namespace jami