Converted TalkMediator to use a delegate instead of a channel

Simplified handling of TalkMediator events.

BUG=none
TEST=unittests + manual testing of notifications

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

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

14 files changed, 238 insertions, 308 deletions

diff --git a/chrome/browser/sync/engine/all_status.cc b/chrome/browser/sync/engine/all_status.cc
index abd0e56..68d2513a 100644
--- a/chrome/browser/sync/engine/all_status.cc
+++ b/chrome/browser/sync/engine/all_status.cc
@@ -257,38 +257,6 @@ void AllStatus::HandleServerConnectionEvent(
   }
 }
 
-void AllStatus::WatchTalkMediator(const notifier::TalkMediator* mediator) {
-  status_.notifications_enabled = false;
-  talk_mediator_hookup_.reset(
-      NewEventListenerHookup(mediator->channel(), this,
-                             &AllStatus::HandleTalkMediatorEvent));
-}
-
-void AllStatus::HandleTalkMediatorEvent(
-    const notifier::TalkMediatorEvent& event) {
-  ScopedStatusLockWithNotify lock(this);
-  switch (event.what_happened) {
-    case notifier::TalkMediatorEvent::SUBSCRIPTIONS_ON:
-      status_.notifications_enabled = true;
-      break;
-    case notifier::TalkMediatorEvent::LOGOUT_SUCCEEDED:
-    case notifier::TalkMediatorEvent::SUBSCRIPTIONS_OFF:
-    case notifier::TalkMediatorEvent::TALKMEDIATOR_DESTROYED:
-      status_.notifications_enabled = false;
-      break;
-    case notifier::TalkMediatorEvent::NOTIFICATION_RECEIVED:
-      status_.notifications_received++;
-      break;
-    case notifier::TalkMediatorEvent::NOTIFICATION_SENT:
-      status_.notifications_sent++;
-      break;
-    case notifier::TalkMediatorEvent::LOGIN_SUCCEEDED:
-    default:
-      lock.set_notify_plan(DONT_NOTIFY);
-      break;
-  }
-}
-
 AllStatus::Status AllStatus::status() const {
   AutoLock lock(mutex_);
   return status_;
@@ -319,6 +287,21 @@ int AllStatus::GetRecommendedDelay(int base_delay_ms) const {
   return GetRecommendedDelaySeconds(base_delay_ms / 1000) * 1000;
 }
 
+void AllStatus::SetNotificationsEnabled(bool notifications_enabled) {
+  ScopedStatusLockWithNotify lock(this);
+  status_.notifications_enabled = notifications_enabled;
+}
+
+void AllStatus::IncrementNotificationsSent() {
+  ScopedStatusLockWithNotify lock(this);
+  ++status_.notifications_sent;
+}
+
+void AllStatus::IncrementNotificationsReceived() {
+  ScopedStatusLockWithNotify lock(this);
+  ++status_.notifications_received;
+}
+
 ScopedStatusLockWithNotify::ScopedStatusLockWithNotify(AllStatus* allstatus)
   : allstatus_(allstatus), plan_(NOTIFY_IF_STATUS_CHANGED) {
   event_.what_changed = 0;
diff --git a/chrome/browser/sync/engine/all_status.h b/chrome/browser/sync/engine/all_status.h
index 3e70119..f19d6ca 100644
--- a/chrome/browser/sync/engine/all_status.h
+++ b/chrome/browser/sync/engine/all_status.h
@@ -20,11 +20,6 @@ class GaiaAuthenticator;
 struct GaiaAuthEvent;
 }
 
-namespace notifier {
-class TalkMediator;
-struct TalkMediatorEvent;
-}
-
 namespace browser_sync {
 
 class AuthWatcher;
@@ -118,11 +113,6 @@ class AllStatus {
   void WatchSyncerThread(SyncerThread* syncer_thread);
   void HandleSyncerEvent(const SyncerEvent& event);
 
-  void WatchTalkMediator(
-      const notifier::TalkMediator* talk_mediator);
-  void HandleTalkMediatorEvent(
-      const notifier::TalkMediatorEvent& event);
-
   // Returns a string description of the SyncStatus (currently just the ascii
   // version of the enum). Will LOG(FATAL) if the status us out of range.
   static const char* GetSyncStatusString(SyncStatus status);
@@ -137,6 +127,12 @@ class AllStatus {
   // This uses AllStatus' max_consecutive_errors as the error count
   int GetRecommendedDelay(int base_delay) const;
 
+  void SetNotificationsEnabled(bool notifications_enabled);
+
+  void IncrementNotificationsSent();
+
+  void IncrementNotificationsReceived();
+
  protected:
   typedef std::map<Syncer*, EventListenerHookup*> Syncers;
 
diff --git a/chrome/browser/sync/engine/auth_watcher.cc b/chrome/browser/sync/engine/auth_watcher.cc
index e75b6de..e674ab3 100644
--- a/chrome/browser/sync/engine/auth_watcher.cc
+++ b/chrome/browser/sync/engine/auth_watcher.cc
@@ -40,7 +40,6 @@ namespace browser_sync {
 
 AuthWatcher::AuthWatcher(DirectoryManager* dirman,
                          ServerConnectionManager* scm,
-                         AllStatus* allstatus,
                          const string& user_agent,
                          const string& service_id,
                          const string& gaia_url,
@@ -50,7 +49,6 @@ AuthWatcher::AuthWatcher(DirectoryManager* dirman,
     : gaia_(gaia_auth),
       dirman_(dirman),
       scm_(scm),
-      allstatus_(allstatus),
       status_(NOT_AUTHENTICATED),
       user_settings_(user_settings),
       talk_mediator_(talk_mediator),
diff --git a/chrome/browser/sync/engine/auth_watcher.h b/chrome/browser/sync/engine/auth_watcher.h
index 5649136..60f3ed8 100644
--- a/chrome/browser/sync/engine/auth_watcher.h
+++ b/chrome/browser/sync/engine/auth_watcher.h
@@ -32,7 +32,6 @@ class DirectoryManager;
 
 namespace browser_sync {
 
-class AllStatus;
 class AuthWatcher;
 class ServerConnectionManager;
 class URLFactory;
@@ -89,7 +88,6 @@ class AuthWatcher : public base::RefCountedThreadSafe<AuthWatcher> {
 
   AuthWatcher(DirectoryManager* dirman,
               ServerConnectionManager* scm,
-              AllStatus* allstatus,
               const std::string& user_agent,
               const std::string& service_id,
               const std::string& gaia_url,
@@ -138,7 +136,6 @@ class AuthWatcher : public base::RefCountedThreadSafe<AuthWatcher> {
   std::string email() const;
   syncable::DirectoryManager* dirman() const { return dirman_; }
   ServerConnectionManager* scm() const { return scm_; }
-  AllStatus* allstatus() const { return allstatus_; }
   UserSettings* settings() const { return user_settings_; }
   Status status() const { return (Status)status_; }
 
@@ -207,7 +204,6 @@ class AuthWatcher : public base::RefCountedThreadSafe<AuthWatcher> {
   syncable::DirectoryManager* const dirman_;
   ServerConnectionManager* const scm_;
   scoped_ptr<EventListenerHookup> connmgr_hookup_;
-  AllStatus* const allstatus_;
   Status status_;
   UserSettings* const user_settings_;
   // Interface to the notifications engine.
diff --git a/chrome/browser/sync/engine/auth_watcher_unittest.cc b/chrome/browser/sync/engine/auth_watcher_unittest.cc
index 9a33e59..eb39772 100644
--- a/chrome/browser/sync/engine/auth_watcher_unittest.cc
+++ b/chrome/browser/sync/engine/auth_watcher_unittest.cc
@@ -93,7 +93,6 @@ class AuthWatcherTest : public testing::Test {
     // Mock out data that would normally be sent back from a server.
     connection()->SetAuthenticationResponseInfo(kValidAuthToken,
         kUserDisplayName, kUserDisplayEmail, "ID");
-    allstatus_.reset(new AllStatus());
     user_settings_.reset(new UserSettings());
     ASSERT_TRUE(temp_dir_.CreateUniqueTempDir());
     FilePath user_settings_path = temp_dir_.path().Append(kUserSettingsDB);
@@ -103,7 +102,7 @@ class AuthWatcherTest : public testing::Test {
     talk_mediator_.reset(new notifier::TalkMediatorImpl(
         &fake_network_change_notifier_thread_, false));
     auth_watcher_ = new AuthWatcher(metadb_.manager(), connection_.get(),
-        allstatus_.get(), kTestUserAgent, kTestServiceId, kTestGaiaURL,
+        kTestUserAgent, kTestServiceId, kTestGaiaURL,
         user_settings_.get(), gaia_auth_, talk_mediator_.get());
     authwatcher_hookup_.reset(NewEventListenerHookup(auth_watcher_->channel(),
         this, &AuthWatcherTest::HandleAuthWatcherEvent));
@@ -150,7 +149,6 @@ class AuthWatcherTest : public testing::Test {
   // The sync engine pieces necessary to run an AuthWatcher.
   TriggeredOpenTestDirectorySetterUpper metadb_;
   scoped_ptr<MockConnectionManager> connection_;
-  scoped_ptr<AllStatus> allstatus_;
   scoped_ptr<UserSettings> user_settings_;
   GaiaAuthMockForAuthWatcher* gaia_auth_;  // Owned by auth_watcher_.
   scoped_ptr<notifier::TalkMediator> talk_mediator_;
diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index e052bc6..b35a549 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -69,7 +69,6 @@ using browser_sync::UserSettings;
 using browser_sync::sessions::SyncSessionContext;
 using notifier::TalkMediator;
 using notifier::TalkMediatorImpl;
-using notifier::TalkMediatorEvent;
 using std::list;
 using std::hex;
 using std::string;
@@ -769,7 +768,8 @@ class BridgedGaiaAuthenticator : public gaia::GaiaAuthenticator {
 //////////////////////////////////////////////////////////////////////////
 // SyncManager's implementation: SyncManager::SyncInternal
 class SyncManager::SyncInternal
-    : public net::NetworkChangeNotifier::Observer {
+    : public net::NetworkChangeNotifier::Observer,
+      public TalkMediator::Delegate {
   static const int kDefaultNudgeDelayMilliseconds;
   static const int kPreferencesNudgeDelayMilliseconds;
  public:
@@ -844,9 +844,15 @@ class SyncManager::SyncInternal
   // on startup, to serve our UI needs.
   void HandleAuthWatcherEvent(const AuthWatcherEvent& event);
 
-  // We listen to TalkMediator events so that we can send an
-  // XMPP notification when subscriptions are on.
-  void HandleTalkMediatorEvent(const TalkMediatorEvent& event);
+  // TalkMediator::Delegate implementation.
+
+  virtual void OnNotificationStateChange(
+      bool notifications_enabled);
+
+  virtual void OnIncomingNotification(
+      const IncomingNotificationData& notification_data);
+
+  virtual void OnOutgoingNotification();
 
   // Accessors for the private members.
   DirectoryManager* dir_manager() { return share_.dir_manager.get(); }
@@ -856,7 +862,6 @@ class SyncManager::SyncInternal
   SyncerThread* syncer_thread() { return syncer_thread_.get(); }
   TalkMediator* talk_mediator() { return talk_mediator_.get(); }
   AuthWatcher* auth_watcher() { return auth_watcher_.get(); }
-  AllStatus* allstatus() { return &allstatus_; }
   void set_observer(SyncManager::Observer* observer) { observer_ = observer; }
   UserShare* GetUserShare() { return &share_; }
 
@@ -1019,9 +1024,6 @@ class SyncManager::SyncInternal
   // Notification (xmpp) handler.
   scoped_ptr<TalkMediator> talk_mediator_;
 
-  // XMPP notifications event handler
-  scoped_ptr<EventListenerHookup> talk_mediator_hookup_;
-
   // A multi-purpose status watch object that aggregates stats from various
   // sync components.
   AllStatus allstatus_;
@@ -1189,7 +1191,7 @@ bool SyncManager::SyncInternal::Init(
       post_factory));
 
   // Watch various objects for aggregated status.
-  allstatus()->WatchConnectionManager(connection_manager());
+  allstatus_.WatchConnectionManager(connection_manager());
 
   network_change_notifier_.reset(
       new chrome_common_net::NetworkChangeNotifierProxy(
@@ -1208,13 +1210,9 @@ bool SyncManager::SyncInternal::Init(
     }
     talk_mediator_->AddSubscribedServiceUrl(browser_sync::kSyncServiceUrl);
   }
-  allstatus()->WatchTalkMediator(talk_mediator());
 
   // Listen to TalkMediator events ourselves
-  talk_mediator_hookup_.reset(
-      NewEventListenerHookup(talk_mediator_->channel(),
-                             this,
-                             &SyncInternal::HandleTalkMediatorEvent));
+  talk_mediator_->SetDelegate(this);
 
   std::string gaia_url = gaia::kGaiaUrl;
   const char* service_id = gaia_service_id ?
@@ -1227,7 +1225,6 @@ bool SyncManager::SyncInternal::Init(
 
   auth_watcher_ = new AuthWatcher(dir_manager(),
                                   connection_manager(),
-                                  &allstatus_,
                                   gaia_source,
                                   service_id,
                                   gaia_url,
@@ -1235,7 +1232,7 @@ bool SyncManager::SyncInternal::Init(
                                   gaia_auth,
                                   talk_mediator());
 
-  allstatus()->WatchAuthWatcher(auth_watcher());
+  allstatus_.WatchAuthWatcher(auth_watcher());
   authwatcher_hookup_.reset(NewEventListenerHookup(auth_watcher_->channel(),
       this, &SyncInternal::HandleAuthWatcherEvent));
 
@@ -1246,8 +1243,7 @@ bool SyncManager::SyncInternal::Init(
 
   // The SyncerThread takes ownership of |context|.
   syncer_thread_ = new SyncerThread(context, &allstatus_);
-  syncer_thread()->WatchTalkMediator(talk_mediator());
-  allstatus()->WatchSyncerThread(syncer_thread());
+  allstatus_.WatchSyncerThread(syncer_thread());
 
   // Subscribe to the syncer thread's channel.
   syncer_event_.reset(
@@ -1323,7 +1319,7 @@ void SyncManager::SyncInternal::Authenticate(const std::string& username,
                                              const std::string& captcha) {
   DCHECK(username_for_share().empty() || username == username_for_share())
         << "Username change from valid username detected";
-  if (allstatus()->status().authenticated)
+  if (allstatus_.status().authenticated)
     return;
   if (password.empty()) {
     // TODO(timsteele): Seems like this shouldn't be needed, but auth_watcher
@@ -1572,7 +1568,7 @@ void SyncManager::SyncInternal::HandleCalculateChangesChangeEventFromSyncer(
 
 SyncManager::Status::Summary
 SyncManager::SyncInternal::ComputeAggregatedStatusSummary() {
-  switch (allstatus()->status().icon) {
+  switch (allstatus_.status().icon) {
     case AllStatus::OFFLINE:
       return Status::OFFLINE;
     case AllStatus::OFFLINE_UNSYNCED:
@@ -1593,23 +1589,23 @@ SyncManager::SyncInternal::ComputeAggregatedStatusSummary() {
 SyncManager::Status SyncManager::SyncInternal::ComputeAggregatedStatus() {
   Status return_status =
       { ComputeAggregatedStatusSummary(),
-        allstatus()->status().authenticated,
-        allstatus()->status().server_up,
-        allstatus()->status().server_reachable,
-        allstatus()->status().server_broken,
-        allstatus()->status().notifications_enabled,
-        allstatus()->status().notifications_received,
-        allstatus()->status().notifications_sent,
-        allstatus()->status().unsynced_count,
-        allstatus()->status().conflicting_count,
-        allstatus()->status().syncing,
-        allstatus()->status().initial_sync_ended,
-        allstatus()->status().syncer_stuck,
-        allstatus()->status().updates_available,
-        allstatus()->status().updates_received,
-        allstatus()->status().disk_full,
+        allstatus_.status().authenticated,
+        allstatus_.status().server_up,
+        allstatus_.status().server_reachable,
+        allstatus_.status().server_broken,
+        allstatus_.status().notifications_enabled,
+        allstatus_.status().notifications_received,
+        allstatus_.status().notifications_sent,
+        allstatus_.status().unsynced_count,
+        allstatus_.status().conflicting_count,
+        allstatus_.status().syncing,
+        allstatus_.status().initial_sync_ended,
+        allstatus_.status().syncer_stuck,
+        allstatus_.status().updates_available,
+        allstatus_.status().updates_received,
+        allstatus_.status().disk_full,
         false,   // TODO(ncarter): invalid store?
-        allstatus()->status().max_consecutive_errors};
+        allstatus_.status().max_consecutive_errors};
   return return_status;
 }
 
@@ -1758,18 +1754,45 @@ void SyncManager::SyncInternal::HandleAuthWatcherEvent(
     observer_->OnAuthError(AuthError(auth_problem_));
 }
 
-void SyncManager::SyncInternal::HandleTalkMediatorEvent(
-    const TalkMediatorEvent& event) {
-  // Send a notification as soon as subscriptions are on.
-  // This is to fix Bug 38563.
-  // See http://code.google.com/p/chromium/issues/detail?id=38563.
-  // This was originally fixed in http://codereview.chromium.org/1545024
-  // but it got moved here when the refactoring of TalkMediator happened.
-  if (event.what_happened == TalkMediatorEvent::SUBSCRIPTIONS_ON) {
+void SyncManager::SyncInternal::OnNotificationStateChange(
+    bool notifications_enabled) {
+  LOG(INFO) << "P2P: Notifications enabled = "
+            << (notifications_enabled ? "true" : "false");
+  allstatus_.SetNotificationsEnabled(notifications_enabled);
+  if (syncer_thread()) {
+    syncer_thread()->SetNotificationsEnabled(notifications_enabled);
+  }
+  if (notifications_enabled) {
+    // Send a notification as soon as subscriptions are on
+    // (see http://code.google.com/p/chromium/issues/detail?id=38563 ).
     SendXMPPNotification();
   }
 }
 
+void SyncManager::SyncInternal::OnIncomingNotification(
+    const IncomingNotificationData& notification_data) {
+  // Check if the service url is a sync URL. An empty service URL
+  // is treated as a legacy sync notification
+  if (notification_data.service_url.empty() ||
+      (notification_data.service_url ==
+       browser_sync::kSyncLegacyServiceUrl) ||
+      (notification_data.service_url ==
+       browser_sync::kSyncServiceUrl)) {
+    LOG(INFO) << "P2P: Updates on server, pushing syncer";
+    if (syncer_thread()) {
+      syncer_thread()->NudgeSyncer(0, SyncerThread::kNotification);
+    }
+    allstatus_.IncrementNotificationsReceived();
+  } else {
+    LOG(WARNING) << "Notification fron unexpected source: "
+                 << notification_data.service_url;
+  }
+}
+
+void SyncManager::SyncInternal::OnOutgoingNotification() {
+  allstatus_.IncrementNotificationsSent();
+}
+
 SyncManager::Status::Summary SyncManager::GetStatusSummary() const {
   return data_->ComputeAggregatedStatusSummary();
 }
diff --git a/chrome/browser/sync/engine/syncer_thread.cc b/chrome/browser/sync/engine/syncer_thread.cc
index c543f6a..ed8b023 100644
--- a/chrome/browser/sync/engine/syncer_thread.cc
+++ b/chrome/browser/sync/engine/syncer_thread.cc
@@ -16,7 +16,6 @@
 #include <queue>
 
 #include "base/third_party/dynamic_annotations/dynamic_annotations.h"
-#include "chrome/browser/sync/sync_constants.h"
 #include "chrome/browser/sync/engine/auth_watcher.h"
 #include "chrome/browser/sync/engine/model_safe_worker.h"
 #include "chrome/browser/sync/engine/net/server_connection_manager.h"
@@ -24,8 +23,6 @@
 #include "chrome/browser/sync/syncable/directory_manager.h"
 #include "chrome/common/chrome_switches.h"
 #include "chrome/common/net/notifier/listener/notification_constants.h"
-#include "chrome/common/net/notifier/listener/talk_mediator.h"
-#include "chrome/common/net/notifier/listener/talk_mediator_impl.h"
 
 using std::priority_queue;
 using std::min;
@@ -72,7 +69,6 @@ SyncerThread::SyncerThread(sessions::SyncSessionContext* context,
       syncer_long_poll_interval_seconds_(kDefaultLongPollIntervalSeconds),
       syncer_polling_interval_(kDefaultShortPollIntervalSeconds),
       syncer_max_interval_(kDefaultMaxPollIntervalMs),
-      talk_mediator_hookup_(NULL),
       directory_manager_hookup_(NULL),
       syncer_events_(NULL),
       session_context_(context),
@@ -98,7 +94,6 @@ SyncerThread::~SyncerThread() {
   directory_manager_hookup_.reset();
   syncer_events_.reset();
   delete vault_.syncer_;
-  talk_mediator_hookup_.reset();
   CHECK(!thread_.IsRunning());
 }
 
@@ -602,56 +597,9 @@ void SyncerThread::NudgeSyncImpl(int milliseconds_from_now,
   vault_field_changed_.Broadcast();
 }
 
-void SyncerThread::WatchTalkMediator(notifier::TalkMediator* mediator) {
-  talk_mediator_hookup_.reset(
-      NewEventListenerHookup(
-          mediator->channel(),
-          this,
-          &SyncerThread::HandleTalkMediatorEvent));
-}
-
-void SyncerThread::HandleTalkMediatorEvent(
-    const notifier::TalkMediatorEvent& event) {
+void SyncerThread::SetNotificationsEnabled(bool notifications_enabled) {
   AutoLock lock(lock_);
-  switch (event.what_happened) {
-    case notifier::TalkMediatorEvent::LOGIN_SUCCEEDED:
-      LOG(INFO) << "P2P: Login succeeded.";
-      p2p_authenticated_ = true;
-      break;
-    case notifier::TalkMediatorEvent::LOGOUT_SUCCEEDED:
-      LOG(INFO) << "P2P: Login succeeded.";
-      p2p_authenticated_ = false;
-      break;
-    case notifier::TalkMediatorEvent::SUBSCRIPTIONS_ON:
-      LOG(INFO) << "P2P: Subscriptions successfully enabled.";
-      p2p_subscribed_ = true;
-      if (NULL != vault_.syncer_) {
-        LOG(INFO) << "Subscriptions on.  Nudging syncer for initial push.";
-        NudgeSyncImpl(0, kLocal);
-      }
-      break;
-    case notifier::TalkMediatorEvent::SUBSCRIPTIONS_OFF:
-      LOG(INFO) << "P2P: Subscriptions are not enabled.";
-      p2p_subscribed_ = false;
-      break;
-    case notifier::TalkMediatorEvent::NOTIFICATION_RECEIVED:
-      // Check if the service url is a sync URL. An empty service URL
-      // is treated as a legacy sync notification
-      if (event.notification_data.service_url.empty() ||
-          (event.notification_data.service_url == kSyncLegacyServiceUrl) ||
-          (event.notification_data.service_url == kSyncServiceUrl)) {
-        LOG(INFO) << "P2P: Updates on server, pushing syncer";
-        if (NULL != vault_.syncer_) {
-          NudgeSyncImpl(0, kNotification);
-        }
-      }
-      break;
-    default:
-      break;
-  }
-
-  session_context_->set_notifications_enabled(p2p_authenticated_ &&
-                                              p2p_subscribed_);
+  session_context_->set_notifications_enabled(notifications_enabled);
 }
 
 // Returns the amount of time since the user last interacted with the computer,
diff --git a/chrome/browser/sync/engine/syncer_thread.h b/chrome/browser/sync/engine/syncer_thread.h
index ac1ea7b..5a34b97 100644
--- a/chrome/browser/sync/engine/syncer_thread.h
+++ b/chrome/browser/sync/engine/syncer_thread.h
@@ -30,10 +30,6 @@
 
 class EventListenerHookup;
 
-namespace notifier {
-class TalkMediator;
-}
-
 namespace syncable {
 class DirectoryManager;
 struct DirectoryManagerEvent;
@@ -48,7 +44,6 @@ class URLFactory;
 struct ServerConnectionEvent;
 struct SyncerEvent;
 struct SyncerShutdownEvent;
-struct TalkMediatorEvent;
 
 class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
                      public sessions::SyncSession::Delegate {
@@ -134,8 +129,7 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
   // from the SyncerThread's controller and will cause a mutex lock.
   virtual void NudgeSyncer(int milliseconds_from_now, NudgeSource source);
 
-  // Registers this thread to watch talk mediator events.
-  virtual void WatchTalkMediator(notifier::TalkMediator* talk_mediator);
+  void SetNotificationsEnabled(bool notifications_enabled);
 
   virtual SyncerEventChannel* relay_channel();
 
@@ -238,8 +232,6 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
 
   void HandleServerConnectionEvent(const ServerConnectionEvent& event);
 
-  void HandleTalkMediatorEvent(const notifier::TalkMediatorEvent& event);
-
   void SyncMain(Syncer* syncer);
 
   // Calculates the next sync wait time and exponential backoff state.
@@ -309,7 +301,6 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
   // this is called.
   void NudgeSyncImpl(int milliseconds_from_now, NudgeSource source);
 
-  scoped_ptr<EventListenerHookup> talk_mediator_hookup_;
   scoped_ptr<EventListenerHookup> directory_manager_hookup_;
   scoped_ptr<EventListenerHookup> syncer_events_;
 
diff --git a/chrome/common/net/notifier/DEPS b/chrome/common/net/notifier/DEPS
index 95cce2f..2d7d13f 100644
--- a/chrome/common/net/notifier/DEPS
+++ b/chrome/common/net/notifier/DEPS
@@ -3,6 +3,4 @@ include_rules = [
   "+talk/base",
   "+talk/xmpp",
   "+talk/xmllite",
-  "+talk/p2p/base",  # TODO(ncarter): Determine if this is necessary/proper.
-
 ]
diff --git a/chrome/common/net/notifier/listener/talk_mediator.h b/chrome/common/net/notifier/listener/talk_mediator.h
index 5580691..c842b1b 100644
--- a/chrome/common/net/notifier/listener/talk_mediator.h
+++ b/chrome/common/net/notifier/listener/talk_mediator.h
@@ -19,39 +19,29 @@
 #include <string>
 
 #include "chrome/common/net/notifier/listener/notification_defines.h"
-#include "chrome/common/deprecated/event_sys.h"
 
 namespace notifier {
 
-struct TalkMediatorEvent {
-  enum WhatHappened {
-    LOGIN_SUCCEEDED,
-    LOGOUT_SUCCEEDED,
-    SUBSCRIPTIONS_ON,
-    SUBSCRIPTIONS_OFF,
-    NOTIFICATION_RECEIVED,
-    NOTIFICATION_SENT,
-    TALKMEDIATOR_DESTROYED,
-  };
+class TalkMediator {
+ public:
+  TalkMediator() {}
+  virtual ~TalkMediator() {}
 
-  // Required by EventChannel.
-  typedef TalkMediatorEvent EventType;
+  class Delegate {
+   public:
+    virtual ~Delegate() {}
 
-  static inline bool IsChannelShutdownEvent(const TalkMediatorEvent& event) {
-    return event.what_happened == TALKMEDIATOR_DESTROYED;
-  }
+    virtual void OnNotificationStateChange(
+        bool notifications_enabled) = 0;
 
-  WhatHappened what_happened;
-  // Data in the case of a NOTIFICATION_RECEIVED event
-  IncomingNotificationData notification_data;
-};
+    virtual void OnIncomingNotification(
+        const IncomingNotificationData& notification_data) = 0;
 
-typedef EventChannel<TalkMediatorEvent, Lock> TalkMediatorChannel;
+    virtual void OnOutgoingNotification() = 0;
+  };
 
-class TalkMediator {
- public:
-  TalkMediator() {}
-  virtual ~TalkMediator() {}
+  // |delegate| may be NULL.
+  virtual void SetDelegate(Delegate* delegate) = 0;
 
   // The following methods are for authorizaiton of the xmpp client.
   virtual bool SetAuthToken(const std::string& email,
@@ -64,9 +54,6 @@ class TalkMediator {
   // occurred.
   virtual bool SendNotification(const OutgoingNotificationData& data) = 0;
 
-  // Channel by which talk mediator events are signaled.
-  virtual TalkMediatorChannel* channel() const = 0;
-
   // Add a URL to subscribe to for notifications.
   virtual void AddSubscribedServiceUrl(const std::string& service_url) = 0;
 };
diff --git a/chrome/common/net/notifier/listener/talk_mediator_impl.cc b/chrome/common/net/notifier/listener/talk_mediator_impl.cc
index f2e3d2e..ff5c62c 100644
--- a/chrome/common/net/notifier/listener/talk_mediator_impl.cc
+++ b/chrome/common/net/notifier/listener/talk_mediator_impl.cc
@@ -7,7 +7,6 @@
 #include "base/logging.h"
 #include "base/singleton.h"
 #include "chrome/common/net/notifier/listener/mediator_thread_impl.h"
-#include "chrome/common/deprecated/event_sys-inl.h"
 #include "talk/base/cryptstring.h"
 #include "talk/base/ssladapter.h"
 #include "talk/xmpp/xmppclientsettings.h"
@@ -43,7 +42,8 @@ TalkMediatorImpl::TalkMediatorImpl(
       chrome_common_net::NetworkChangeNotifierThread*
           network_change_notifier_thread,
       bool invalidate_xmpp_auth_token)
-    : mediator_thread_(
+    : delegate_(NULL),
+      mediator_thread_(
         new MediatorThreadImpl(network_change_notifier_thread)),
       invalidate_xmpp_auth_token_(invalidate_xmpp_auth_token) {
   // Ensure the SSL library is initialized.
@@ -54,15 +54,14 @@ TalkMediatorImpl::TalkMediatorImpl(
 }
 
 TalkMediatorImpl::TalkMediatorImpl(MediatorThread *thread)
-    : mediator_thread_(thread),
+    : delegate_(NULL),
+      mediator_thread_(thread),
       invalidate_xmpp_auth_token_(false) {
   // When testing we do not initialize the SSL library.
   TalkMediatorInitialization(true);
 }
 
 void TalkMediatorImpl::TalkMediatorInitialization(bool should_connect) {
-  TalkMediatorEvent done = { TalkMediatorEvent::TALKMEDIATOR_DESTROYED };
-  channel_.reset(new TalkMediatorChannel(done));
   if (should_connect) {
     mediator_thread_->SignalStateChange.connect(
         this,
@@ -130,8 +129,9 @@ bool TalkMediatorImpl::SendNotification(const OutgoingNotificationData& data) {
   return false;
 }
 
-TalkMediatorChannel* TalkMediatorImpl::channel() const {
-  return channel_.get();
+void TalkMediatorImpl::SetDelegate(Delegate* delegate) {
+  AutoLock lock(mutex_);
+  delegate_ = delegate;
 }
 
 bool TalkMediatorImpl::SetAuthToken(const std::string& email,
@@ -197,9 +197,9 @@ void TalkMediatorImpl::MediatorThreadNotificationHandler(
     const IncomingNotificationData& notification_data) {
   LOG(INFO) << "P2P: Updates are available on the server.";
   AutoLock lock(mutex_);
-  TalkMediatorEvent event = { TalkMediatorEvent::NOTIFICATION_RECEIVED };
-  event.notification_data = notification_data;
-  channel_->NotifyListeners(event);
+  if (delegate_) {
+    delegate_->OnIncomingNotification(notification_data);
+  }
 }
 
 
@@ -213,8 +213,6 @@ void TalkMediatorImpl::OnLogin() {
   mediator_thread_->ListenForUpdates();
   // Now subscribe for updates to all the services we are interested in
   mediator_thread_->SubscribeForUpdates(subscribed_services_list_);
-  TalkMediatorEvent event = { TalkMediatorEvent::LOGIN_SUCCEEDED };
-  channel_->NotifyListeners(event);
 }
 
 void TalkMediatorImpl::OnLogout() {
@@ -223,8 +221,6 @@ void TalkMediatorImpl::OnLogout() {
   AutoLock lock(mutex_);
   state_.logging_in = 0;
   state_.logged_in = 0;
-  TalkMediatorEvent event = { TalkMediatorEvent::LOGOUT_SUCCEEDED };
-  channel_->NotifyListeners(event);
 }
 
 void TalkMediatorImpl::OnSubscriptionSuccess() {
@@ -238,24 +234,27 @@ void TalkMediatorImpl::OnSubscriptionSuccess() {
   // Notifying listeners with a lock held can cause the lock to be
   // recursively taken if the listener decides to call back into us
   // in the event handler.
-  TalkMediatorEvent event = { TalkMediatorEvent::SUBSCRIPTIONS_ON };
-  channel_->NotifyListeners(event);
+  if (delegate_) {
+    delegate_->OnNotificationStateChange(true);
+  }
 }
 
 void TalkMediatorImpl::OnSubscriptionFailure() {
   LOG(INFO) << "P2P: Update subscription failure.";
   AutoLock lock(mutex_);
   state_.subscribed = 0;
-  TalkMediatorEvent event = { TalkMediatorEvent::SUBSCRIPTIONS_OFF };
-  channel_->NotifyListeners(event);
+  if (delegate_) {
+    delegate_->OnNotificationStateChange(false);
+  }
 }
 
 void TalkMediatorImpl::OnNotificationSent() {
   LOG(INFO) <<
       "P2P: Peers were notified that updates are available on the server.";
   AutoLock lock(mutex_);
-  TalkMediatorEvent event = { TalkMediatorEvent::NOTIFICATION_SENT };
-  channel_->NotifyListeners(event);
+  if (delegate_) {
+    delegate_->OnOutgoingNotification();
+  }
 }
 
 }  // namespace notifier
diff --git a/chrome/common/net/notifier/listener/talk_mediator_impl.h b/chrome/common/net/notifier/listener/talk_mediator_impl.h
index cb965b7..de28fc5 100644
--- a/chrome/common/net/notifier/listener/talk_mediator_impl.h
+++ b/chrome/common/net/notifier/listener/talk_mediator_impl.h
@@ -20,8 +20,6 @@
 #include "talk/xmpp/xmppclientsettings.h"
 #include "testing/gtest/include/gtest/gtest_prod.h"  // For FRIEND_TEST
 
-class EventListenerHookup;
-
 namespace chrome_common_net {
 class NetworkChangeNotifierThread;
 }  // namespace chrome_common_net
@@ -40,6 +38,9 @@ class TalkMediatorImpl
   virtual ~TalkMediatorImpl();
 
   // Overriden from TalkMediator.
+
+  virtual void SetDelegate(Delegate* delegate);
+
   virtual bool SetAuthToken(const std::string& email,
                             const std::string& token,
                             const std::string& token_service);
@@ -48,8 +49,6 @@ class TalkMediatorImpl
 
   virtual bool SendNotification(const OutgoingNotificationData& data);
 
-  TalkMediatorChannel* channel() const;
-
   virtual void AddSubscribedServiceUrl(const std::string& service_url);
 
  private:
@@ -91,21 +90,18 @@ class TalkMediatorImpl
   // this mutex.
   Lock mutex_;
 
+  // Delegate.  May be NULL.
+  Delegate* delegate_;
+
   // Internal state.
   TalkMediatorState state_;
 
   // Cached and verfied from the SetAuthToken method.
   buzz::XmppClientSettings xmpp_settings_;
 
-  // Interface to listen to authentication events.
-  scoped_ptr<EventListenerHookup> auth_hookup_;
-
   // The worker thread through which talk events are posted and received.
   scoped_ptr<MediatorThread> mediator_thread_;
 
-  // Channel through which to broadcast events.
-  scoped_ptr<TalkMediatorChannel> channel_;
-
   bool invalidate_xmpp_auth_token_;
 
   std::vector<std::string> subscribed_services_list_;
diff --git a/chrome/common/net/notifier/listener/talk_mediator_unittest.cc b/chrome/common/net/notifier/listener/talk_mediator_unittest.cc
index 3bdac80..621e688 100644
--- a/chrome/common/net/notifier/listener/talk_mediator_unittest.cc
+++ b/chrome/common/net/notifier/listener/talk_mediator_unittest.cc
@@ -4,81 +4,88 @@
 
 #include <string>
 
+#include "base/basictypes.h"
 #include "base/logging.h"
 #include "chrome/common/net/fake_network_change_notifier_thread.h"
 #include "chrome/common/net/notifier/listener/mediator_thread_mock.h"
 #include "chrome/common/net/notifier/listener/talk_mediator_impl.h"
-#include "chrome/common/deprecated/event_sys-inl.h"
 #include "talk/xmpp/xmppengine.h"
+#include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace notifier {
 
-class TalkMediatorImplTest : public testing::Test {
+using ::testing::_;
+
+class MockTalkMediatorDelegate : public TalkMediator::Delegate {
  public:
-  void HandleTalkMediatorEvent(
-      const notifier::TalkMediatorEvent& event) {
-    last_message_ = event.what_happened;
-  }
+  MockTalkMediatorDelegate() {}
+  virtual ~MockTalkMediatorDelegate() {}
 
- protected:
-  TalkMediatorImplTest() {}
-  ~TalkMediatorImplTest() {}
+  MOCK_METHOD1(OnNotificationStateChange,
+               void(bool notification_changed));
+  MOCK_METHOD1(OnIncomingNotification,
+               void(const IncomingNotificationData& data));
+  MOCK_METHOD0(OnOutgoingNotification, void());
 
-  virtual void SetUp() {
-    last_message_ = -1;
-  }
+ private:
+  DISALLOW_COPY_AND_ASSIGN(MockTalkMediatorDelegate);
+};
 
-  virtual void TearDown() {
-  }
+class TalkMediatorImplTest : public testing::Test {
+ protected:
+  TalkMediatorImplTest() {}
+  virtual ~TalkMediatorImplTest() {}
 
   chrome_common_net::FakeNetworkChangeNotifierThread
       fake_network_change_notifier_thread_;
   int last_message_;
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(TalkMediatorImplTest);
 };
 
 TEST_F(TalkMediatorImplTest, ConstructionOfTheClass) {
   // Constructing a single talk mediator enables SSL through the singleton.
   scoped_ptr<TalkMediatorImpl> talk1(
       new TalkMediatorImpl(&fake_network_change_notifier_thread_, false));
-  talk1.reset(NULL);
 }
 
 TEST_F(TalkMediatorImplTest, SetAuthTokenWithBadInput) {
   scoped_ptr<TalkMediatorImpl> talk1(new TalkMediatorImpl(
       new MockMediatorThread()));
-  ASSERT_FALSE(talk1->SetAuthToken("@missinguser.com", "", "fake_service"));
-  ASSERT_TRUE(talk1->state_.initialized == 0);
+  EXPECT_FALSE(talk1->SetAuthToken("@missinguser.com", "", "fake_service"));
+  EXPECT_FALSE(talk1->state_.initialized);
 
   scoped_ptr<TalkMediatorImpl> talk2(new TalkMediatorImpl(
       new MockMediatorThread()));
-  ASSERT_FALSE(talk2->SetAuthToken("", "1234567890", "fake_service"));
-  ASSERT_TRUE(talk2->state_.initialized == 0);
+  EXPECT_FALSE(talk2->SetAuthToken("", "1234567890", "fake_service"));
+  EXPECT_FALSE(talk2->state_.initialized);
 
   scoped_ptr<TalkMediatorImpl> talk3(new TalkMediatorImpl(
       new MockMediatorThread()));
-  ASSERT_FALSE(talk3->SetAuthToken("missingdomain", "abcde",  "fake_service"));
-  ASSERT_TRUE(talk3->state_.initialized == 0);
+  EXPECT_FALSE(talk3->SetAuthToken("missingdomain", "abcde",  "fake_service"));
+  EXPECT_FALSE(talk3->state_.initialized);
 }
 
 TEST_F(TalkMediatorImplTest, SetAuthTokenWithGoodInput) {
   scoped_ptr<TalkMediatorImpl> talk1(new TalkMediatorImpl(
       new MockMediatorThread()));
-  ASSERT_TRUE(talk1->SetAuthToken("chromium@gmail.com", "token",
+  EXPECT_TRUE(talk1->SetAuthToken("chromium@gmail.com", "token",
                                   "fake_service"));
-  ASSERT_TRUE(talk1->state_.initialized == 1);
+  EXPECT_TRUE(talk1->state_.initialized);
 
   scoped_ptr<TalkMediatorImpl> talk2(new TalkMediatorImpl(
       new MockMediatorThread()));
-  ASSERT_TRUE(talk2->SetAuthToken("chromium@mail.google.com", "token",
+  EXPECT_TRUE(talk2->SetAuthToken("chromium@mail.google.com", "token",
                                   "fake_service"));
-  ASSERT_TRUE(talk2->state_.initialized == 1);
+  EXPECT_TRUE(talk2->state_.initialized);
 
   scoped_ptr<TalkMediatorImpl> talk3(new TalkMediatorImpl(
       new MockMediatorThread()));
-  ASSERT_TRUE(talk3->SetAuthToken("chromium@chromium.org", "token",
+  EXPECT_TRUE(talk3->SetAuthToken("chromium@chromium.org", "token",
                                   "fake_service"));
-  ASSERT_TRUE(talk3->state_.initialized == 1);
+  EXPECT_TRUE(talk3->state_.initialized);
 }
 
 TEST_F(TalkMediatorImplTest, LoginWiring) {
@@ -87,25 +94,25 @@ TEST_F(TalkMediatorImplTest, LoginWiring) {
   scoped_ptr<TalkMediatorImpl> talk1(new TalkMediatorImpl(mock));
 
   // Login checks states for initialization.
-  ASSERT_TRUE(talk1->Login() == false);
-  ASSERT_TRUE(mock->login_calls == 0);
+  EXPECT_FALSE(talk1->Login());
+  EXPECT_EQ(0, mock->login_calls);
 
-  ASSERT_TRUE(talk1->SetAuthToken("chromium@gmail.com", "token",
-                                  "fake_service") == true);
-  ASSERT_TRUE(talk1->Login() == true);
-  ASSERT_TRUE(mock->login_calls == 1);
+  EXPECT_TRUE(talk1->SetAuthToken("chromium@gmail.com", "token",
+                                  "fake_service"));
+  EXPECT_TRUE(talk1->Login());
+  EXPECT_EQ(1, mock->login_calls);
 
   // Successive calls to login will fail.  One needs to create a new talk
   // mediator object.
-  ASSERT_TRUE(talk1->Login() == false);
-  ASSERT_TRUE(mock->login_calls == 1);
+  EXPECT_FALSE(talk1->Login());
+  EXPECT_EQ(1, mock->login_calls);
 
-  ASSERT_TRUE(talk1->Logout() == true);
-  ASSERT_TRUE(mock->logout_calls == 1);
+  EXPECT_TRUE(talk1->Logout());
+  EXPECT_EQ(1, mock->logout_calls);
 
   // Successive logout calls do nothing.
-  ASSERT_TRUE(talk1->Logout() == false);
-  ASSERT_TRUE(mock->logout_calls == 1);
+  EXPECT_FALSE(talk1->Logout());
+  EXPECT_EQ(1, mock->logout_calls);
 }
 
 TEST_F(TalkMediatorImplTest, SendNotification) {
@@ -115,33 +122,33 @@ TEST_F(TalkMediatorImplTest, SendNotification) {
 
   // Failure due to not being logged in.
   OutgoingNotificationData data;
-  ASSERT_TRUE(talk1->SendNotification(data) == false);
-  ASSERT_TRUE(mock->send_calls == 0);
+  EXPECT_FALSE(talk1->SendNotification(data));
+  EXPECT_EQ(0, mock->send_calls);
 
-  ASSERT_TRUE(talk1->SetAuthToken("chromium@gmail.com", "token",
-                                  "fake_service") == true);
-  ASSERT_TRUE(talk1->Login() == true);
+  EXPECT_TRUE(talk1->SetAuthToken("chromium@gmail.com", "token",
+                                  "fake_service"));
+  EXPECT_TRUE(talk1->Login());
   talk1->OnLogin();
-  ASSERT_TRUE(mock->login_calls == 1);
+  EXPECT_EQ(1, mock->login_calls);
 
   // Failure due to not being subscribed.
-  ASSERT_TRUE(talk1->SendNotification(data) == false);
-  ASSERT_TRUE(mock->send_calls == 0);
+  EXPECT_FALSE(talk1->SendNotification(data));
+  EXPECT_EQ(0, mock->send_calls);
 
   // Fake subscription
   talk1->OnSubscriptionSuccess();
-  ASSERT_TRUE(talk1->state_.subscribed == 1);
-  ASSERT_TRUE(talk1->SendNotification(data) == true);
-  ASSERT_TRUE(mock->send_calls == 1);
-  ASSERT_TRUE(talk1->SendNotification(data) == true);
-  ASSERT_TRUE(mock->send_calls == 2);
+  EXPECT_TRUE(talk1->state_.subscribed);
+  EXPECT_TRUE(talk1->SendNotification(data));
+  EXPECT_EQ(1, mock->send_calls);
+  EXPECT_TRUE(talk1->SendNotification(data));
+  EXPECT_EQ(2, mock->send_calls);
 
-  ASSERT_TRUE(talk1->Logout() == true);
-  ASSERT_TRUE(mock->logout_calls == 1);
+  EXPECT_TRUE(talk1->Logout());
+  EXPECT_EQ(1, mock->logout_calls);
 
   // Failure due to being logged out.
-  ASSERT_TRUE(talk1->SendNotification(data) == false);
-  ASSERT_TRUE(mock->send_calls == 2);
+  EXPECT_FALSE(talk1->SendNotification(data));
+  EXPECT_EQ(2, mock->send_calls);
 }
 
 TEST_F(TalkMediatorImplTest, MediatorThreadCallbacks) {
@@ -149,41 +156,44 @@ TEST_F(TalkMediatorImplTest, MediatorThreadCallbacks) {
   MockMediatorThread* mock = new MockMediatorThread();
   scoped_ptr<TalkMediatorImpl> talk1(new TalkMediatorImpl(mock));
 
-  scoped_ptr<EventListenerHookup> callback(NewEventListenerHookup(
-      talk1->channel(), this, &TalkMediatorImplTest::HandleTalkMediatorEvent));
+  MockTalkMediatorDelegate mock_delegate;
+  EXPECT_CALL(mock_delegate, OnNotificationStateChange(true));
+  EXPECT_CALL(mock_delegate, OnIncomingNotification(_));
+  EXPECT_CALL(mock_delegate, OnOutgoingNotification());
+
+  talk1->SetDelegate(&mock_delegate);
 
-  ASSERT_TRUE(talk1->SetAuthToken("chromium@gmail.com", "token",
-                                  "fake_service") == true);
-  ASSERT_TRUE(talk1->Login() == true);
-  ASSERT_TRUE(mock->login_calls == 1);
+  EXPECT_TRUE(talk1->SetAuthToken("chromium@gmail.com", "token",
+                                  "fake_service"));
+  EXPECT_TRUE(talk1->Login());
+  EXPECT_EQ(1, mock->login_calls);
 
   mock->ChangeState(MediatorThread::MSG_LOGGED_IN);
-  ASSERT_TRUE(last_message_ == TalkMediatorEvent::LOGIN_SUCCEEDED);
 
   // The message triggers calls to listen and subscribe.
-  ASSERT_TRUE(mock->listen_calls == 1);
-  ASSERT_TRUE(mock->subscribe_calls == 1);
-  ASSERT_TRUE(talk1->state_.subscribed == 0);
+  EXPECT_EQ(1, mock->listen_calls);
+  EXPECT_EQ(1, mock->subscribe_calls);
+  EXPECT_FALSE(talk1->state_.subscribed);
 
   mock->ChangeState(MediatorThread::MSG_SUBSCRIPTION_SUCCESS);
-  ASSERT_TRUE(last_message_ == TalkMediatorEvent::SUBSCRIPTIONS_ON);
-  ASSERT_TRUE(talk1->state_.subscribed == 1);
+  EXPECT_TRUE(talk1->state_.subscribed);
 
   // After subscription success is receieved, the talk mediator will allow
   // sending of notifications.
   OutgoingNotificationData outgoing_data;
-  ASSERT_TRUE(talk1->SendNotification(outgoing_data) == true);
-  ASSERT_TRUE(mock->send_calls == 1);
+  EXPECT_TRUE(talk1->SendNotification(outgoing_data));
+  // TODO(akalin): Fix locking issues issues and move this into
+  // MediatorThreadMock.
+  mock->ChangeState(MediatorThread::MSG_NOTIFICATION_SENT);
+  EXPECT_EQ(1, mock->send_calls);
 
   IncomingNotificationData incoming_data;
   incoming_data.service_url = "service_url";
   incoming_data.service_specific_data = "service_data";
   mock->Notify(incoming_data);
-  ASSERT_TRUE(last_message_ == TalkMediatorEvent::NOTIFICATION_RECEIVED);
 
-  // A |TALKMEDIATOR_DESTROYED| message is received during tear down.
   talk1.reset();
-  ASSERT_TRUE(last_message_ == TalkMediatorEvent::TALKMEDIATOR_DESTROYED);
+  EXPECT_EQ(1, mock->logout_calls);
 }
 
 }  // namespace notifier
diff --git a/chrome/service/cloud_print/cloud_print_proxy_backend.cc b/chrome/service/cloud_print/cloud_print_proxy_backend.cc
index 4a8da05..cad6f38 100644
--- a/chrome/service/cloud_print/cloud_print_proxy_backend.cc
+++ b/chrome/service/cloud_print/cloud_print_proxy_backend.cc
@@ -12,7 +12,6 @@
 #include "chrome/service/cloud_print/cloud_print_consts.h"
 #include "chrome/service/cloud_print/cloud_print_helpers.h"
 #include "chrome/service/cloud_print/printer_job_handler.h"
-#include "chrome/common/deprecated/event_sys-inl.h"
 #include "chrome/common/net/notifier/listener/talk_mediator_impl.h"
 #include "chrome/service/gaia/service_gaia_authenticator.h"
 #include "chrome/service/net/service_network_change_notifier_thread.h"
@@ -26,7 +25,8 @@ class CloudPrintProxyBackend::Core
     : public base::RefCountedThreadSafe<CloudPrintProxyBackend::Core>,
       public URLFetcherDelegate,
       public cloud_print::PrinterChangeNotifierDelegate,
-      public PrinterJobHandlerDelegate {
+      public PrinterJobHandlerDelegate,
+      public notifier::TalkMediator::Delegate {
  public:
   explicit Core(CloudPrintProxyBackend* backend,
                 const GURL& cloud_print_server_url);
@@ -71,6 +71,15 @@ class CloudPrintProxyBackend::Core
   void OnPrinterJobHandlerShutdown(PrinterJobHandler* job_handler,
                                    const std::string& printer_id);
 
+  // notifier::TalkMediator::Delegate implementation.
+  virtual void OnNotificationStateChange(
+      bool notifications_enabled);
+
+  virtual void OnIncomingNotification(
+      const IncomingNotificationData& notification_data);
+
+  virtual void OnOutgoingNotification();
+
  protected:
   // Prototype for a response handler.
   typedef void (CloudPrintProxyBackend::Core::*ResponseHandler)(
@@ -117,7 +126,6 @@ class CloudPrintProxyBackend::Core
   // handler is responsible for checking for pending print jobs for this
   // printer and print them.
   void InitJobHandlerForPrinter(DictionaryValue* printer_data);
-  void HandleTalkMediatorEvent(const notifier::TalkMediatorEvent& event);
 
   // Our parent CloudPrintProxyBackend
   CloudPrintProxyBackend* backend_;
@@ -155,7 +163,6 @@ class CloudPrintProxyBackend::Core
   bool new_printers_available_;
   // Notification (xmpp) handler.
   scoped_ptr<notifier::TalkMediator> talk_mediator_;
-  scoped_ptr<EventListenerHookup> talk_mediator_hookup_;
 
   DISALLOW_COPY_AND_ASSIGN(Core);
 };
@@ -277,10 +284,7 @@ void CloudPrintProxyBackend::Core::DoInitializeWithToken(
   talk_mediator_.reset(new notifier::TalkMediatorImpl(
       g_service_process->network_change_notifier_thread(), false));
   talk_mediator_->AddSubscribedServiceUrl(kCloudPrintTalkServiceUrl);
-  talk_mediator_hookup_.reset(
-      NewEventListenerHookup(
-          talk_mediator_->channel(), this,
-          &CloudPrintProxyBackend::Core::HandleTalkMediatorEvent));
+  talk_mediator_->SetDelegate(this);
   talk_mediator_->SetAuthToken(email, cloud_print_xmpp_token,
                                kSyncGaiaServiceId);
   talk_mediator_->Login();
@@ -580,20 +584,23 @@ bool CloudPrintProxyBackend::Core::RemovePrinterFromList(
   return ret;
 }
 
-void CloudPrintProxyBackend::Core::HandleTalkMediatorEvent(
-    const notifier::TalkMediatorEvent& event) {
-  if ((event.what_happened ==
-      notifier::TalkMediatorEvent::NOTIFICATION_RECEIVED) &&
-      (0 == base::strcasecmp(kCloudPrintTalkServiceUrl,
-                             event.notification_data.service_url.c_str()))) {
+void CloudPrintProxyBackend::Core::OnNotificationStateChange(
+    bool notification_enabled) {}
+
+void CloudPrintProxyBackend::Core::OnIncomingNotification(
+    const IncomingNotificationData& notification_data) {
+  if (0 == base::strcasecmp(kCloudPrintTalkServiceUrl,
+                             notification_data.service_url.c_str())) {
     backend_->core_thread_.message_loop()->PostTask(
         FROM_HERE,
         NewRunnableMethod(
             this, &CloudPrintProxyBackend::Core::DoHandlePrinterNotification,
-            event.notification_data.service_specific_data));
+            notification_data.service_specific_data));
   }
 }
 
+void CloudPrintProxyBackend::Core::OnOutgoingNotification() {}
+
 // cloud_print::PrinterChangeNotifier::Delegate implementation
 void CloudPrintProxyBackend::Core::OnPrinterAdded() {
   if (request_.get()) {