Move sync notifier contruction out of syncer thread.

Add thread safety checks to ensure that all the methods are called on
the same thread.

BUG=
TEST=


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

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

4 files changed, 61 insertions, 30 deletions

diff --git a/jingle/notifier/listener/mediator_thread_impl.cc b/jingle/notifier/listener/mediator_thread_impl.cc
index 6887f10..0d96d01 100644
--- a/jingle/notifier/listener/mediator_thread_impl.cc
+++ b/jingle/notifier/listener/mediator_thread_impl.cc
@@ -22,14 +22,15 @@ namespace notifier {
 MediatorThreadImpl::MediatorThreadImpl(
     const NotifierOptions& notifier_options)
     : observers_(new ObserverListThreadSafe<Observer>()),
-      parent_message_loop_(MessageLoop::current()),
+      construction_message_loop_(MessageLoop::current()),
+      method_message_loop_(NULL),
       notifier_options_(notifier_options),
       worker_thread_("MediatorThread worker thread") {
-  DCHECK(parent_message_loop_);
+  DCHECK(construction_message_loop_);
 }
 
 MediatorThreadImpl::~MediatorThreadImpl() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  DCHECK_EQ(MessageLoop::current(), construction_message_loop_);
   // If the worker thread is still around, we need to call Logout() so
   // that all the variables living it get destroyed properly (i.e., on
   // the worker thread).
@@ -39,15 +40,17 @@ MediatorThreadImpl::~MediatorThreadImpl() {
 }
 
 void MediatorThreadImpl::AddObserver(Observer* observer) {
+  CheckOrSetValidThread();
   observers_->AddObserver(observer);
 }
 
 void MediatorThreadImpl::RemoveObserver(Observer* observer) {
+  CheckOrSetValidThread();
   observers_->RemoveObserver(observer);
 }
 
 void MediatorThreadImpl::Start() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  DCHECK_EQ(MessageLoop::current(), construction_message_loop_);
   // We create the worker thread as an IO thread in preparation for
   // making this use Chrome sockets.
   const base::Thread::Options options(MessageLoop::TYPE_IO, 0);
@@ -57,14 +60,15 @@ void MediatorThreadImpl::Start() {
 }
 
 void MediatorThreadImpl::Login(const buzz::XmppClientSettings& settings) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  CheckOrSetValidThread();
+
   worker_message_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(this, &MediatorThreadImpl::DoLogin, settings));
 }
 
 void MediatorThreadImpl::Logout() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  CheckOrSetValidThread();
   worker_message_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(this, &MediatorThreadImpl::DoDisconnect));
