[Sync] Replace TalkMediator delegate handling with thread-safe observer lists

Speculative fix for crash in bug.

BUG=chromium-os:7797
TEST=Manually with bug repro

Review URL: http://codereview.chromium.org/3837002

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@62825 0039d316-1c4b-4281-b951-d872f2087c98

6 files changed, 49 insertions, 110 deletions

diff --git a/jingle/notifier/listener/mediator_thread.h b/jingle/notifier/listener/mediator_thread.h
index 643b1e5..a0047f5 100644
--- a/jingle/notifier/listener/mediator_thread.h
+++ b/jingle/notifier/listener/mediator_thread.h
@@ -23,9 +23,9 @@ class MediatorThread {
  public:
   virtual ~MediatorThread() {}
 
-  class Delegate {
+  class Observer {
    public:
-    virtual ~Delegate() {}
+    virtual ~Observer() {}
 
     virtual void OnConnectionStateChange(bool logged_in) = 0;
 
@@ -37,10 +37,13 @@ class MediatorThread {
     virtual void OnOutgoingNotification() = 0;
   };
 
-  // |delegate| can be NULL if we're shutting down.
-  // TODO(akalin): Handle messages during shutdown gracefully so that
-  // we don't have to deal with NULL delegates.
-  virtual void SetDelegate(Delegate* delegate) = 0;
+  // Must be thread-safe (i.e., callable on any thread).
+  virtual void AddObserver(Observer* observer) = 0;
+
+  // Must be called on the same thread that AddObserver() was called
+  // with the given observer.
+  virtual void RemoveObserver(Observer* observer) = 0;
+
   virtual void Login(const buzz::XmppClientSettings& settings) = 0;
   virtual void Logout() = 0;
   virtual void Start() = 0;
diff --git a/jingle/notifier/listener/mediator_thread_impl.cc b/jingle/notifier/listener/mediator_thread_impl.cc
index 362da19..8432198 100644
--- a/jingle/notifier/listener/mediator_thread_impl.cc
+++ b/jingle/notifier/listener/mediator_thread_impl.cc
@@ -25,7 +25,7 @@ DISABLE_RUNNABLE_METHOD_REFCOUNT(notifier::MediatorThreadImpl);
 namespace notifier {
 
 MediatorThreadImpl::MediatorThreadImpl(const NotifierOptions& notifier_options)
-    : delegate_(NULL),
+    : observers_(new ObserverListThreadSafe<Observer>()),
       parent_message_loop_(MessageLoop::current()),
       notifier_options_(notifier_options),
       worker_thread_("MediatorThread worker thread") {
@@ -42,9 +42,12 @@ MediatorThreadImpl::~MediatorThreadImpl() {
   }
 }
 
-void MediatorThreadImpl::SetDelegate(Delegate* delegate) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  delegate_ = delegate;
+void MediatorThreadImpl::AddObserver(Observer* observer) {
+  observers_->AddObserver(observer);
+}
+
+void MediatorThreadImpl::RemoveObserver(Observer* observer) {
+  observers_->RemoveObserver(observer);
 }
 
 void MediatorThreadImpl::Start() {
@@ -71,13 +74,7 @@ void MediatorThreadImpl::Logout() {
       NewRunnableMethod(this, &MediatorThreadImpl::DoDisconnect));
   // TODO(akalin): Decomp this into a separate stop method.
   worker_thread_.Stop();
-  // Process any messages the worker thread may be posted on our
-  // thread.
-  bool old_state = parent_message_loop_->NestableTasksAllowed();
-  parent_message_loop_->SetNestableTasksAllowed(true);
-  parent_message_loop_->RunAllPending();
-  parent_message_loop_->SetNestableTasksAllowed(old_state);
-  // worker_thread_ should have cleaned all this up.
+  // worker_thread_ should have cleaned this up.
   CHECK(!login_.get());
 }
 
@@ -219,90 +216,31 @@ void MediatorThreadImpl::DoSendNotification(
 void MediatorThreadImpl::OnIncomingNotification(
     const IncomingNotificationData& notification_data) {
   DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  parent_message_loop_->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this,
-          &MediatorThreadImpl::OnIncomingNotificationOnParentThread,
-          notification_data));
-}
-
-void MediatorThreadImpl::OnIncomingNotificationOnParentThread(
-    const IncomingNotificationData& notification_data) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  if (delegate_) {
-    delegate_->OnIncomingNotification(notification_data);
-  }
+  observers_->Notify(&Observer::OnIncomingNotification, notification_data);
 }
 
 void MediatorThreadImpl::OnOutgoingNotification(bool success) {
   DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  parent_message_loop_->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this,
-          &MediatorThreadImpl::OnOutgoingNotificationOnParentThread,
-          success));
-}
-
-void MediatorThreadImpl::OnOutgoingNotificationOnParentThread(
-    bool success) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  if (delegate_ && success) {
-    delegate_->OnOutgoingNotification();
+  if (success) {
+    observers_->Notify(&Observer::OnOutgoingNotification);
   }
 }
 
 void MediatorThreadImpl::OnConnect(base::WeakPtr<talk_base::Task> base_task) {
   DCHECK_EQ(MessageLoop::current(), worker_message_loop());
   base_task_ = base_task;
-  parent_message_loop_->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this,
-          &MediatorThreadImpl::OnConnectOnParentThread));
-}
-
-void MediatorThreadImpl::OnConnectOnParentThread() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  if (delegate_) {
-    delegate_->OnConnectionStateChange(true);
-  }
+  observers_->Notify(&Observer::OnConnectionStateChange, true);
 }
 
 void MediatorThreadImpl::OnDisconnect() {
   DCHECK_EQ(MessageLoop::current(), worker_message_loop());
   base_task_.reset();
-  parent_message_loop_->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this,
-          &MediatorThreadImpl::OnDisconnectOnParentThread));
-}
-
-void MediatorThreadImpl::OnDisconnectOnParentThread() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  if (delegate_) {
-    delegate_->OnConnectionStateChange(false);
-  }
+  observers_->Notify(&Observer::OnConnectionStateChange, false);
 }
 
 void MediatorThreadImpl::OnSubscriptionStateChange(bool success) {
   DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  parent_message_loop_->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this,
-          &MediatorThreadImpl::OnSubscriptionStateChangeOnParentThread,
-          success));
-}
-
-void MediatorThreadImpl::OnSubscriptionStateChangeOnParentThread(
-    bool success) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  if (delegate_) {
-    delegate_->OnSubscriptionStateChange(success);
-  }
+  observers_->Notify(&Observer::OnSubscriptionStateChange, success);
 }
 
 }  // namespace notifier
