michael@0: // Copyright (c) 2012 The Chromium Authors. All rights reserved.
michael@0: // Use of this source code is governed by a BSD-style license that can be
michael@0: // found in the LICENSE file.
michael@0: 
michael@0: #ifndef BASE_OBSERVER_LIST_THREADSAFE_H_
michael@0: #define BASE_OBSERVER_LIST_THREADSAFE_H_
michael@0: 
michael@0: #include <algorithm>
michael@0: #include <map>
michael@0: 
michael@0: #include "base/basictypes.h"
michael@0: #include "base/bind.h"
michael@0: #include "base/location.h"
michael@0: #include "base/logging.h"
michael@0: #include "base/memory/ref_counted.h"
michael@0: #include "base/message_loop/message_loop.h"
michael@0: #include "base/message_loop/message_loop_proxy.h"
michael@0: #include "base/observer_list.h"
michael@0: #include "base/stl_util.h"
michael@0: #include "base/threading/platform_thread.h"
michael@0: 
michael@0: ///////////////////////////////////////////////////////////////////////////////
michael@0: //
michael@0: // OVERVIEW:
michael@0: //
michael@0: //   A thread-safe container for a list of observers.
michael@0: //   This is similar to the observer_list (see observer_list.h), but it
michael@0: //   is more robust for multi-threaded situations.
michael@0: //
michael@0: //   The following use cases are supported:
michael@0: //    * Observers can register for notifications from any thread.
michael@0: //      Callbacks to the observer will occur on the same thread where
michael@0: //      the observer initially called AddObserver() from.
michael@0: //    * Any thread may trigger a notification via Notify().
michael@0: //    * Observers can remove themselves from the observer list inside
michael@0: //      of a callback.
michael@0: //    * If one thread is notifying observers concurrently with an observer
michael@0: //      removing itself from the observer list, the notifications will
michael@0: //      be silently dropped.
michael@0: //
michael@0: //   The drawback of the threadsafe observer list is that notifications
michael@0: //   are not as real-time as the non-threadsafe version of this class.
michael@0: //   Notifications will always be done via PostTask() to another thread,
michael@0: //   whereas with the non-thread-safe observer_list, notifications happen
michael@0: //   synchronously and immediately.
michael@0: //
michael@0: //   IMPLEMENTATION NOTES
michael@0: //   The ObserverListThreadSafe maintains an ObserverList for each thread
michael@0: //   which uses the ThreadSafeObserver.  When Notifying the observers,
michael@0: //   we simply call PostTask to each registered thread, and then each thread
michael@0: //   will notify its regular ObserverList.
michael@0: //
michael@0: ///////////////////////////////////////////////////////////////////////////////
michael@0: 
michael@0: // Forward declaration for ObserverListThreadSafeTraits.
michael@0: template <class ObserverType>
michael@0: class ObserverListThreadSafe;
michael@0: 
michael@0: // An UnboundMethod is a wrapper for a method where the actual object is
michael@0: // provided at Run dispatch time.
michael@0: template <class T, class Method, class Params>
michael@0: class UnboundMethod {
michael@0:  public:
michael@0:   UnboundMethod(Method m, const Params& p) : m_(m), p_(p) {
michael@0:     COMPILE_ASSERT(
michael@0:         (base::internal::ParamsUseScopedRefptrCorrectly<Params>::value),
michael@0:         badunboundmethodparams);
michael@0:   }
michael@0:   void Run(T* obj) const {
michael@0:     DispatchToMethod(obj, m_, p_);
michael@0:   }
michael@0:  private:
michael@0:   Method m_;
michael@0:   Params p_;
michael@0: };
michael@0: 
michael@0: // This class is used to work around VS2005 not accepting:
michael@0: //
michael@0: // friend class
michael@0: //     base::RefCountedThreadSafe<ObserverListThreadSafe<ObserverType> >;
michael@0: //
michael@0: // Instead of friending the class, we could friend the actual function
michael@0: // which calls delete.  However, this ends up being
michael@0: // RefCountedThreadSafe::DeleteInternal(), which is private.  So we
michael@0: // define our own templated traits class so we can friend it.
michael@0: template <class T>
michael@0: struct ObserverListThreadSafeTraits {
michael@0:   static void Destruct(const ObserverListThreadSafe<T>* x) {
michael@0:     delete x;
michael@0:   }
michael@0: };
michael@0: 
michael@0: template <class ObserverType>
michael@0: class ObserverListThreadSafe
michael@0:     : public base::RefCountedThreadSafe<
michael@0:         ObserverListThreadSafe<ObserverType>,
michael@0:         ObserverListThreadSafeTraits<ObserverType> > {
michael@0:  public:
michael@0:   typedef typename ObserverList<ObserverType>::NotificationType
michael@0:       NotificationType;
michael@0: 
michael@0:   ObserverListThreadSafe()
michael@0:       : type_(ObserverListBase<ObserverType>::NOTIFY_ALL) {}
michael@0:   explicit ObserverListThreadSafe(NotificationType type) : type_(type) {}
michael@0: 
michael@0:   // Add an observer to the list.  An observer should not be added to
michael@0:   // the same list more than once.
michael@0:   void AddObserver(ObserverType* obs) {
michael@0:     // If there is not a current MessageLoop, it is impossible to notify on it,
michael@0:     // so do not add the observer.
michael@0:     if (!base::MessageLoop::current())
michael@0:       return;
michael@0: 
michael@0:     ObserverList<ObserverType>* list = NULL;
michael@0:     base::PlatformThreadId thread_id = base::PlatformThread::CurrentId();
michael@0:     {
michael@0:       base::AutoLock lock(list_lock_);
michael@0:       if (observer_lists_.find(thread_id) == observer_lists_.end())
michael@0:         observer_lists_[thread_id] = new ObserverListContext(type_);
michael@0:       list = &(observer_lists_[thread_id]->list);
michael@0:     }
michael@0:     list->AddObserver(obs);
michael@0:   }
michael@0: 
michael@0:   // Remove an observer from the list if it is in the list.
michael@0:   // If there are pending notifications in-transit to the observer, they will
michael@0:   // be aborted.
michael@0:   // If the observer to be removed is in the list, RemoveObserver MUST
michael@0:   // be called from the same thread which called AddObserver.
michael@0:   void RemoveObserver(ObserverType* obs) {
michael@0:     ObserverListContext* context = NULL;
michael@0:     ObserverList<ObserverType>* list = NULL;
michael@0:     base::PlatformThreadId thread_id = base::PlatformThread::CurrentId();
michael@0:     {
michael@0:       base::AutoLock lock(list_lock_);
michael@0:       typename ObserversListMap::iterator it = observer_lists_.find(thread_id);
michael@0:       if (it == observer_lists_.end()) {
michael@0:         // This will happen if we try to remove an observer on a thread
michael@0:         // we never added an observer for.
michael@0:         return;
michael@0:       }
michael@0:       context = it->second;
michael@0:       list = &context->list;
michael@0: 
michael@0:       // If we're about to remove the last observer from the list,
michael@0:       // then we can remove this observer_list entirely.
michael@0:       if (list->HasObserver(obs) && list->size() == 1)
michael@0:         observer_lists_.erase(it);
michael@0:     }
michael@0:     list->RemoveObserver(obs);
michael@0: 
michael@0:     // If RemoveObserver is called from a notification, the size will be
michael@0:     // nonzero.  Instead of deleting here, the NotifyWrapper will delete
michael@0:     // when it finishes iterating.
michael@0:     if (list->size() == 0)
michael@0:       delete context;
michael@0:   }
michael@0: 
michael@0:   // Verifies that the list is currently empty (i.e. there are no observers).
michael@0:   void AssertEmpty() const {
michael@0:     base::AutoLock lock(list_lock_);
michael@0:     DCHECK(observer_lists_.empty());
michael@0:   }
michael@0: 
michael@0:   // Notify methods.
michael@0:   // Make a thread-safe callback to each Observer in the list.
michael@0:   // Note, these calls are effectively asynchronous.  You cannot assume
michael@0:   // that at the completion of the Notify call that all Observers have
michael@0:   // been Notified.  The notification may still be pending delivery.
michael@0:   template <class Method>
michael@0:   void Notify(Method m) {
michael@0:     UnboundMethod<ObserverType, Method, Tuple0> method(m, MakeTuple());
michael@0:     Notify<Method, Tuple0>(method);
michael@0:   }
michael@0: 
michael@0:   template <class Method, class A>
michael@0:   void Notify(Method m, const A& a) {
michael@0:     UnboundMethod<ObserverType, Method, Tuple1<A> > method(m, MakeTuple(a));
michael@0:     Notify<Method, Tuple1<A> >(method);
michael@0:   }
michael@0: 
michael@0:   template <class Method, class A, class B>
michael@0:   void Notify(Method m, const A& a, const B& b) {
michael@0:     UnboundMethod<ObserverType, Method, Tuple2<A, B> > method(
michael@0:         m, MakeTuple(a, b));
michael@0:     Notify<Method, Tuple2<A, B> >(method);
michael@0:   }
michael@0: 
michael@0:   template <class Method, class A, class B, class C>
michael@0:   void Notify(Method m, const A& a, const B& b, const C& c) {
michael@0:     UnboundMethod<ObserverType, Method, Tuple3<A, B, C> > method(
michael@0:         m, MakeTuple(a, b, c));
michael@0:     Notify<Method, Tuple3<A, B, C> >(method);
michael@0:   }
michael@0: 
michael@0:   template <class Method, class A, class B, class C, class D>
michael@0:   void Notify(Method m, const A& a, const B& b, const C& c, const D& d) {
michael@0:     UnboundMethod<ObserverType, Method, Tuple4<A, B, C, D> > method(
michael@0:         m, MakeTuple(a, b, c, d));
michael@0:     Notify<Method, Tuple4<A, B, C, D> >(method);
michael@0:   }
michael@0: 
michael@0:   // TODO(mbelshe):  Add more wrappers for Notify() with more arguments.
michael@0: 
michael@0:  private:
michael@0:   // See comment above ObserverListThreadSafeTraits' definition.
michael@0:   friend struct ObserverListThreadSafeTraits<ObserverType>;
michael@0: 
michael@0:   struct ObserverListContext {
michael@0:     explicit ObserverListContext(NotificationType type)
michael@0:         : loop(base::MessageLoopProxy::current()),
michael@0:           list(type) {
michael@0:     }
michael@0: 
michael@0:     scoped_refptr<base::MessageLoopProxy> loop;
michael@0:     ObserverList<ObserverType> list;
michael@0: 
michael@0:     DISALLOW_COPY_AND_ASSIGN(ObserverListContext);
michael@0:   };
michael@0: 
michael@0:   ~ObserverListThreadSafe() {
michael@0:     STLDeleteValues(&observer_lists_);
michael@0:   }
michael@0: 
michael@0:   template <class Method, class Params>
michael@0:   void Notify(const UnboundMethod<ObserverType, Method, Params>& method) {
michael@0:     base::AutoLock lock(list_lock_);
michael@0:     typename ObserversListMap::iterator it;
michael@0:     for (it = observer_lists_.begin(); it != observer_lists_.end(); ++it) {
michael@0:       ObserverListContext* context = (*it).second;
michael@0:       context->loop->PostTask(
michael@0:           FROM_HERE,
michael@0:           base::Bind(&ObserverListThreadSafe<ObserverType>::
michael@0:               template NotifyWrapper<Method, Params>, this, context, method));
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   // Wrapper which is called to fire the notifications for each thread's
michael@0:   // ObserverList.  This function MUST be called on the thread which owns
michael@0:   // the unsafe ObserverList.
michael@0:   template <class Method, class Params>
michael@0:   void NotifyWrapper(ObserverListContext* context,
michael@0:       const UnboundMethod<ObserverType, Method, Params>& method) {
michael@0: 
michael@0:     // Check that this list still needs notifications.
michael@0:     {
michael@0:       base::AutoLock lock(list_lock_);
michael@0:       typename ObserversListMap::iterator it =
michael@0:           observer_lists_.find(base::PlatformThread::CurrentId());
michael@0: 
michael@0:       // The ObserverList could have been removed already.  In fact, it could
michael@0:       // have been removed and then re-added!  If the master list's loop
michael@0:       // does not match this one, then we do not need to finish this
michael@0:       // notification.
michael@0:       if (it == observer_lists_.end() || it->second != context)
michael@0:         return;
michael@0:     }
michael@0: 
michael@0:     {
michael@0:       typename ObserverList<ObserverType>::Iterator it(context->list);
michael@0:       ObserverType* obs;
michael@0:       while ((obs = it.GetNext()) != NULL)
michael@0:         method.Run(obs);
michael@0:     }
michael@0: 
michael@0:     // If there are no more observers on the list, we can now delete it.
michael@0:     if (context->list.size() == 0) {
michael@0:       {
michael@0:         base::AutoLock lock(list_lock_);
michael@0:         // Remove |list| if it's not already removed.
michael@0:         // This can happen if multiple observers got removed in a notification.
michael@0:         // See http://crbug.com/55725.
michael@0:         typename ObserversListMap::iterator it =
michael@0:             observer_lists_.find(base::PlatformThread::CurrentId());
michael@0:         if (it != observer_lists_.end() && it->second == context)
michael@0:           observer_lists_.erase(it);
michael@0:       }
michael@0:       delete context;
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   // Key by PlatformThreadId because in tests, clients can attempt to remove
michael@0:   // observers without a MessageLoop. If this were keyed by MessageLoop, that
michael@0:   // operation would be silently ignored, leaving garbage in the ObserverList.
michael@0:   typedef std::map<base::PlatformThreadId, ObserverListContext*>
michael@0:       ObserversListMap;
michael@0: 
michael@0:   mutable base::Lock list_lock_;  // Protects the observer_lists_.
michael@0:   ObserversListMap observer_lists_;
michael@0:   const NotificationType type_;
michael@0: 
michael@0:   DISALLOW_COPY_AND_ASSIGN(ObserverListThreadSafe);
michael@0: };
michael@0: 
michael@0: #endif  // BASE_OBSERVER_LIST_THREADSAFE_H_