@@ -76,7 +80,7 @@ void MediatorThreadImpl::Logout() {
 }
 
 void MediatorThreadImpl::ListenForUpdates() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  CheckOrSetValidThread();
   worker_message_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(this,
@@ -85,7 +89,7 @@ void MediatorThreadImpl::ListenForUpdates() {
 
 void MediatorThreadImpl::SubscribeForUpdates(
     const SubscriptionList& subscriptions) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  CheckOrSetValidThread();
   worker_message_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(
@@ -96,7 +100,7 @@ void MediatorThreadImpl::SubscribeForUpdates(
 
 void MediatorThreadImpl::SendNotification(
     const Notification& data) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  CheckOrSetValidThread();
   worker_message_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(this, &MediatorThreadImpl::DoSendNotification,
@@ -105,7 +109,7 @@ void MediatorThreadImpl::SendNotification(
 
 void MediatorThreadImpl::UpdateXmppSettings(
     const buzz::XmppClientSettings& settings) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  CheckOrSetValidThread();
   worker_message_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(this,
@@ -118,7 +122,7 @@ MessageLoop* MediatorThreadImpl::worker_message_loop() {
   DCHECK(current_message_loop);
   MessageLoop* worker_message_loop = worker_thread_.message_loop();
   DCHECK(worker_message_loop);
-  DCHECK(current_message_loop == parent_message_loop_ ||
+  DCHECK(current_message_loop == method_message_loop_ ||
          current_message_loop == worker_message_loop);
   return worker_message_loop;
 }
@@ -229,4 +233,12 @@ void MediatorThreadImpl::OnDisconnect() {
   observers_->Notify(&Observer::OnConnectionStateChange, false);
 }
 
+void MediatorThreadImpl::CheckOrSetValidThread() {
+  if (method_message_loop_) {
+    DCHECK_EQ(MessageLoop::current(), method_message_loop_);
+  } else {
+    method_message_loop_ = MessageLoop::current();
+  }
+}
+
 }  // namespace notifier
diff --git a/jingle/notifier/listener/mediator_thread_impl.h b/jingle/notifier/listener/mediator_thread_impl.h
index e5ea7b4..0d1f1c4 100644
--- a/jingle/notifier/listener/mediator_thread_impl.h
+++ b/jingle/notifier/listener/mediator_thread_impl.h
@@ -86,7 +86,8 @@ class MediatorThreadImpl : public MediatorThread, public LoginDelegate,
   MessageLoop* worker_message_loop();
 
   scoped_refptr<ObserverListThreadSafe<Observer> > observers_;
-  MessageLoop* parent_message_loop_;
+  MessageLoop* construction_message_loop_;
+  MessageLoop* method_message_loop_;
   base::WeakPtr<talk_base::Task> base_task_;
 
  private:
@@ -101,6 +102,7 @@ class MediatorThreadImpl : public MediatorThread, public LoginDelegate,
   void DoSendNotification(
       const Notification& data);
   void DoUpdateXmppSettings(const buzz::XmppClientSettings& settings);
+  void CheckOrSetValidThread();
 
   const NotifierOptions notifier_options_;
 
diff --git a/jingle/notifier/listener/talk_mediator_impl.cc b/jingle/notifier/listener/talk_mediator_impl.cc
index fa0fc4fc..5bd70dc 100644
--- a/jingle/notifier/listener/talk_mediator_impl.cc
+++ b/jingle/notifier/listener/talk_mediator_impl.cc
@@ -5,6 +5,7 @@
 #include "jingle/notifier/listener/talk_mediator_impl.h"
 
 #include "base/logging.h"
+#include "base/message_loop.h"
 #include "jingle/notifier/base/notifier_options_util.h"
 
 namespace notifier {
@@ -14,21 +15,20 @@ TalkMediatorImpl::TalkMediatorImpl(
     const NotifierOptions& notifier_options)
     : delegate_(NULL),
       mediator_thread_(mediator_thread),
-      notifier_options_(notifier_options) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+      notifier_options_(notifier_options),
+      construction_message_loop_(MessageLoop::current()),
+      method_message_loop_(NULL) {
   mediator_thread_->Start();
   state_.started = 1;
 }
 
 TalkMediatorImpl::~TalkMediatorImpl() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  if (state_.started) {
-    Logout();
-  }
+  DCHECK_EQ(MessageLoop::current(), construction_message_loop_);
+  DCHECK(!state_.started);
 }
 
 bool TalkMediatorImpl::Login() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  CheckOrSetValidThread();
   // Connect to the mediator thread and start processing messages.
   mediator_thread_->AddObserver(this);
   if (state_.initialized && !state_.logging_in && !state_.logged_in) {
@@ -40,7 +40,7 @@ bool TalkMediatorImpl::Login() {
 }
 
 bool TalkMediatorImpl::Logout() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  CheckOrSetValidThread();
   if (state_.started) {
     state_.started = 0;
     state_.logging_in = 0;
@@ -56,7 +56,7 @@ bool TalkMediatorImpl::Logout() {
 }
 
 bool TalkMediatorImpl::SendNotification(const Notification& data) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  CheckOrSetValidThread();
   if (state_.logged_in && state_.subscribed) {
     mediator_thread_->SendNotification(data);
     return true;
@@ -65,15 +65,14 @@ bool TalkMediatorImpl::SendNotification(const Notification& data) {
 }
 
 void TalkMediatorImpl::SetDelegate(TalkMediator::Delegate* delegate) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  DCHECK_EQ(MessageLoop::current(), construction_message_loop_);
   delegate_ = delegate;
 }
 
 void TalkMediatorImpl::SetAuthToken(const std::string& email,
                                     const std::string& token,
                                     const std::string& token_service) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-
+  CheckOrSetValidThread();
   xmpp_settings_ =
       MakeXmppClientSettings(notifier_options_, email, token, token_service);
 
@@ -87,7 +86,7 @@ void TalkMediatorImpl::SetAuthToken(const std::string& email,
 }
 
 void TalkMediatorImpl::AddSubscription(const Subscription& subscription) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  CheckOrSetValidThread();
   subscriptions_.push_back(subscription);
   if (state_.logged_in) {
     VLOG(1) << "Resubscribing for updates, a new service got added";
@@ -97,7 +96,7 @@ void TalkMediatorImpl::AddSubscription(const Subscription& subscription) {
 
 
 void TalkMediatorImpl::OnConnectionStateChange(bool logged_in) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  CheckOrSetValidThread();
   // If we just lost connection, then the MediatorThread implementation will
   // try to log in again. We need to set state_.logging_in to true in that case.
   state_.logging_in = !logged_in;
@@ -116,7 +115,7 @@ void TalkMediatorImpl::OnConnectionStateChange(bool logged_in) {
 }
 
 void TalkMediatorImpl::OnSubscriptionStateChange(bool subscribed) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  CheckOrSetValidThread();
   state_.subscribed = subscribed;
   VLOG(1) << "P2P: " << (subscribed ? "subscribed" : "unsubscribed");
   if (delegate_)
@@ -125,18 +124,26 @@ void TalkMediatorImpl::OnSubscriptionStateChange(bool subscribed) {
 
 void TalkMediatorImpl::OnIncomingNotification(
     const Notification& notification) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  CheckOrSetValidThread();
   VLOG(1) << "P2P: Updates are available on the server.";
   if (delegate_)
     delegate_->OnIncomingNotification(notification);
 }
 
 void TalkMediatorImpl::OnOutgoingNotification() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
+  CheckOrSetValidThread();
   VLOG(1) << "P2P: Peers were notified that updates are available on the "
              "server.";
   if (delegate_)
     delegate_->OnOutgoingNotification();
 }
 
+void TalkMediatorImpl::CheckOrSetValidThread() {
+  if (method_message_loop_) {
+    DCHECK_EQ(MessageLoop::current(), method_message_loop_);
+  } else {
+    method_message_loop_ = MessageLoop::current();
+  }
+}
+
 }  // namespace notifier
diff --git a/jingle/notifier/listener/talk_mediator_impl.h b/jingle/notifier/listener/talk_mediator_impl.h
index 5ae0030..d563ed0 100644
--- a/jingle/notifier/listener/talk_mediator_impl.h
+++ b/jingle/notifier/listener/talk_mediator_impl.h
@@ -20,6 +20,8 @@
 #include "jingle/notifier/listener/talk_mediator.h"
 #include "talk/xmpp/xmppclientsettings.h"
 
+class MessageLoop;
+
 namespace notifier {
 
 class TalkMediatorImpl
@@ -36,13 +38,18 @@ class TalkMediatorImpl
 
   // TalkMediator implementation.
 
+  // Should be called on the same thread as the constructor.
   virtual void SetDelegate(TalkMediator::Delegate* delegate);
 
+  // All the methods below should be called on the same thread. It may or may
+  // not be same as the thread on which the object was constructed.
+
   // |email| must be a valid email address (e.g., foo@bar.com).
   virtual void SetAuthToken(const std::string& email,
                             const std::string& token,
                             const std::string& token_service);
   virtual bool Login();
+  // Users must call Logout once Login is called.
   virtual bool Logout();
 
   virtual bool SendNotification(const Notification& data);
@@ -74,7 +81,7 @@ class TalkMediatorImpl
     unsigned int subscribed : 1;   // Subscribed to the xmpp receiving channel.
   };
 
-  base::NonThreadSafe non_thread_safe_;
+  void CheckOrSetValidThread();
 
   // Delegate, which we don't own.  May be NULL.
   TalkMediator::Delegate* delegate_;
@@ -92,6 +99,9 @@ class TalkMediatorImpl
 
   SubscriptionList subscriptions_;
 
+  MessageLoop* construction_message_loop_;
+  MessageLoop* method_message_loop_;
+
   FRIEND_TEST_ALL_PREFIXES(TalkMediatorImplTest, SetAuthToken);
   FRIEND_TEST_ALL_PREFIXES(TalkMediatorImplTest, SendNotification);
   FRIEND_TEST_ALL_PREFIXES(TalkMediatorImplTest, MediatorThreadCallbacks);