diff --git a/jingle/notifier/listener/mediator_thread_impl.h b/jingle/notifier/listener/mediator_thread_impl.h
index 1792ca4..32a49e8 100644
--- a/jingle/notifier/listener/mediator_thread_impl.h
+++ b/jingle/notifier/listener/mediator_thread_impl.h
@@ -24,6 +24,8 @@
 #include <vector>
 
 #include "base/basictypes.h"
+#include "base/observer_list_threadsafe.h"
+#include "base/ref_counted.h"
 #include "base/scoped_ptr.h"
 #include "base/thread.h"
 #include "jingle/notifier/base/notifier_options.h"
@@ -60,7 +62,8 @@ class MediatorThreadImpl
   explicit MediatorThreadImpl(const NotifierOptions& notifier_options);
   virtual ~MediatorThreadImpl();
 
-  virtual void SetDelegate(Delegate* delegate);
+  virtual void AddObserver(Observer* observer);
+  virtual void RemoveObserver(Observer* observer);
 
   // Start the thread.
   virtual void Start();
@@ -89,7 +92,7 @@ class MediatorThreadImpl
   void OnConnect(base::WeakPtr<talk_base::Task> parent);
   void OnSubscriptionStateChange(bool success);
 
-  Delegate* delegate_;
+  scoped_refptr<ObserverListThreadSafe<Observer> > observers_;
   MessageLoop* parent_message_loop_;
   base::WeakPtr<talk_base::Task> base_task_;
 
@@ -102,15 +105,6 @@ class MediatorThreadImpl
   void DoSendNotification(
       const OutgoingNotificationData& data);
 
-  // Equivalents of the above functions called from the parent thread.
-  void OnIncomingNotificationOnParentThread(
-      const IncomingNotificationData& notification_data);
-  void OnOutgoingNotificationOnParentThread(bool success);
-  void OnConnectOnParentThread();
-  void OnDisconnectOnParentThread();
-  void OnSubscriptionStateChangeOnParentThread(
-      bool success);
-
   const NotifierOptions notifier_options_;
 
   base::Thread worker_thread_;
