proxygen: wangle::Observable< T, InlineObservers

#include <Observable.h>

Classes
class	Unsubscriber

Public Member Functions
	Observable ()

	Observable (Observable &&other)=delete

virtual	~Observable ()

virtual Subscription< T >	subscribe (ObserverPtr< T > observer)

virtual void	observe (ObserverPtr< T > observer)

virtual void	observe (Observer< T > *observer)

ObservablePtr< T >	observeOn (SchedulerPtr scheduler)

std::unique_ptr< Observable >	subscribeOn (SchedulerPtr scheduler)

Protected Member Functions
template<class F >
void	forEachObserver (F f)

Private Types
typedef folly::small_vector< Observer< T > *, InlineObservers >	ObserverList

typedef std::map< uint64_t, ObserverPtr< T > >	SubscriberMap

Private Member Functions
Subscription< T >	subscribeImpl (ObserverPtr< T > observer, bool indefinite)

void	unsubscribe (uint64_t id)

Subscription< T >	makeSubscription (bool indefinite)

Private Attributes
std::shared_ptr< Unsubscriber >	unsubscriber_ {nullptr}

folly::MicroSpinLock	unsubscriberLock_ {0}

std::atomic< uint64_t >	nextSubscriptionId_

folly::SharedMutex	observersLock_

folly::ThreadLocalPtr< bool >	inCallback_

ObserverList	observers_

folly::ThreadLocalPtr< ObserverList >	newObservers_

SubscriberMap	subscribers_

folly::ThreadLocalPtr< SubscriberMap >	newSubscribers_

folly::ThreadLocalPtr< std::vector< uint64_t > >	oldSubscribers_

Friends
class	Subscription< T >

Detailed Description

template<class T, size_t InlineObservers>
class wangle::Observable< T, InlineObservers >

Definition at line 35 of file Observable.h.

Member Typedef Documentation

template<class T, size_t InlineObservers>

typedef folly::small_vector<Observer<T>*, InlineObservers> wangle::Observable< T, InlineObservers >::ObserverList

private

Definition at line 275 of file Observable.h.

template<class T, size_t InlineObservers>

typedef std::map<uint64_t, ObserverPtr<T> > wangle::Observable< T, InlineObservers >::SubscriberMap

private

Definition at line 279 of file Observable.h.

Constructor & Destructor Documentation

template<class T, size_t InlineObservers>

wangle::Observable< T, InlineObservers >::Observable ( )

inline

Definition at line 37 of file Observable.h.

Referenced by wangle::Observable< T >::Observable().

37 : nextSubscriptionId_{1} {}

wangle::Observable::nextSubscriptionId_

std::atomic< uint64_t > nextSubscriptionId_

Definition: Observable.h:271

template<class T, size_t InlineObservers>

wangle::Observable< T, InlineObservers >::Observable ( Observable< T, InlineObservers > && other )

delete

template<class T, size_t InlineObservers>

virtual wangle::Observable< T, InlineObservers >::~Observable ( )

inlinevirtual

Definition at line 42 of file Observable.h.

                         {
     if (unsubscriber_) {
       unsubscriber_->disable();
     }
   }

Member Function Documentation

template<class T, size_t InlineObservers>

template<class F >

void wangle::Observable< T, InlineObservers >::forEachObserver ( F f )

inlineprotected

Definition at line 147 of file Observable.h.

                             {
     if (UNLIKELY(!inCallback_)) {
       inCallback_.reset(new bool{false});
     }
     CHECK(!(*inCallback_));
     *inCallback_ = true;
 
     {
       folly::SharedMutex::ReadHolder rh(observersLock_);
       for (auto o : observers_) {
         f(o);
       }
 
       for (auto& kv : subscribers_) {
         f(kv.second.get());
       }
     }
 
     if (UNLIKELY((newObservers_ && !newObservers_->empty()) ||
                  (newSubscribers_ && !newSubscribers_->empty()) ||
                  (oldSubscribers_ && !oldSubscribers_->empty()))) {
       {
         folly::SharedMutex::WriteHolder wh(observersLock_);
         if (newObservers_) {
           for (auto observer : *(newObservers_)) {
             observers_.push_back(observer);
           }
           newObservers_->clear();
         }
         if (newSubscribers_) {
           for (auto& kv : *(newSubscribers_)) {
             subscribers_.insert(std::move(kv));
           }
           newSubscribers_->clear();
         }
         if (oldSubscribers_) {
           for (auto id : *(oldSubscribers_)) {
             subscribers_.erase(id);
           }
           oldSubscribers_->clear();
         }
       }
     }
     *inCallback_ = false;
   }

template<class T, size_t InlineObservers>

Subscription<T> wangle::Observable< T, InlineObservers >::makeSubscription ( bool indefinite )

inlineprivate

Definition at line 257 of file Observable.h.

Referenced by wangle::Observable< T >::subscribeImpl().

                                                     {
     if (indefinite) {
       return Subscription<T>(nullptr, nextSubscriptionId_++);
     } else {
       if (!unsubscriber_) {
         std::lock_guard<folly::MicroSpinLock> guard(unsubscriberLock_);
         if (!unsubscriber_) {
           unsubscriber_ = std::make_shared<Unsubscriber>(this);
         }
       }
       return Subscription<T>(unsubscriber_, nextSubscriptionId_++);
     }
   }

