Make sure XMPP logins and notification sends are done on the sync

core thread only.

BUG=none
TEST=unit tests, manual testing of notifications

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

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

6 files changed, 142 insertions, 126 deletions

diff --git a/chrome/browser/sync/engine/all_status.cc b/chrome/browser/sync/engine/all_status.cc
index 68d2513a..64ea9a7 100644
--- a/chrome/browser/sync/engine/all_status.cc
+++ b/chrome/browser/sync/engine/all_status.cc
@@ -17,7 +17,6 @@
 #include "chrome/browser/sync/sessions/session_state.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
 #include "chrome/common/deprecated/event_sys-inl.h"
-#include "chrome/common/net/gaia/gaia_authenticator.h"
 #include "chrome/common/net/notifier/listener/talk_mediator.h"
 
 namespace browser_sync {
@@ -59,17 +58,6 @@ void AllStatus::WatchConnectionManager(ServerConnectionManager* conn_mgr) {
                          &AllStatus::HandleServerConnectionEvent));
 }
 
-void AllStatus::WatchAuthenticator(gaia::GaiaAuthenticator* gaia) {
-  gaia_hookup_.reset(NewEventListenerHookup(gaia->channel(), this,
-                     &AllStatus::HandleGaiaAuthEvent));
-}
-
-void AllStatus::WatchAuthWatcher(AuthWatcher* auth_watcher) {
-  authwatcher_hookup_.reset(
-      NewEventListenerHookup(auth_watcher->channel(), this,
-                             &AllStatus::HandleAuthWatcherEvent));
-}
-
 void AllStatus::WatchSyncerThread(SyncerThread* syncer_thread) {
   syncer_thread_hookup_.reset(
       NewEventListenerHookup(syncer_thread->relay_channel(), this,
@@ -171,21 +159,6 @@ int AllStatus::CalcStatusChanges(Status* old_status) {
   return what_changed;
 }
 
-void AllStatus::HandleGaiaAuthEvent(const gaia::GaiaAuthEvent& gaia_event) {
-  ScopedStatusLockWithNotify lock(this);
-  switch (gaia_event.what_happened) {
-    case gaia::GaiaAuthEvent::GAIA_AUTH_FAILED:
-      status_.authenticated = false;
-      break;
-    case gaia::GaiaAuthEvent::GAIA_AUTH_SUCCEEDED:
-      status_.authenticated = true;
-      break;
-    default:
-      lock.set_notify_plan(DONT_NOTIFY);
-      break;
-  }
-}
-
 void AllStatus::HandleAuthWatcherEvent(const AuthWatcherEvent& auth_event) {
   ScopedStatusLockWithNotify lock(this);
   switch (auth_event.what_happened) {
diff --git a/chrome/browser/sync/engine/all_status.h b/chrome/browser/sync/engine/all_status.h
index f19d6ca..d34a15a 100644
--- a/chrome/browser/sync/engine/all_status.h
+++ b/chrome/browser/sync/engine/all_status.h
@@ -15,14 +15,8 @@
 #include "base/scoped_ptr.h"
 #include "chrome/common/deprecated/event_sys.h"
 
-namespace gaia {
-class GaiaAuthenticator;
-struct GaiaAuthEvent;
-}
-
 namespace browser_sync {
 
-class AuthWatcher;
 class ScopedStatusLockWithNotify;
 class ServerConnectionManager;
 class Syncer;
@@ -100,14 +94,6 @@ class AllStatus {
   void WatchConnectionManager(ServerConnectionManager* conn_mgr);
   void HandleServerConnectionEvent(const ServerConnectionEvent& event);
 
-  // Both WatchAuthenticator/HandleGaiaAuthEvent and WatchAuthWatcher/
-  // HandleAuthWatcherEventachieve have the same goal; use only one of the
-  // following two. (The AuthWatcher is watched under Windows; the
-  // GaiaAuthenticator is watched under Mac/Linux.)
-  void WatchAuthenticator(gaia::GaiaAuthenticator* gaia);
-  void HandleGaiaAuthEvent(const gaia::GaiaAuthEvent& event);
-
-  void WatchAuthWatcher(AuthWatcher* auth_watcher);
   void HandleAuthWatcherEvent(const AuthWatcherEvent& event);
 
   void WatchSyncerThread(SyncerThread* syncer_thread);
@@ -148,8 +134,6 @@ class AllStatus {
   Status status_;
   Channel* const channel_;
   scoped_ptr<EventListenerHookup> conn_mgr_hookup_;
-  scoped_ptr<EventListenerHookup> gaia_hookup_;
-  scoped_ptr<EventListenerHookup> authwatcher_hookup_;
   scoped_ptr<EventListenerHookup> syncer_thread_hookup_;
   scoped_ptr<EventListenerHookup> diskfull_hookup_;
   scoped_ptr<EventListenerHookup> talk_mediator_hookup_;
diff --git a/chrome/browser/sync/engine/auth_watcher.cc b/chrome/browser/sync/engine/auth_watcher.cc
index e674ab3..569dbe9 100644
--- a/chrome/browser/sync/engine/auth_watcher.cc
+++ b/chrome/browser/sync/engine/auth_watcher.cc
@@ -14,7 +14,6 @@
 #include "chrome/browser/sync/util/user_settings.h"
 #include "chrome/common/deprecated/event_sys-inl.h"
 #include "chrome/common/net/gaia/gaia_authenticator.h"
-#include "chrome/common/net/notifier/listener/talk_mediator.h"
 
 // How authentication happens:
 //
@@ -44,14 +43,12 @@ AuthWatcher::AuthWatcher(DirectoryManager* dirman,
                          const string& service_id,
                          const string& gaia_url,
                          UserSettings* user_settings,
-                         gaia::GaiaAuthenticator* gaia_auth,
-                         notifier::TalkMediator* talk_mediator)
+                         gaia::GaiaAuthenticator* gaia_auth)
     : gaia_(gaia_auth),
       dirman_(dirman),
       scm_(scm),
       status_(NOT_AUTHENTICATED),
       user_settings_(user_settings),
-      talk_mediator_(talk_mediator),
       auth_backend_thread_("SyncEngine_AuthWatcherThread"),
       current_attempt_trigger_(AuthWatcherEvent::USER_INITIATED) {
 
@@ -101,16 +98,11 @@ void AuthWatcher::DoRenewAuthToken(const std::string& updated_token) {
   LOG(INFO) << "Updating auth token:" << updated_token;
   scm_->set_auth_token(updated_token);
   gaia_->RenewAuthToken(updated_token);  // Must be on AuthWatcher thread
-  talk_mediator_->SetAuthToken(user_settings_->email(), updated_token,
-                               SYNC_SERVICE_NAME);
-  // TODO(akalin): to see if we need to call Login() here, too.
-
   user_settings_->SetAuthTokenForService(user_settings_->email(),
                                          SYNC_SERVICE_NAME,
                                          updated_token);
 
-  AuthWatcherEvent event = { AuthWatcherEvent::AUTH_RENEWED };
-  NotifyListeners(&event);
+  NotifyAuthChanged(user_settings_->email(), updated_token, true);
 }
 
 void AuthWatcher::AuthenticateWithLsid(const std::string& lsid) {
@@ -164,10 +156,6 @@ void AuthWatcher::DoAuthenticateWithToken(const std::string& gaia_email,
         status_ = GAIA_AUTHENTICATED;
         const std::string& share_name = email;
         user_settings_->SwitchUser(email);
-
-        // Set the authentication token for notifications
-        talk_mediator_->SetAuthToken(email, auth_token, SYNC_SERVICE_NAME);
-        talk_mediator_->Login();
         scm_->set_auth_token(auth_token);
 
         if (!was_authenticated) {
@@ -175,7 +163,7 @@ void AuthWatcher::DoAuthenticateWithToken(const std::string& gaia_email,
                     << share_name << ")";
           dirman_->Open(share_name);
         }
-        NotifyAuthSucceeded(email);
+        NotifyAuthChanged(email, auth_token, false);
         return;
       }
   case Authenticator::BAD_AUTH_TOKEN:
@@ -210,7 +198,7 @@ bool AuthWatcher::AuthenticateLocally(string email) {
     user_settings_->SwitchUser(email);
     LOG(INFO) << "Opening DB for AuthenticateLocally (" << email << ")";
     dirman_->Open(email);
-    NotifyAuthSucceeded(email);
+    NotifyAuthChanged(email, "", false);
     return true;
   } else {
     return false;
@@ -284,11 +272,18 @@ void AuthWatcher::DoAuthenticate(const AuthRequest& request) {
   }
 }
 
-void AuthWatcher::NotifyAuthSucceeded(const string& email) {
+void AuthWatcher::NotifyAuthChanged(const string& email,
+                                    const string& auth_token,
+                                    bool renewed) {
   DCHECK_EQ(MessageLoop::current(), message_loop());
   LOG(INFO) << "NotifyAuthSucceeded";
-  AuthWatcherEvent event = { AuthWatcherEvent::AUTH_SUCCEEDED };
+  AuthWatcherEvent event = {
+    renewed ?
+        AuthWatcherEvent::AUTH_RENEWED :
+        AuthWatcherEvent::AUTH_SUCCEEDED
+  };
   event.user_email = email;
+  event.auth_token = auth_token;
 
   NotifyListeners(&event);
 }
diff --git a/chrome/browser/sync/engine/auth_watcher.h b/chrome/browser/sync/engine/auth_watcher.h
index 60f3ed8..2f7b36a 100644
--- a/chrome/browser/sync/engine/auth_watcher.h
+++ b/chrome/browser/sync/engine/auth_watcher.h
@@ -21,10 +21,6 @@
 #include "chrome/common/net/gaia/gaia_authenticator.h"
 #include "testing/gtest/include/gtest/gtest_prod.h"  // For FRIEND_TEST
 
-namespace notifier {
-class TalkMediator;
-}
-
 namespace syncable {
 struct DirectoryManagerEvent;
 class DirectoryManager;
@@ -59,8 +55,10 @@ struct AuthWatcherEvent {
     return event.what_happened == AUTHWATCHER_DESTROYED;
   }
 
-  // Used for AUTH_SUCCEEDED notification
+  // Used for AUTH_SUCCEEDED/AUTH_RENEWED notification.
   std::string user_email;
+  // May be empty if we're only locally authenticated.
+  std::string auth_token;
 
   // How was this auth attempt initiated?
   enum AuthenticationTrigger {
@@ -92,8 +90,7 @@ class AuthWatcher : public base::RefCountedThreadSafe<AuthWatcher> {
               const std::string& service_id,
               const std::string& gaia_url,
               UserSettings* user_settings,
-              gaia::GaiaAuthenticator* gaia_auth,
-              notifier::TalkMediator* talk_mediator);
+              gaia::GaiaAuthenticator* gaia_auth);
   ~AuthWatcher();
 
   typedef EventChannel<AuthWatcherEvent, Lock> Channel;
@@ -118,7 +115,6 @@ class AuthWatcher : public base::RefCountedThreadSafe<AuthWatcher> {
 
   // Use this to update only the token of the current email address.
   void RenewAuthToken(const std::string& updated_token);
-  void DoRenewAuthToken(const std::string& updated_token);
 
   // Use this version when you don't need the gaia authentication step because
   // you already have a valid LSID cookie for |gaia_email|.
@@ -139,10 +135,12 @@ class AuthWatcher : public base::RefCountedThreadSafe<AuthWatcher> {
   UserSettings* settings() const { return user_settings_; }
   Status status() const { return (Status)status_; }
 
- protected:
+ private:
   void ClearAuthenticationData();
 
-  void NotifyAuthSucceeded(const std::string& email);
+  void NotifyAuthChanged(const std::string& email,
+                         const std::string& auth_token,
+                         bool renewed);
   void HandleServerConnectionEvent(const ServerConnectionEvent& event);
 
   void SaveUserSettings(const std::string& username,
@@ -150,7 +148,8 @@ class AuthWatcher : public base::RefCountedThreadSafe<AuthWatcher> {
 
   MessageLoop* message_loop() { return auth_backend_thread_.message_loop(); }
 
- private:
+  void DoRenewAuthToken(const std::string& updated_token);
+
   // These two helpers should only be called from the auth function.
   // Called when authentication with gaia succeeds, to save credential info.
   void PersistCredentials();
@@ -206,8 +205,6 @@ class AuthWatcher : public base::RefCountedThreadSafe<AuthWatcher> {
   scoped_ptr<EventListenerHookup> connmgr_hookup_;
   Status status_;
   UserSettings* const user_settings_;
-  // Interface to the notifications engine.
-  notifier::TalkMediator* talk_mediator_;
   scoped_ptr<Channel> channel_;
 
   base::Thread auth_backend_thread_;
diff --git a/chrome/browser/sync/engine/auth_watcher_unittest.cc b/chrome/browser/sync/engine/auth_watcher_unittest.cc
index eb39772..e795d31 100644
--- a/chrome/browser/sync/engine/auth_watcher_unittest.cc
+++ b/chrome/browser/sync/engine/auth_watcher_unittest.cc
@@ -10,9 +10,7 @@
 #include "chrome/browser/sync/engine/auth_watcher.h"
 #include "chrome/browser/sync/util/user_settings.h"
 #include "chrome/common/deprecated/event_sys-inl.h"
-#include "chrome/common/net/fake_network_change_notifier_thread.h"
 #include "chrome/common/net/http_return.h"
-#include "chrome/common/net/notifier/listener/talk_mediator_impl.h"
 #include "chrome/common/net/gaia/gaia_authenticator.h"
 #include "chrome/test/sync/engine/mock_server_connection.h"
 #include "chrome/test/sync/engine/test_directory_setter_upper.h"
@@ -98,12 +96,9 @@ class AuthWatcherTest : public testing::Test {
     FilePath user_settings_path = temp_dir_.path().Append(kUserSettingsDB);
     user_settings_->Init(user_settings_path);
     gaia_auth_ = new GaiaAuthMockForAuthWatcher();
-    fake_network_change_notifier_thread_.Start();
-    talk_mediator_.reset(new notifier::TalkMediatorImpl(
-        &fake_network_change_notifier_thread_, false));
     auth_watcher_ = new AuthWatcher(metadb_.manager(), connection_.get(),
         kTestUserAgent, kTestServiceId, kTestGaiaURL,
-        user_settings_.get(), gaia_auth_, talk_mediator_.get());
+        user_settings_.get(), gaia_auth_);
     authwatcher_hookup_.reset(NewEventListenerHookup(auth_watcher_->channel(),
         this, &AuthWatcherTest::HandleAuthWatcherEvent));
   }
@@ -112,8 +107,6 @@ class AuthWatcherTest : public testing::Test {
     metadb_.TearDown();
     auth_watcher_->Shutdown();
     EXPECT_FALSE(auth_watcher()->message_loop());
-    talk_mediator_.reset();
-    fake_network_change_notifier_thread_.Stop();
   }
 
   void HandleAuthWatcherEvent(const AuthWatcherEvent& event) {
@@ -151,10 +144,6 @@ class AuthWatcherTest : public testing::Test {
   scoped_ptr<MockConnectionManager> connection_;
   scoped_ptr<UserSettings> user_settings_;
   GaiaAuthMockForAuthWatcher* gaia_auth_;  // Owned by auth_watcher_.
-  scoped_ptr<notifier::TalkMediator> talk_mediator_;
-  // Needed by talk_mediator_.
-  chrome_common_net::FakeNetworkChangeNotifierThread
-      fake_network_change_notifier_thread_;
   scoped_refptr<AuthWatcher> auth_watcher_;
 
   // This is used to block the AuthWatcherThread when it raises events until we
diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index b35a549..54a7188 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -15,6 +15,7 @@
 #include "base/base64.h"
 #include "base/lock.h"
 #include "base/logging.h"
+#include "base/message_loop.h"
 #include "base/platform_thread.h"
 #include "base/scoped_ptr.h"
 #include "base/sha1.h"
@@ -82,6 +83,13 @@ typedef GoogleServiceAuthError AuthError;
 static const int kThreadExitTimeoutMsec = 60000;
 static const int kSSLPort = 443;
 
+// We manage the lifetime of sync_api::SyncManager::SyncInternal ourselves.
+template <>
+struct RunnableMethodTraits<sync_api::SyncManager::SyncInternal> {
+  void RetainCallee(sync_api::SyncManager::SyncInternal*) {}
+  void ReleaseCallee(sync_api::SyncManager::SyncInternal*) {}
+};
+
 namespace sync_api {
 
 static const FilePath::CharType kBookmarkSyncUserSettingsDatabase[] =
@@ -774,16 +782,21 @@ class SyncManager::SyncInternal
   static const int kPreferencesNudgeDelayMilliseconds;
  public:
   explicit SyncInternal(SyncManager* sync_manager)
-      : observer_(NULL),
+      : core_message_loop_(NULL),
+        observer_(NULL),
         auth_problem_(AuthError::NONE),
         sync_manager_(sync_manager),
         registrar_(NULL),
         notification_pending_(false),
         initialized_(false),
         notification_method_(browser_sync::kDefaultNotificationMethod) {
+    DCHECK(ChromeThread::CurrentlyOn(ChromeThread::UI));
   }
 
-  ~SyncInternal() { }
+  ~SyncInternal() {
+    DCHECK(!core_message_loop_);
+    DCHECK(ChromeThread::CurrentlyOn(ChromeThread::UI));
+  }
 
   bool Init(const FilePath& database_location,
             const std::string& sync_server_and_path,
@@ -844,6 +857,10 @@ class SyncManager::SyncInternal
   // on startup, to serve our UI needs.
   void HandleAuthWatcherEvent(const AuthWatcherEvent& event);
 
+  // Login to the talk mediator with the given credentials.
+  void TalkMediatorLogin(
+      const std::string& email, const std::string& token);
+
   // TalkMediator::Delegate implementation.
 
   virtual void OnNotificationStateChange(
@@ -948,7 +965,10 @@ class SyncManager::SyncInternal
   // already initialized, this is a no-op.
   void MarkAndNotifyInitializationComplete();
 
-  bool SendXMPPNotification();
+  // If there's a pending notification to be sent, either from the
+  // new_pending_notification flag or a previous unsuccessfully sent
+  // notification, tries to send a notification.
+  void SendPendingXMPPNotification(bool new_pending_notification);
 
   // Determine if the parents or predecessors differ between the old and new
   // versions of an entry stored in |a| and |b|.  Note that a node's index may
@@ -1010,6 +1030,8 @@ class SyncManager::SyncInternal
   // last user information, current sync-related URLs, and more.
   scoped_ptr<UserSettings> user_settings_;
 
+  MessageLoop* core_message_loop_;
+
   // Observer registered via SetObserver/RemoveObserver.
   // WARNING: This can be NULL!
   SyncManager::Observer* observer_;
@@ -1170,6 +1192,8 @@ bool SyncManager::SyncInternal::Init(
 
   LOG(INFO) << "Starting SyncInternal initialization.";
 
+  core_message_loop_ = MessageLoop::current();
+  DCHECK(core_message_loop_);
   notification_method_ = notification_method;
   // Set up UserSettings, creating the db if necessary. We need this to
   // instantiate a URLFactory to give to the Syncer.
@@ -1229,10 +1253,8 @@ bool SyncManager::SyncInternal::Init(
                                   service_id,
                                   gaia_url,
                                   user_settings_.get(),
-                                  gaia_auth,
-                                  talk_mediator());
+                                  gaia_auth);
 
-  allstatus_.WatchAuthWatcher(auth_watcher());
   authwatcher_hookup_.reset(NewEventListenerHookup(auth_watcher_->channel(),
       this, &SyncInternal::HandleAuthWatcherEvent));
 
@@ -1291,7 +1313,20 @@ void SyncManager::SyncInternal::MarkAndNotifyInitializationComplete() {
     observer_->OnInitializationComplete();
 }
 
-bool SyncManager::SyncInternal::SendXMPPNotification() {
+void SyncManager::SyncInternal::SendPendingXMPPNotification(
+    bool new_pending_notification) {
+  DCHECK_EQ(MessageLoop::current(), core_message_loop_);
+  notification_pending_ = notification_pending_ || new_pending_notification;
+  if (!notification_pending_) {
+    LOG(INFO) << "Not sending notification: no pending notification";
+    return;
+  }
+  if (!talk_mediator_.get()) {
+    LOG(INFO) << "Not sending notification: shutting down "
+              << "(talk_mediator_ is NULL)";
+    return;
+  }
+  LOG(INFO) << "Sending XMPP notification...";
   OutgoingNotificationData notification_data;
   if (notification_method_ == browser_sync::NOTIFICATION_LEGACY) {
     notification_data.service_id = browser_sync::kSyncLegacyServiceId;
@@ -1311,7 +1346,13 @@ bool SyncManager::SyncInternal::SendXMPPNotification() {
       notification_data.require_subscription = false;
     }
   }
-  return talk_mediator()->SendNotification(notification_data);
+  bool success = talk_mediator_->SendNotification(notification_data);
+  if (success) {
+    notification_pending_ = false;
+    LOG(INFO) << "Sent XMPP notification";
+  } else {
+    LOG(INFO) << "Could not send XMPP notification";
+  }
 }
 
 void SyncManager::SyncInternal::Authenticate(const std::string& username,
@@ -1389,11 +1430,15 @@ void SyncManager::SyncInternal::Shutdown() {
   if (auth_watcher_) {
     auth_watcher_->Shutdown();
     auth_watcher_ = NULL;
+    authwatcher_hookup_.reset();
   }
 
   if (syncer_thread()) {
-    if (!syncer_thread()->Stop(kThreadExitTimeoutMsec))
-      DCHECK(false) << "Unable to stop the syncer, it won't be happy...";
+    if (!syncer_thread()->Stop(kThreadExitTimeoutMsec)) {
+      LOG(FATAL) << "Unable to stop the syncer, it won't be happy...";
+    }
+    syncer_event_.reset();
+    syncer_thread_ = NULL;
   }
 
   // Shutdown the xmpp buzz connection.
@@ -1405,6 +1450,18 @@ void SyncManager::SyncInternal::Shutdown() {
     LOG(INFO) << "P2P: Mediator destroyed.";
   }
 
+  // Pump any messages the auth watcher, syncer thread, or talk
+  // mediator posted before they shut down. (See HandleSyncerEvent(),
+  // HandleAuthWatcherEvent(), and HandleTalkMediatorEvent() for the
+  // events that may be posted.)
+  {
+    CHECK(core_message_loop_);
+    bool old_state = core_message_loop_->NestableTasksAllowed();
+    core_message_loop_->SetNestableTasksAllowed(true);
+    core_message_loop_->RunAllPending();
+    core_message_loop_->SetNestableTasksAllowed(old_state);
+  }
+
   if (network_change_notifier_.get()) {
     network_change_notifier_->RemoveObserver(this);
     network_change_notifier_.reset();
@@ -1422,8 +1479,8 @@ void SyncManager::SyncInternal::Shutdown() {
 
   // We don't want to process any more events.
   dir_change_hookup_.reset();
-  syncer_event_.reset();
-  authwatcher_hookup_.reset();
+
+  core_message_loop_ = NULL;
 }
 
 void SyncManager::SyncInternal::OnIPAddressChanged() {
@@ -1635,33 +1692,14 @@ void SyncManager::SyncInternal::HandleSyncerEvent(const SyncerEvent& event) {
 
     // TODO(chron): Consider changing this back to track has_more_to_sync
     // only notify peers if a successful commit has occurred.
-    if (event.snapshot->syncer_status.num_successful_commits > 0) {
-      // We use a member variable here because talk may not have connected yet.
-      // The notification must be stored until it can be sent.
-      notification_pending_ = true;
-    }
-
-    // SyncCycles are started by the following events: creation of the syncer,
-    // (re)connection to buzz, local changes, peer notifications of updates.
-    // Peers will be notified of changes made while there is no buzz connection
-    // immediately after a connection has been re-established.
-    // the next sync cycle.
-    // TODO(brg): Move this to TalkMediatorImpl as a SyncerThread event hook.
-    if (notification_pending_ && talk_mediator()) {
-        LOG(INFO) << "Sending XMPP notification...";
-        bool success = SendXMPPNotification();
-        if (success) {
-          notification_pending_ = false;
-          LOG(INFO) << "Sent XMPP notification";
-        } else {
-          LOG(INFO) << "Could not send XMPP notification";
-        }
-    } else {
-      LOG(INFO) << "Didn't send XMPP notification!"
-                   << " event.snapshot.did_commit_items: "
-                   << event.snapshot->did_commit_items
-                   << " talk_mediator(): " << talk_mediator();
-    }
+    bool new_pending_notification =
+        (event.snapshot->syncer_status.num_successful_commits > 0);
+    core_message_loop_->PostTask(
+        FROM_HERE,
+        NewRunnableMethod(
+            this,
+            &SyncManager::SyncInternal::SendPendingXMPPNotification,
+            new_pending_notification));
   }
 
   if (event.what_happened == SyncerEvent::PAUSED) {
@@ -1677,6 +1715,7 @@ void SyncManager::SyncInternal::HandleSyncerEvent(const SyncerEvent& event) {
 
 void SyncManager::SyncInternal::HandleAuthWatcherEvent(
     const AuthWatcherEvent& event) {
+  allstatus_.HandleAuthWatcherEvent(event);
   // We don't care about an authentication attempt starting event, and we
   // don't want to reset our state to GoogleServiceAuthError::NONE because the
   // fact that an _attempt_ is starting doesn't change the fact that we have an
@@ -1688,6 +1727,7 @@ void SyncManager::SyncInternal::HandleAuthWatcherEvent(
   auth_problem_ = AuthError::NONE;
   switch (event.what_happened) {
     case AuthWatcherEvent::AUTH_SUCCEEDED:
+      DCHECK(!event.user_email.empty());
       // We now know the supplied username and password were valid. If this
       // wasn't the first sync, authenticated_name should already be assigned.
       if (username_for_share().empty()) {
@@ -1716,6 +1756,23 @@ void SyncManager::SyncInternal::HandleAuthWatcherEvent(
       }
       if (InitialSyncEndedForAllEnabledTypes())
         MarkAndNotifyInitializationComplete();
+
+      if (!event.auth_token.empty()) {
+        core_message_loop_->PostTask(
+            FROM_HERE,
+            NewRunnableMethod(
+                this, &SyncManager::SyncInternal::TalkMediatorLogin,
+                event.user_email, event.auth_token));
+      }
+      return;
+    case AuthWatcherEvent::AUTH_RENEWED:
+      DCHECK(!event.user_email.empty());
+      DCHECK(!event.auth_token.empty());
+      core_message_loop_->PostTask(
+          FROM_HERE,
+            NewRunnableMethod(
+                this, &SyncManager::SyncInternal::TalkMediatorLogin,
+                event.user_email, event.auth_token));
       return;
     // Authentication failures translate to GoogleServiceAuthError events.
     case AuthWatcherEvent::GAIA_AUTH_FAILED:     // Invalid GAIA credentials.
@@ -1765,8 +1822,29 @@ void SyncManager::SyncInternal::OnNotificationStateChange(
   if (notifications_enabled) {
     // Send a notification as soon as subscriptions are on
     // (see http://code.google.com/p/chromium/issues/detail?id=38563 ).
-    SendXMPPNotification();
+    core_message_loop_->PostTask(
+        FROM_HERE,
+        NewRunnableMethod(
+            this,
+            &SyncManager::SyncInternal::SendPendingXMPPNotification,
+            true));
+  }
+}
+
+void SyncManager::SyncInternal::TalkMediatorLogin(
+    const std::string& email, const std::string& token) {
+  DCHECK_EQ(MessageLoop::current(), core_message_loop_);
+  DCHECK(!email.empty());
+  DCHECK(!token.empty());
+  if (!talk_mediator_.get()) {
+    LOG(INFO) << "Not logging in: shutting down "
+              << "(talk_mediator_ is NULL)";
+    return;
   }
+  // TODO(akalin): Make talk_mediator automatically login on
+  // auth token change.
+  talk_mediator_->SetAuthToken(email, token, SYNC_SERVICE_NAME);
+  talk_mediator_->Login();
 }
 
 void SyncManager::SyncInternal::OnIncomingNotification(