diff --git a/jingle/notifier/listener/mediator_thread_mock.h b/jingle/notifier/listener/mediator_thread_mock.h
index 8a1eaec..99b880e 100644
--- a/jingle/notifier/listener/mediator_thread_mock.h
+++ b/jingle/notifier/listener/mediator_thread_mock.h
@@ -19,7 +19,7 @@ namespace notifier {
 
 class MockMediatorThread : public MediatorThread {
  public:
-  MockMediatorThread() : delegate_(NULL) {
+  MockMediatorThread() : observer_(NULL) {
     Reset();
   }
 
@@ -34,22 +34,26 @@ class MockMediatorThread : public MediatorThread {
     send_calls = 0;
   }
 
-  virtual void SetDelegate(Delegate* delegate) {
-    delegate_ = delegate;
+  virtual void AddObserver(Observer* observer) {
+    observer_ = observer;
+  }
+
+  virtual void RemoveObserver(Observer* observer) {
+    observer_ = NULL;
   }
 
   // Overridden from MediatorThread
   virtual void Login(const buzz::XmppClientSettings& settings) {
     login_calls++;
-    if (delegate_) {
-      delegate_->OnConnectionStateChange(true);
+    if (observer_) {
+      observer_->OnConnectionStateChange(true);
     }
   }
 
   virtual void Logout() {
     logout_calls++;
-    if (delegate_) {
-      delegate_->OnConnectionStateChange(false);
+    if (observer_) {
+      observer_->OnConnectionStateChange(false);
     }
   }
 
@@ -60,8 +64,8 @@ class MockMediatorThread : public MediatorThread {
   virtual void SubscribeForUpdates(
       const std::vector<std::string>& subscribed_services_list) {
     subscribe_calls++;
-    if (delegate_) {
-      delegate_->OnSubscriptionStateChange(true);
+    if (observer_) {
+      observer_->OnSubscriptionStateChange(true);
     }
   }
 
@@ -71,18 +75,18 @@ class MockMediatorThread : public MediatorThread {
 
   virtual void SendNotification(const OutgoingNotificationData &) {
     send_calls++;
-    if (delegate_) {
-      delegate_->OnOutgoingNotification();
+    if (observer_) {
+      observer_->OnOutgoingNotification();
     }
   }
 
   void ReceiveNotification(const IncomingNotificationData& data) {
-    if (delegate_) {
-      delegate_->OnIncomingNotification(data);
+    if (observer_) {
+      observer_->OnIncomingNotification(data);
     }
   }
 
-  Delegate* delegate_;
+  Observer* observer_;
   // Intneral State
   int login_calls;
   int logout_calls;
diff --git a/jingle/notifier/listener/talk_mediator_impl.cc b/jingle/notifier/listener/talk_mediator_impl.cc
index a1fec8a..2221e04 100644
--- a/jingle/notifier/listener/talk_mediator_impl.cc
+++ b/jingle/notifier/listener/talk_mediator_impl.cc
@@ -34,7 +34,7 @@ TalkMediatorImpl::~TalkMediatorImpl() {
 bool TalkMediatorImpl::Login() {
   DCHECK(non_thread_safe_.CalledOnValidThread());
   // Connect to the mediator thread and start processing messages.
-  mediator_thread_->SetDelegate(this);
+  mediator_thread_->AddObserver(this);
   if (state_.initialized && !state_.logging_in && !state_.logged_in) {
     state_.logging_in = true;
     mediator_thread_->Login(xmpp_settings_);
@@ -52,7 +52,7 @@ bool TalkMediatorImpl::Logout() {
     state_.subscribed = 0;
     // We do not want to be called back during logout since we may be
     // closing.
-    mediator_thread_->SetDelegate(NULL);
+    mediator_thread_->RemoveObserver(this);
     mediator_thread_->Logout();
     return true;
   }
diff --git a/jingle/notifier/listener/talk_mediator_impl.h b/jingle/notifier/listener/talk_mediator_impl.h
index 1da4d25..d5b9d2f 100644
--- a/jingle/notifier/listener/talk_mediator_impl.h
+++ b/jingle/notifier/listener/talk_mediator_impl.h
@@ -23,7 +23,7 @@
 namespace notifier {
 
 class TalkMediatorImpl
-    : public TalkMediator, public MediatorThread::Delegate {
+    : public TalkMediator, public MediatorThread::Observer {
  public:
   // Takes ownership of |mediator_thread|.
   TalkMediatorImpl(