template<class T, size_t InlineObservers>

virtual void wangle::Observable< T, InlineObservers >::observe ( ObserverPtr< T > observer )

inlinevirtual

Definition at line 68 of file Observable.h.

                                                 {
     subscribeImpl(observer, true);
   }

template<class T, size_t InlineObservers>

virtual void wangle::Observable< T, InlineObservers >::observe ( Observer< T > * observer )

inlinevirtual

Definition at line 72 of file Observable.h.

                                               {
     if (inCallback_ && *inCallback_) {
       if (!newObservers_) {
         newObservers_.reset(new ObserverList());
       }
       newObservers_->push_back(observer);
     } else {
       folly::SharedMutex::WriteHolder{&observersLock_};
       observers_.push_back(observer);
     }
   }

template<class T, size_t InlineObservers>

ObservablePtr<T> wangle::Observable< T, InlineObservers >::observeOn ( SchedulerPtr scheduler )

inline

Returns a new Observable that will call back on the given Scheduler. The returned Observable must outlive the parent Observable.

Definition at line 93 of file Observable.h.

                                                      {
     // you're right Hannes, if we have Observable::create we don't need this
     // helper class.
     struct ViaSubject : public Observable<T>
     {
       ViaSubject(SchedulerPtr sched,
                  Observable* obs)
         : scheduler_(sched), observable_(obs)
       {}
 
       Subscription<T> subscribe(ObserverPtr<T> o) override {
         return observable_->subscribe(
           Observer<T>::create(
             [=](T val) { scheduler_->add([o, val] { o->onNext(val); }); },
             [=](Error e) { scheduler_->add([o, e] { o->onError(e); }); },
             [=]() { scheduler_->add([o] { o->onCompleted(); }); }));
       }
 
      protected:
       SchedulerPtr scheduler_;
       Observable* observable_;
     };
 
     return std::make_shared<ViaSubject>(scheduler, this);
   }

template<class T, size_t InlineObservers>

virtual Subscription<T> wangle::Observable< T, InlineObservers >::subscribe ( ObserverPtr< T > observer )

inlinevirtual

Definition at line 64 of file Observable.h.

Referenced by wangle::Observable< T >::observeOn(), and wangle::Observable< T >::subscribeOn().

                                                              {
     return subscribeImpl(observer, false);
   }

template<class T, size_t InlineObservers>

Subscription<T> wangle::Observable< T, InlineObservers >::subscribeImpl	(	ObserverPtr< T >	observer,
		bool	indefinite
	)

inlineprivate

Definition at line 194 of file Observable.h.

Referenced by wangle::Observable< T >::observe(), and wangle::Observable< T >::subscribe().

                                                                           {
     auto subscription = makeSubscription(indefinite);
     typename SubscriberMap::value_type kv{subscription.id_, std::move(observer)};
     if (inCallback_ && *inCallback_) {
       if (!newSubscribers_) {
         newSubscribers_.reset(new SubscriberMap());
       }
       newSubscribers_->insert(std::move(kv));
     } else {
       folly::SharedMutex::WriteHolder{&observersLock_};
       subscribers_.insert(std::move(kv));
     }
     return subscription;
   }

template<class T, size_t InlineObservers>

std::unique_ptr<Observable> wangle::Observable< T, InlineObservers >::subscribeOn ( SchedulerPtr scheduler )

inline

Returns a new Observable that will subscribe to this parent Observable via the given Scheduler. This can be subtle and confusing at first, see http://www.introtorx.com/Content/v1.0.10621.0/15_SchedulingAndThreading.html#SubscribeOnObserveOn

Definition at line 122 of file Observable.h.

                                                                 {
     struct Subject_ : public Subject<T> {
      public:
       Subject_(SchedulerPtr s, Observable* o) : scheduler_(s), observable_(o) {
       }
 
       Subscription<T> subscribe(ObserverPtr<T> o) {
         scheduler_->add([=] {
           observable_->subscribe(o);
         });
         return Subscription<T>(nullptr, 0); // TODO
       }
 
      protected:
       SchedulerPtr scheduler_;
       Observable* observable_;
     };
 
     return std::make_unique<Subject_>(scheduler, this);
   }

template<class T, size_t InlineObservers>

void wangle::Observable< T, InlineObservers >::unsubscribe ( uint64_t id )

inlineprivate

Definition at line 238 of file Observable.h.

                                 {
     if (inCallback_ && *inCallback_) {
       if (!oldSubscribers_) {
         oldSubscribers_.reset(new std::vector<uint64_t>());
       }
       if (newSubscribers_) {
         auto it = newSubscribers_->find(id);
         if (it != newSubscribers_->end()) {
           newSubscribers_->erase(it);
           return;
         }
       }
       oldSubscribers_->push_back(id);
     } else {
       folly::SharedMutex::WriteHolder{&observersLock_};
       subscribers_.erase(id);
     }
   }