[Sync] Replace TalkMediator*/MediatorThread* with PushClient

Streamline methods of PushClient and its Observer subclass.

Remove sync/protocol/service_constants.h and consolidate use of each
constant in one place.  Remove duplicate constant in cloud print code.

BUG=76764
TEST=
TBR=estade@chromium.org

Review URL: https://chromiumcodereview.appspot.com/10398051

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

16 files changed, 530 insertions, 1197 deletions

diff --git a/jingle/jingle.gyp b/jingle/jingle.gyp
index 02a5a05..9f9af8b 100644
--- a/jingle/jingle.gyp
+++ b/jingle/jingle.gyp
@@ -75,9 +75,8 @@
         'notifier/communicator/login_settings.h',
         'notifier/communicator/single_login_attempt.cc',
         'notifier/communicator/single_login_attempt.h',
-        'notifier/listener/mediator_thread.h',
-        'notifier/listener/mediator_thread_impl.cc',
-        'notifier/listener/mediator_thread_impl.h',
+        'notifier/listener/push_client.cc',
+        'notifier/listener/push_client.h',
         'notifier/listener/notification_constants.cc',
         'notifier/listener/notification_constants.h',
         'notifier/listener/notification_defines.cc',
@@ -88,9 +87,6 @@
         'notifier/listener/push_notifications_send_update_task.h',
         'notifier/listener/push_notifications_subscribe_task.cc',
         'notifier/listener/push_notifications_subscribe_task.h',
-        'notifier/listener/talk_mediator.h',
-        'notifier/listener/talk_mediator_impl.cc',
-        'notifier/listener/talk_mediator_impl.h',
         'notifier/listener/xml_element_util.cc',
         'notifier/listener/xml_element_util.h',
       ],
@@ -164,12 +160,9 @@
         'notifier/communicator/connection_settings_unittest.cc',
         'notifier/communicator/login_settings_unittest.cc',
         'notifier/communicator/single_login_attempt_unittest.cc',
-        'notifier/listener/mediator_thread_mock.cc',
-        'notifier/listener/mediator_thread_mock.h',
-        'notifier/listener/mediator_thread_unittest.cc',
+        'notifier/listener/push_client_unittest.cc',
         'notifier/listener/push_notifications_send_update_task_unittest.cc',
         'notifier/listener/push_notifications_subscribe_task_unittest.cc',
-        'notifier/listener/talk_mediator_unittest.cc',
         'notifier/listener/xml_element_util_unittest.cc',
         'run_all_unittests.cc',
       ],
diff --git a/jingle/notifier/base/notifier_options_util.cc b/jingle/notifier/base/notifier_options_util.cc
index 9e107bc..67620a0 100644
--- a/jingle/notifier/base/notifier_options_util.cc
+++ b/jingle/notifier/base/notifier_options_util.cc
@@ -13,8 +13,7 @@ namespace notifier {
 
 buzz::XmppClientSettings MakeXmppClientSettings(
     const NotifierOptions& notifier_options,
-    const std::string& email, const std::string& token,
-    const std::string& token_service) {
+    const std::string& email, const std::string& token) {
   buzz::Jid jid = buzz::Jid(email);
   DCHECK(!jid.node().empty());
   DCHECK(jid.IsValid());
@@ -27,7 +26,7 @@ buzz::XmppClientSettings MakeXmppClientSettings(
   xmpp_client_settings.set_auth_cookie(
       notifier_options.invalidate_xmpp_login ?
       token + "bogus" : token);
-  xmpp_client_settings.set_token_service(token_service);
+  xmpp_client_settings.set_token_service("chromiumsync");
   if (notifier_options.allow_insecure_connection) {
     xmpp_client_settings.set_allow_plain(true);
     xmpp_client_settings.set_use_tls(buzz::TLS_DISABLED);
diff --git a/jingle/notifier/base/notifier_options_util.h b/jingle/notifier/base/notifier_options_util.h
index 4278329..6ba2471 100644
--- a/jingle/notifier/base/notifier_options_util.h
+++ b/jingle/notifier/base/notifier_options_util.h
@@ -1,4 +1,4 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 //
@@ -19,8 +19,7 @@ struct NotifierOptions;
 
 buzz::XmppClientSettings MakeXmppClientSettings(
     const NotifierOptions& notifier_options,
-    const std::string& email, const std::string& token,
-    const std::string& token_service);
+    const std::string& email, const std::string& token);
 
 ServerList GetServerList(const NotifierOptions& notifier_options);
 
diff --git a/jingle/notifier/listener/mediator_thread.h b/jingle/notifier/listener/mediator_thread.h
deleted file mode 100644
index 157639e..0000000
--- a/jingle/notifier/listener/mediator_thread.h
+++ /dev/null
@@ -1,63 +0,0 @@
-// Copyright (c) 2009 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-//
-// These methods should post messages to a queue which a different thread will
-// later come back and read from.
-
-#ifndef JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_H_
-#define JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_H_
-
-#include <string>
-#include <vector>
-
-#include "jingle/notifier/listener/notification_defines.h"
-
-namespace buzz {
-class XmppClientSettings;
-}  // namespace buzz
-
-namespace notifier {
-
-class MediatorThread {
- public:
-  virtual ~MediatorThread() {}
-
-  class Observer {
-   public:
-    virtual ~Observer() {}
-
-    virtual void OnConnectionStateChange(bool logged_in) = 0;
-
-    virtual void OnSubscriptionStateChange(bool subscribed) = 0;
-
-    virtual void OnIncomingNotification(const Notification& notification) = 0;
-
-    virtual void OnOutgoingNotification() = 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;
-  virtual void SubscribeForUpdates(const SubscriptionList& subscriptions) = 0;
-  virtual void ListenForUpdates() = 0;
-  virtual void SendNotification(const Notification& data) = 0;
-  // UpdateXmppSettings is used to update the user information (typically the
-  // auth token) AFTER a call to Login and BEFORE a call to Logout(). This will
-  // not cause an immediate reconnect. The updated settings will be used the
-  // next time we attempt a reconnect because of a broken connection.
-  // The typical use-case for this is to update OAuth2 access tokens which
-  // expire every hour.
-  virtual void UpdateXmppSettings(const buzz::XmppClientSettings& settings) = 0;
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_H_
diff --git a/jingle/notifier/listener/mediator_thread_impl.cc b/jingle/notifier/listener/mediator_thread_impl.cc
deleted file mode 100644
index ea6d18d..0000000
--- a/jingle/notifier/listener/mediator_thread_impl.cc
+++ /dev/null
@@ -1,299 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "jingle/notifier/listener/mediator_thread_impl.h"
-
-#include "base/bind.h"
-#include "base/logging.h"
-#include "base/memory/scoped_ptr.h"
-#include "base/memory/weak_ptr.h"
-#include "base/message_loop.h"
-#include "base/observer_list_threadsafe.h"
-#include "base/threading/thread.h"
-#include "jingle/notifier/base/const_communicator.h"
-#include "jingle/notifier/base/notifier_options_util.h"
-#include "jingle/notifier/base/task_pump.h"
-#include "jingle/notifier/communicator/login.h"
-#include "jingle/notifier/listener/push_notifications_listen_task.h"
-#include "jingle/notifier/listener/push_notifications_send_update_task.h"
-#include "jingle/notifier/listener/push_notifications_subscribe_task.h"
-#include "net/base/host_port_pair.h"
-#include "net/url_request/url_request_context.h"
-#include "net/url_request/url_request_context_getter.h"
-#include "talk/xmpp/xmppclientsettings.h"
-
-namespace notifier {
-
-class MediatorThreadImpl::Core
-    : public base::RefCountedThreadSafe<MediatorThreadImpl::Core>,
-      public LoginDelegate,
-      public PushNotificationsListenTaskDelegate,
-      public PushNotificationsSubscribeTaskDelegate {
- public:
-  // Invoked on the caller thread.
-  explicit Core(const NotifierOptions& notifier_options);
-  void AddObserver(Observer* observer);
-  void RemoveObserver(Observer* observer);
-
-  // Login::Delegate implementation. Called on I/O thread.
-  virtual void OnConnect(
-      base::WeakPtr<buzz::XmppTaskParentInterface> base_task);
-  virtual void OnDisconnect();
-
-  // PushNotificationsListenTaskDelegate implementation. Called on I/O thread.
-  virtual void OnNotificationReceived(
-      const Notification& notification);
-  // PushNotificationsSubscribeTaskDelegate implementation. Called on I/O
-  // thread.
-  virtual void OnSubscribed();
-  virtual void OnSubscriptionError();
-
-  // Helpers invoked on I/O thread.
-  void Login(const buzz::XmppClientSettings& settings);
-  void Disconnect();
-  void ListenForPushNotifications();
-  void SubscribeForPushNotifications(
-      const SubscriptionList& subscriptions);
-  void SendNotification(const Notification& data);
-  void UpdateXmppSettings(const buzz::XmppClientSettings& settings);
-
- private:
-  friend class base::RefCountedThreadSafe<MediatorThreadImpl::Core>;
-  // Invoked on either the caller thread or the I/O thread.
-  virtual ~Core();
-  scoped_refptr<ObserverListThreadSafe<Observer> > observers_;
-  base::WeakPtr<buzz::XmppTaskParentInterface> base_task_;
-
-  const NotifierOptions notifier_options_;
-
-  scoped_ptr<notifier::Login> login_;
-
-  std::vector<Notification> pending_notifications_to_send_;
-
-  DISALLOW_COPY_AND_ASSIGN(Core);
-};
-
-MediatorThreadImpl::Core::Core(
-    const NotifierOptions& notifier_options)
-    : observers_(new ObserverListThreadSafe<Observer>()),
-      notifier_options_(notifier_options) {
-  DCHECK(notifier_options_.request_context_getter);
-}
-
-MediatorThreadImpl::Core::~Core() {
-}
-
-void MediatorThreadImpl::Core::AddObserver(Observer* observer) {
-  observers_->AddObserver(observer);
-}
-
-void MediatorThreadImpl::Core::RemoveObserver(Observer* observer) {
-  observers_->RemoveObserver(observer);
-}
-
-void MediatorThreadImpl::Core::Login(const buzz::XmppClientSettings& settings) {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  VLOG(1) << "P2P: Thread logging into talk network.";
-
-  base_task_.reset();
-  login_.reset(new notifier::Login(this,
-                                   settings,
-                                   notifier_options_.request_context_getter,
-                                   GetServerList(notifier_options_),
-                                   notifier_options_.try_ssltcp_first,
-                                   notifier_options_.auth_mechanism));
-  login_->StartConnection();
-}
-
-void MediatorThreadImpl::Core::Disconnect() {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  VLOG(1) << "P2P: Thread logging out of talk network.";
-  login_.reset();
-  base_task_.reset();
-}
-
-void MediatorThreadImpl::Core::ListenForPushNotifications() {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  if (!base_task_.get())
-    return;
-  PushNotificationsListenTask* listener =
-      new PushNotificationsListenTask(base_task_, this);
-  listener->Start();
-}
-
-void MediatorThreadImpl::Core::SubscribeForPushNotifications(
-    const SubscriptionList& subscriptions) {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  if (!base_task_.get())
-    return;
-  PushNotificationsSubscribeTask* subscribe_task =
-      new PushNotificationsSubscribeTask(base_task_, subscriptions, this);
-  subscribe_task->Start();
-}
-
-void MediatorThreadImpl::Core::OnSubscribed() {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  observers_->Notify(&Observer::OnSubscriptionStateChange, true);
-}
-
-void MediatorThreadImpl::Core::OnSubscriptionError() {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  observers_->Notify(&Observer::OnSubscriptionStateChange, false);
-}
-
-void MediatorThreadImpl::Core::OnNotificationReceived(
-    const Notification& notification) {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  observers_->Notify(&Observer::OnIncomingNotification, notification);
-}
-
-void MediatorThreadImpl::Core::SendNotification(const Notification& data) {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  if (!base_task_.get()) {
-    VLOG(1) << "P2P: Cannot send notification " << data.ToString()
-            << "; sending later";
-    pending_notifications_to_send_.push_back(data);
-    return;
-  }
-  // Owned by |base_task_|.
-  PushNotificationsSendUpdateTask* task =
-      new PushNotificationsSendUpdateTask(base_task_, data);
-  task->Start();
-  observers_->Notify(&Observer::OnOutgoingNotification);
-}
-
-void MediatorThreadImpl::Core::UpdateXmppSettings(
-    const buzz::XmppClientSettings& settings) {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  VLOG(1) << "P2P: Thread Updating login settings.";
-  // The caller should only call UpdateXmppSettings after a Login call.
-  if (login_.get())
-    login_->UpdateXmppSettings(settings);
-  else
-    NOTREACHED() <<
-        "P2P: Thread UpdateXmppSettings called when login_ was NULL";
-}
-
-void MediatorThreadImpl::Core::OnConnect(
-    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  base_task_ = base_task;
-  observers_->Notify(&Observer::OnConnectionStateChange, true);
-  std::vector<Notification> notifications_to_send;
-  notifications_to_send.swap(pending_notifications_to_send_);
-  for (std::vector<Notification>::const_iterator it =
-           notifications_to_send.begin();
-       it != notifications_to_send.end(); ++it) {
-    VLOG(1) << "P2P: Sending pending notification " << it->ToString();
-    SendNotification(*it);
-  }
-}
-
-void MediatorThreadImpl::Core::OnDisconnect() {
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
-  base_task_.reset();
-  observers_->Notify(&Observer::OnConnectionStateChange, false);
-}
-
-
-MediatorThreadImpl::MediatorThreadImpl(const NotifierOptions& notifier_options)
-    : core_(new Core(notifier_options)),
-      parent_message_loop_proxy_(
-          base::MessageLoopProxy::current()),
-      io_message_loop_proxy_(
-          notifier_options.request_context_getter->GetIOMessageLoopProxy()) {
-}
-
-MediatorThreadImpl::~MediatorThreadImpl() {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  LogoutImpl();
-}
-
-void MediatorThreadImpl::AddObserver(Observer* observer) {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  core_->AddObserver(observer);
-}
-
-void MediatorThreadImpl::RemoveObserver(Observer* observer) {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  core_->RemoveObserver(observer);
-}
-
-void MediatorThreadImpl::Start() {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-}
-
-void MediatorThreadImpl::Login(const buzz::XmppClientSettings& settings) {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&MediatorThreadImpl::Core::Login, core_.get(), settings));
-}
-
-void MediatorThreadImpl::Logout() {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  LogoutImpl();
-}
-
-void MediatorThreadImpl::ListenForUpdates() {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&MediatorThreadImpl::Core::ListenForPushNotifications,
-                 core_.get()));
-}
-
-void MediatorThreadImpl::SubscribeForUpdates(
-    const SubscriptionList& subscriptions) {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&MediatorThreadImpl::Core::SubscribeForPushNotifications,
-                 core_.get(), subscriptions));
-}
-
-void MediatorThreadImpl::SendNotification(
-    const Notification& data) {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&MediatorThreadImpl::Core::SendNotification, core_.get(),
-                 data));
-}
-
-void MediatorThreadImpl::UpdateXmppSettings(
-    const buzz::XmppClientSettings& settings) {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&MediatorThreadImpl::Core::UpdateXmppSettings, core_.get(),
-                 settings));
-}
-
-void MediatorThreadImpl::TriggerOnConnectForTest(
-    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
-    DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&MediatorThreadImpl::Core::OnConnect, core_.get(), base_task));
-}
-
-void MediatorThreadImpl::LogoutImpl() {
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&MediatorThreadImpl::Core::Disconnect, core_.get()));
-}
-
-}  // namespace notifier
diff --git a/jingle/notifier/listener/mediator_thread_impl.h b/jingle/notifier/listener/mediator_thread_impl.h
deleted file mode 100644
index bf86933..0000000
--- a/jingle/notifier/listener/mediator_thread_impl.h
+++ /dev/null
@@ -1,91 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-//
-// This object runs on a thread and knows how to interpret messages sent by the
-// talk mediator. The mediator posts messages to a queue which the thread polls
-// (in a super class).
-//
-// Example usage:
-//
-//   MediatorThread m = new MediatorThreadImpl(pass in stuff);
-//   m.start(); // Start the thread
-//   // Once the thread is started, you can do server stuff.
-//   m.Login(loginInformation);
-//   // Events happen, the mediator finds out through its pump more messages
-//   // are dispatched to the thread eventually we want to log out.
-//   m.Logout();
-//   delete m; // Also stops the thread.
-
-#ifndef JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_IMPL_H_
-#define JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_IMPL_H_
-
-#include <string>
-#include <vector>
-
-#include "base/basictypes.h"
-#include "base/compiler_specific.h"
-#include "base/memory/ref_counted.h"
-#include "base/memory/weak_ptr.h"
-#include "jingle/notifier/base/notifier_options.h"
-#include "jingle/notifier/listener/mediator_thread.h"
-
-namespace base {
-class MessageLoopProxy;
-}
-
-namespace buzz {
-class XmppClientSettings;
-class XmppTaskParentInterface;
-}  // namespace buzz
-
-namespace talk_base {
-class Task;
-}  // namespace talk_base
-
-namespace notifier {
-
-class MediatorThreadImpl : public MediatorThread {
- public:
-  explicit MediatorThreadImpl(const NotifierOptions& notifier_options);
-  virtual ~MediatorThreadImpl();
-
-  virtual void AddObserver(Observer* observer) OVERRIDE;
-  virtual void RemoveObserver(Observer* observer) OVERRIDE;
-
-  // Start the thread.
-  virtual void Start() OVERRIDE;
-
-  // These are called from outside threads, by the talk mediator object.
-  // They add messages to a queue which we poll in this thread.
-  virtual void Login(const buzz::XmppClientSettings& settings) OVERRIDE;
-  virtual void Logout() OVERRIDE;
-  virtual void ListenForUpdates() OVERRIDE;
-  virtual void SubscribeForUpdates(
-      const SubscriptionList& subscriptions) OVERRIDE;
-  virtual void SendNotification(const Notification& data) OVERRIDE;
-  virtual void UpdateXmppSettings(
-      const buzz::XmppClientSettings& settings) OVERRIDE;
-
-  // Used by unit tests.  Make sure that tests that use this have the
-  // IO message loop proxy passed in via |notifier_options| pointing
-  // to the current thread.
-  void TriggerOnConnectForTest(
-      base::WeakPtr<buzz::XmppTaskParentInterface> base_task);
-
- private:
-  // The logic of Logout without the thread check so it can be called in the
-  // d'tor.
-  void LogoutImpl();
-  // The real guts of MediatorThreadImpl, which allows this class to not be
-  // refcounted.
-  class Core;
-  scoped_refptr<Core> core_;
-  scoped_refptr<base::MessageLoopProxy> parent_message_loop_proxy_;
-  scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy_;
-  DISALLOW_COPY_AND_ASSIGN(MediatorThreadImpl);
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_IMPL_H_
diff --git a/jingle/notifier/listener/mediator_thread_mock.cc b/jingle/notifier/listener/mediator_thread_mock.cc
deleted file mode 100644
index 980f81a2..0000000
--- a/jingle/notifier/listener/mediator_thread_mock.cc
+++ /dev/null
@@ -1,85 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "jingle/notifier/listener/mediator_thread_mock.h"
-
-#include "talk/xmpp/xmppclientsettings.h"
-
-namespace notifier {
-
-MockMediatorThread::MockMediatorThread() : observer_(NULL) {
-  Reset();
-}
-
-MockMediatorThread::~MockMediatorThread() {}
-
-void MockMediatorThread::Reset() {
-  login_calls = 0;
-  logout_calls = 0;
-  start_calls = 0;
-  subscribe_calls = 0;
-  listen_calls = 0;
-  send_calls = 0;
-  update_settings_calls = 0;
-}
-
-void MockMediatorThread::AddObserver(Observer* observer) {
-  observer_ = observer;
-}
-
-void MockMediatorThread::RemoveObserver(Observer* observer) {
-  observer_ = NULL;
-}
-
-// Overridden from MediatorThread
-void MockMediatorThread::Login(const buzz::XmppClientSettings& settings) {
-  login_calls++;
-  if (observer_) {
-    observer_->OnConnectionStateChange(true);
-  }
-}
-
-void MockMediatorThread::Logout() {
-  logout_calls++;
-  if (observer_) {
-    observer_->OnConnectionStateChange(false);
-  }
-}
-
-void MockMediatorThread::Start() {
-  start_calls++;
-}
-
-void MockMediatorThread::SubscribeForUpdates(
-    const SubscriptionList& subscriptions) {
-  subscribe_calls++;
-  if (observer_) {
-    observer_->OnSubscriptionStateChange(true);
-  }
-}
-
-void MockMediatorThread::ListenForUpdates() {
-  listen_calls++;
-}
-
-void MockMediatorThread::SendNotification(const Notification &) {
-  send_calls++;
-  if (observer_) {
-    observer_->OnOutgoingNotification();
-  }
-}
-
-void MockMediatorThread::ReceiveNotification(
-    const Notification& data) {
-  if (observer_) {
-    observer_->OnIncomingNotification(data);
-  }
-}
-
-void MockMediatorThread::UpdateXmppSettings(
-    const buzz::XmppClientSettings& settings) {
-  update_settings_calls++;
-}
-
-}  // namespace notifier
diff --git a/jingle/notifier/listener/mediator_thread_mock.h b/jingle/notifier/listener/mediator_thread_mock.h
deleted file mode 100644
index 44b2043..0000000
--- a/jingle/notifier/listener/mediator_thread_mock.h
+++ /dev/null
@@ -1,68 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-// This is mock for delicious testing.
-// It's very primitive, and it would have been better to use gmock, except
-// that gmock is only for linux.
-
-#ifndef JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_MOCK_H_
-#define JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_MOCK_H_
-
-#include <string>
-#include <vector>
-
-#include "base/compiler_specific.h"
-#include "jingle/notifier/listener/mediator_thread.h"
-
-namespace buzz {
-class XmppClientSettings;
-}
-
-namespace notifier {
-
-class MockMediatorThread : public MediatorThread {
- public:
-  MockMediatorThread();
-
-  virtual ~MockMediatorThread();
-
-  void Reset();
-
-  virtual void AddObserver(Observer* observer) OVERRIDE;
-
-  virtual void RemoveObserver(Observer* observer) OVERRIDE;
-
-  // Overridden from MediatorThread
-  virtual void Login(const buzz::XmppClientSettings& settings) OVERRIDE;
-
-  virtual void Logout() OVERRIDE;
-
-  virtual void Start() OVERRIDE;
-
-  virtual void SubscribeForUpdates(
-      const SubscriptionList& subscriptions) OVERRIDE;
-
-  virtual void ListenForUpdates() OVERRIDE;
-
-  virtual void SendNotification(const Notification &) OVERRIDE;
-  virtual void UpdateXmppSettings(
-      const buzz::XmppClientSettings& settings) OVERRIDE;
-
-
-  void ReceiveNotification(const Notification& data);
-
-  Observer* observer_;
-  // Internal State
-  int login_calls;
-  int logout_calls;
-  int start_calls;
-  int subscribe_calls;
-  int listen_calls;
-  int send_calls;
-  int update_settings_calls;
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_MEDIATOR_THREAD_MOCK_H_
diff --git a/jingle/notifier/listener/mediator_thread_unittest.cc b/jingle/notifier/listener/mediator_thread_unittest.cc
deleted file mode 100644
index afb8ce1..0000000
--- a/jingle/notifier/listener/mediator_thread_unittest.cc
+++ /dev/null
@@ -1,98 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "jingle/notifier/listener/mediator_thread_impl.h"
-
-#include "base/basictypes.h"
-#include "base/memory/scoped_ptr.h"
-#include "base/message_loop.h"
-#include "jingle/notifier/base/fake_base_task.h"
-#include "jingle/notifier/base/notifier_options.h"
-#include "jingle/notifier/listener/notification_defines.h"
-#include "net/base/capturing_net_log.h"
-#include "net/url_request/url_request_context_getter.h"
-#include "net/url_request/url_request_test_util.h"
-#include "testing/gmock/include/gmock/gmock.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-namespace notifier {
-
-namespace {
-
-using ::testing::StrictMock;
-
-class MockObserver : public MediatorThread::Observer {
- public:
-  MOCK_METHOD1(OnConnectionStateChange, void(bool));
-  MOCK_METHOD1(OnSubscriptionStateChange, void(bool));
-  MOCK_METHOD1(OnIncomingNotification, void(const Notification&));
-  MOCK_METHOD0(OnOutgoingNotification, void());
-};
-
-}  // namespace
-
-class MediatorThreadTest : public testing::Test {
- protected:
-  MediatorThreadTest() {
-    notifier_options_.request_context_getter =
-        new TestURLRequestContextGetter(message_loop_.message_loop_proxy());
-  }
-
-  virtual ~MediatorThreadTest() {}
-
-  virtual void SetUp() {
-    mediator_thread_.reset(new MediatorThreadImpl(notifier_options_));
-    mediator_thread_->AddObserver(&mock_observer_);
-  }
-
-  virtual void TearDown() {
-    // Clear out any messages posted by MediatorThread's
-    // destructor.
-    message_loop_.RunAllPending();
-    mediator_thread_->RemoveObserver(&mock_observer_);
-    mediator_thread_.reset();
-  }
-
-  MessageLoop message_loop_;
-  NotifierOptions notifier_options_;
-  StrictMock<MockObserver> mock_observer_;
-  scoped_ptr<MediatorThreadImpl> mediator_thread_;
-  FakeBaseTask fake_base_task_;
-};
-
-TEST_F(MediatorThreadTest, SendNotificationBasic) {
-  EXPECT_CALL(mock_observer_, OnConnectionStateChange(true));
-  EXPECT_CALL(mock_observer_, OnOutgoingNotification());
-
-  mediator_thread_->TriggerOnConnectForTest(fake_base_task_.AsWeakPtr());
-  mediator_thread_->SendNotification(Notification());
-  mediator_thread_->Logout();
-
-  // Shouldn't trigger.
-  mediator_thread_->SendNotification(Notification());
-}
-
-TEST_F(MediatorThreadTest, SendNotificationDelayed) {
-  EXPECT_CALL(mock_observer_, OnConnectionStateChange(true));
-  EXPECT_CALL(mock_observer_, OnOutgoingNotification()).Times(5);
-
-  for (int i = 0; i < 5; ++i) {
-    mediator_thread_->SendNotification(Notification());
-  }
-  mediator_thread_->TriggerOnConnectForTest(fake_base_task_.AsWeakPtr());
-}
-
-TEST_F(MediatorThreadTest, SendNotificationDelayedTwice) {
-  EXPECT_CALL(mock_observer_, OnConnectionStateChange(true)).Times(2);
-  EXPECT_CALL(mock_observer_, OnOutgoingNotification()).Times(5);
-
-  for (int i = 0; i < 5; ++i) {
-    mediator_thread_->SendNotification(Notification());
-  }
-  mediator_thread_->TriggerOnConnectForTest(
-      base::WeakPtr<buzz::XmppTaskParentInterface>());
-  mediator_thread_->TriggerOnConnectForTest(fake_base_task_.AsWeakPtr());
-}
-
-}  // namespace notifier
diff --git a/jingle/notifier/listener/push_client.cc b/jingle/notifier/listener/push_client.cc
new file mode 100644
index 0000000..8c82b44
--- /dev/null
+++ b/jingle/notifier/listener/push_client.cc
@@ -0,0 +1,320 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "jingle/notifier/listener/push_client.h"
+
+#include "base/bind.h"
+#include "base/compiler_specific.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/observer_list_threadsafe.h"
+#include "jingle/notifier/base/notifier_options_util.h"
+#include "jingle/notifier/communicator/login.h"
+#include "jingle/notifier/listener/push_notifications_listen_task.h"
+#include "jingle/notifier/listener/push_notifications_send_update_task.h"
+#include "jingle/notifier/listener/push_notifications_subscribe_task.h"
+#include "talk/xmpp/xmppclientsettings.h"
+
+namespace notifier {
+
+PushClient::Observer::~Observer() {}
+
+// All member functions except for the constructor, destructor, and
+// {Add,Remove}Observer() must be called on the IO thread (as taken from
+// |notifier_options|).
+class PushClient::Core
+    : public base::RefCountedThreadSafe<PushClient::Core>,
+      public LoginDelegate,
+      public PushNotificationsListenTaskDelegate,
+      public PushNotificationsSubscribeTaskDelegate {
+ public:
+  // Called on the parent thread.
+  explicit Core(const NotifierOptions& notifier_options);
+
+  // Must be called before being destroyed.
+  void DestroyOnIOThread();
+
+  // Login::Delegate implementation.
+  virtual void OnConnect(
+      base::WeakPtr<buzz::XmppTaskParentInterface> base_task) OVERRIDE;
+  virtual void OnDisconnect();
+
+  // PushNotificationsListenTaskDelegate implementation.
+  virtual void OnNotificationReceived(
+      const Notification& notification) OVERRIDE;
+
+  // PushNotificationsSubscribeTaskDelegate implementation.
+  virtual void OnSubscribed() OVERRIDE;
+  virtual void OnSubscriptionError() OVERRIDE;
+
+  // Called on the parent thread.
+  void AddObserver(Observer* observer);
+  void RemoveObserver(Observer* observer);
+
+  void UpdateSubscriptions(const SubscriptionList& subscriptions);
+  void UpdateCredentials(const std::string& email, const std::string& token);
+  void SendNotification(const Notification& data);
+
+  // Any notifications sent after this is called will be reflected,
+  // i.e. will be treated as an incoming notification also.
+  void ReflectSentNotificationsForTest();
+
+ private:
+  friend class base::RefCountedThreadSafe<PushClient::Core>;
+
+  // Called on either the parent thread or the I/O thread.
+  virtual ~Core();
+
+  const NotifierOptions notifier_options_;
+  const scoped_refptr<base::MessageLoopProxy> parent_message_loop_proxy_;
+  const scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy_;
+  const scoped_refptr<ObserverListThreadSafe<Observer> > observers_;
+
+  // XMPP connection settings.
+  SubscriptionList subscriptions_;
+  buzz::XmppClientSettings xmpp_settings_;
+
+  // Must be created/used/destroyed only on the IO thread.
+  scoped_ptr<notifier::Login> login_;
+
+  // The XMPP connection.
+  base::WeakPtr<buzz::XmppTaskParentInterface> base_task_;
+
+  std::vector<Notification> pending_notifications_to_send_;
+
+  bool reflect_sent_notifications_for_test_;
+
+  DISALLOW_COPY_AND_ASSIGN(Core);
+};
+
+PushClient::Core::Core(const NotifierOptions& notifier_options)
+    : notifier_options_(notifier_options),
+      parent_message_loop_proxy_(base::MessageLoopProxy::current()),
+      io_message_loop_proxy_(
+          notifier_options_.request_context_getter->GetIOMessageLoopProxy()),
+      observers_(new ObserverListThreadSafe<Observer>()),
+      reflect_sent_notifications_for_test_(false) {}
+
+PushClient::Core::~Core() {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread() ||
+         io_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(!login_.get());
+  DCHECK(!base_task_.get());
+  observers_->AssertEmpty();
+}
+
+void PushClient::Core::DestroyOnIOThread() {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  login_.reset();
+  base_task_.reset();
+}
+
+void PushClient::Core::OnConnect(
+    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  base_task_ = base_task;
+
+  if (!base_task_.get()) {
+    NOTREACHED();
+    return;
+  }
+
+  // Listen for notifications.
+  {
+    // Owned by |base_task_|.
+    PushNotificationsListenTask* listener =
+        new PushNotificationsListenTask(base_task_, this);
+    listener->Start();
+  }
+
+  // Send subscriptions.
+  {
+    // Owned by |base_task_|.
+    PushNotificationsSubscribeTask* subscribe_task =
+        new PushNotificationsSubscribeTask(base_task_, subscriptions_, this);
+    subscribe_task->Start();
+  }
+
+  std::vector<Notification> notifications_to_send;
+  notifications_to_send.swap(pending_notifications_to_send_);
+  for (std::vector<Notification>::const_iterator it =
+           notifications_to_send.begin();
+       it != notifications_to_send.end(); ++it) {
+    DVLOG(1) << "Push: Sending pending notification " << it->ToString();
+    SendNotification(*it);
+  }
+}
+
+void PushClient::Core::OnDisconnect() {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  base_task_.reset();
+  observers_->Notify(&Observer::OnNotificationStateChange, false);
+}
+
+void PushClient::Core::OnNotificationReceived(
+    const Notification& notification) {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  observers_->Notify(&Observer::OnIncomingNotification, notification);
+}
+
+void PushClient::Core::OnSubscribed() {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  observers_->Notify(&Observer::OnNotificationStateChange, true);
+}
+
+void PushClient::Core::OnSubscriptionError() {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  observers_->Notify(&Observer::OnNotificationStateChange, false);
+}
+
+void PushClient::Core::AddObserver(Observer* observer) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  observers_->AddObserver(observer);
+}
+
+void PushClient::Core::RemoveObserver(Observer* observer) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  observers_->RemoveObserver(observer);
+}
+
+void PushClient::Core::UpdateSubscriptions(
+    const SubscriptionList& subscriptions) {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  subscriptions_ = subscriptions;
+}
+
+void PushClient::Core::UpdateCredentials(
+      const std::string& email, const std::string& token) {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  DVLOG(1) << "Push: Updating credentials for " << email;
+  xmpp_settings_ = MakeXmppClientSettings(notifier_options_, email, token);
+  if (login_.get()) {
+    login_->UpdateXmppSettings(xmpp_settings_);
+  } else {
+    DVLOG(1) << "Push: Starting XMPP connection";
+    base_task_.reset();
+    login_.reset(new notifier::Login(this,
+                                     xmpp_settings_,
+                                     notifier_options_.request_context_getter,
+                                     GetServerList(notifier_options_),
+                                     notifier_options_.try_ssltcp_first,
+                                     notifier_options_.auth_mechanism));
+    login_->StartConnection();
+  }
+}
+
+void PushClient::Core::SendNotification(const Notification& notification) {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  if (!base_task_.get()) {
+    DVLOG(1) << "Push: Cannot send notification "
+             << notification.ToString() << "; sending later";
+    pending_notifications_to_send_.push_back(notification);
+    return;
+  }
+  // Owned by |base_task_|.
+  PushNotificationsSendUpdateTask* task =
+      new PushNotificationsSendUpdateTask(base_task_, notification);
+  task->Start();
+
+  if (reflect_sent_notifications_for_test_) {
+    OnNotificationReceived(notification);
+  }
+}
+
+void PushClient::Core::ReflectSentNotificationsForTest() {
+  DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
+  reflect_sent_notifications_for_test_ = true;
+}
+
+PushClient::PushClient(const NotifierOptions& notifier_options)
+    : core_(new Core(notifier_options)),
+      parent_message_loop_proxy_(base::MessageLoopProxy::current()),
+      io_message_loop_proxy_(
+          notifier_options.request_context_getter->GetIOMessageLoopProxy()) {
+}
+
+PushClient::~PushClient() {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::DestroyOnIOThread, core_.get()));
+}
+
+void PushClient::AddObserver(Observer* observer) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  core_->AddObserver(observer);
+}
+
+void PushClient::RemoveObserver(Observer* observer) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  core_->RemoveObserver(observer);
+}
+
+void PushClient::UpdateSubscriptions(const SubscriptionList& subscriptions) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::UpdateSubscriptions,
+                 core_.get(), subscriptions));
+}
+
+void PushClient::UpdateCredentials(
+    const std::string& email, const std::string& token) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::UpdateCredentials,
+                 core_.get(), email, token));
+}
+
+void PushClient::SendNotification(const Notification& notification) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::SendNotification, core_.get(),
+                 notification));
+}
+
+void PushClient::SimulateOnNotificationReceivedForTest(
+    const Notification& notification) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::OnNotificationReceived,
+                 core_.get(), notification));
+}
+
+void PushClient::SimulateConnectAndSubscribeForTest(
+    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::OnConnect, core_.get(), base_task));
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::OnSubscribed, core_.get()));
+}
+
+void PushClient::SimulateDisconnectForTest() {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::OnDisconnect, core_.get()));
+}
+
+void PushClient::SimulateSubscriptionErrorForTest() {
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::OnSubscriptionError, core_.get()));
+}
+
+void PushClient::ReflectSentNotificationsForTest() {
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::ReflectSentNotificationsForTest,
+                 core_.get()));
+}
+
+}  // namespace notifier
diff --git a/jingle/notifier/listener/push_client.h b/jingle/notifier/listener/push_client.h
new file mode 100644
index 0000000..fdac8ba
--- /dev/null
+++ b/jingle/notifier/listener/push_client.h
@@ -0,0 +1,98 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef JINGLE_NOTIFIER_LISTENER_PUSH_CLIENT_H_
+#define JINGLE_NOTIFIER_LISTENER_PUSH_CLIENT_H_
+
+#include <string>
+#include <vector>
+
+#include "base/basictypes.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
+#include "jingle/notifier/base/notifier_options.h"
+#include "jingle/notifier/listener/notification_defines.h"
+
+namespace base {
+class MessageLoopProxy;
+} // namespace base
+
+namespace buzz {
+class XmppTaskParentInterface;
+}  // namespace buzz
+
+namespace notifier {
+
+// This class implements a client for the XMPP google:push protocol.
+//
+// This class must be used on a single thread.
+class PushClient {
+ public:
+  // An Observer is sent messages whenever a notification is received
+  // or when the state of the push client changes.
+  class Observer {
+   public:
+    // Called when the state of the push client changes.  If
+    // |notifications_enabled| is true, that means notifications can
+    // be sent and received freely.  If it is false, that means no
+    // notifications can be sent or received.
+    virtual void OnNotificationStateChange(bool notifications_enabled) = 0;
+
+    // Called when a notification is received.  The details of the
+    // notification are in |notification|.
+    virtual void OnIncomingNotification(const Notification& notification) = 0;
+
+   protected:
+    virtual ~Observer();
+  };
+
+  explicit PushClient(const NotifierOptions& notifier_options);
+  ~PushClient();
+
+  void AddObserver(Observer* observer);
+  void RemoveObserver(Observer* observer);
+
+  // Takes effect only on the next (re-)connection.  Therefore, you
+  // probably want to call this before UpdateCredentials().
+  void UpdateSubscriptions(const SubscriptionList& subscriptions);
+
+  // If not connected, connects with the given credentials.  If
+  // already connected, the next connection attempt will use the given
+  // credentials.
+  void UpdateCredentials(const std::string& email, const std::string& token);
+
+  // Sends a notification.  Can be called when notifications are
+  // disabled; the notification will be sent when notifications become
+  // enabled.
+  void SendNotification(const Notification& notification);
+
+  void SimulateOnNotificationReceivedForTest(
+      const Notification& notification);
+
+  void SimulateConnectAndSubscribeForTest(
+      base::WeakPtr<buzz::XmppTaskParentInterface> base_task);
+
+  void SimulateDisconnectForTest();
+
+  void SimulateSubscriptionErrorForTest();
+
+  // Any notifications sent after this is called will be reflected,
+  // i.e. will be treated as an incoming notification also.
+  void ReflectSentNotificationsForTest();
+
+ private:
+  class Core;
+
+  // The real guts of PushClient, which allows this class to not be
+  // refcounted.
+  const scoped_refptr<Core> core_;
+  const scoped_refptr<base::MessageLoopProxy> parent_message_loop_proxy_;
+  const scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy_;
+
+  DISALLOW_COPY_AND_ASSIGN(PushClient);
+};
+
+}  // namespace notifier
+
+#endif  // JINGLE_NOTIFIER_LISTENER_PUSH_CLIENT_H_
diff --git a/jingle/notifier/listener/push_client_unittest.cc b/jingle/notifier/listener/push_client_unittest.cc
new file mode 100644
index 0000000..a5663d7
--- /dev/null
+++ b/jingle/notifier/listener/push_client_unittest.cc
@@ -0,0 +1,105 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "jingle/notifier/listener/push_client.h"
+
+#include "base/compiler_specific.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/message_loop.h"
+#include "jingle/notifier/base/fake_base_task.h"
+#include "jingle/notifier/base/notifier_options.h"
+#include "net/url_request/url_request_test_util.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace notifier {
+
+namespace {
+
+using ::testing::_;
+using ::testing::Mock;
+using ::testing::StrictMock;
+
+class MockObserver : public PushClient::Observer {
+ public:
+  MOCK_METHOD1(OnNotificationStateChange, void(bool));
+  MOCK_METHOD1(OnIncomingNotification, void(const Notification&));
+};
+
+}  // namespace
+
+class PushClientTest : public testing::Test {
+ protected:
+  PushClientTest() {
+    notifier_options_.request_context_getter =
+        new TestURLRequestContextGetter(message_loop_.message_loop_proxy());
+  }
+
+  virtual ~PushClientTest() {}
+
+  virtual void SetUp() OVERRIDE {
+    push_client_.reset(new PushClient(notifier_options_));
+    push_client_->AddObserver(&mock_observer_);
+  }
+
+  virtual void TearDown() OVERRIDE {
+    // Clear out any messages posted by PushClient.
+    message_loop_.RunAllPending();
+    push_client_->RemoveObserver(&mock_observer_);
+    push_client_.reset();
+  }
+
+  // The sockets created by the XMPP code expect an IO loop.
+  MessageLoopForIO message_loop_;
+  NotifierOptions notifier_options_;
+  StrictMock<MockObserver> mock_observer_;
+  scoped_ptr<PushClient> push_client_;
+  FakeBaseTask fake_base_task_;
+};
+
+TEST_F(PushClientTest, OnIncomingNotification) {
+  EXPECT_CALL(mock_observer_, OnIncomingNotification(_));
+  push_client_->SimulateOnNotificationReceivedForTest(Notification());
+}
+
+TEST_F(PushClientTest, ConnectAndSubscribe) {
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
+  push_client_->SimulateConnectAndSubscribeForTest(
+      fake_base_task_.AsWeakPtr());
+}
+
+TEST_F(PushClientTest, Disconnect) {
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(false));
+  push_client_->SimulateDisconnectForTest();
+}
+
+TEST_F(PushClientTest, SubscriptionError) {
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(false));
+  push_client_->SimulateSubscriptionErrorForTest();
+}
+
+TEST_F(PushClientTest, SendNotification) {
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
+  EXPECT_CALL(mock_observer_, OnIncomingNotification(_));
+
+  push_client_->SimulateConnectAndSubscribeForTest(
+      fake_base_task_.AsWeakPtr());
+  push_client_->ReflectSentNotificationsForTest();
+  push_client_->SendNotification(Notification());
+}
+
+TEST_F(PushClientTest, SendNotificationPending) {
+  push_client_->ReflectSentNotificationsForTest();
+  push_client_->SendNotification(Notification());
+
+  Mock::VerifyAndClearExpectations(&mock_observer_);
+
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
+  EXPECT_CALL(mock_observer_, OnIncomingNotification(_));
+
+  push_client_->SimulateConnectAndSubscribeForTest(
+      fake_base_task_.AsWeakPtr());
+}
+
+}  // namespace notifier
diff --git a/jingle/notifier/listener/talk_mediator.h b/jingle/notifier/listener/talk_mediator.h
deleted file mode 100644
index 7a6fa79..0000000
--- a/jingle/notifier/listener/talk_mediator.h
+++ /dev/null
@@ -1,61 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-//
-// Interface to the code which handles talk logic.  Used to initialize SSL
-// before the underlying talk login occurs.
-//
-// Example usage:
-//
-//   TalkMediator mediator();
-//   mediator.SetAuthToken("email", "token", "service_id");
-//   mediator.Login();
-//   ...
-//   mediator.Logout();
-
-#ifndef JINGLE_NOTIFIER_LISTENER_TALK_MEDIATOR_H_
-#define JINGLE_NOTIFIER_LISTENER_TALK_MEDIATOR_H_
-
-#include <string>
-
-#include "jingle/notifier/listener/notification_defines.h"
-
-namespace notifier {
-
-class TalkMediator {
- public:
-  TalkMediator() {}
-  virtual ~TalkMediator() {}
-
-  class Delegate {
-   public:
-    virtual ~Delegate() {}
-
-    virtual void OnNotificationStateChange(bool notifications_enabled) = 0;
-
-    virtual void OnIncomingNotification(const Notification& notification) = 0;
-
-    virtual void OnOutgoingNotification() = 0;
-  };
-
-  // |delegate| may be NULL.
-  virtual void SetDelegate(Delegate* delegate) = 0;
-
-  // The following methods are for authorizaiton of the xmpp client.
-  virtual void SetAuthToken(const std::string& email,
-                            const std::string& token,
-                            const std::string& token_service) = 0;
-  virtual bool Login() = 0;
-  virtual bool Logout() = 0;
-
-  // Method for the owner of this object to notify peers that an update has
-  // occurred.
-  virtual void SendNotification(const Notification& data) = 0;
-
-  // Add a subscription to subscribe to.
-  virtual void AddSubscription(const Subscription& subscription) = 0;
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_TALK_MEDIATOR_H_
diff --git a/jingle/notifier/listener/talk_mediator_impl.cc b/jingle/notifier/listener/talk_mediator_impl.cc
deleted file mode 100644
index 7760261..0000000
--- a/jingle/notifier/listener/talk_mediator_impl.cc
+++ /dev/null
@@ -1,134 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#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 {
-
-TalkMediatorImpl::TalkMediatorImpl(
-    MediatorThread* mediator_thread,
-    const NotifierOptions& notifier_options)
-    : delegate_(NULL),
-      mediator_thread_(mediator_thread),
-      notifier_options_(notifier_options),
-      parent_message_loop_(MessageLoop::current()) {
-  mediator_thread_->Start();
-  state_.started = 1;
-}
-
-TalkMediatorImpl::~TalkMediatorImpl() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  DCHECK(!state_.started);
-}
-
-bool TalkMediatorImpl::Login() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  // Connect to the mediator thread and start processing messages.
-  mediator_thread_->AddObserver(this);
-  if (state_.initialized && !state_.logging_in && !state_.logged_in) {
-    state_.logging_in = true;
-    mediator_thread_->Login(xmpp_settings_);
-    return true;
-  }
-  return false;
-}
-
-bool TalkMediatorImpl::Logout() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  if (state_.started) {
-    state_.started = 0;
-    state_.logging_in = 0;
-    state_.logged_in = 0;
-    // We do not want to be called back during logout since we may be
-    // closing.
-    mediator_thread_->RemoveObserver(this);
-    mediator_thread_->Logout();
-    return true;
-  }
-  return false;
-}
-
-void TalkMediatorImpl::SendNotification(const Notification& data) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  mediator_thread_->SendNotification(data);
-}
-
-void TalkMediatorImpl::SetDelegate(TalkMediator::Delegate* delegate) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  delegate_ = delegate;
-}
-
-void TalkMediatorImpl::SetAuthToken(const std::string& email,
-                                    const std::string& token,
-                                    const std::string& token_service) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  xmpp_settings_ =
-      MakeXmppClientSettings(notifier_options_, email, token, token_service);
-
-  // The auth token got updated and we are already in the logging_in or
-  // logged_in state. Update the token.
-  if (state_.logging_in || state_.logged_in) {
-    mediator_thread_->UpdateXmppSettings(xmpp_settings_);
-  }
-
-  state_.initialized = 1;
-}
-
-void TalkMediatorImpl::AddSubscription(const Subscription& subscription) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  subscriptions_.push_back(subscription);
-  if (state_.logged_in) {
-    VLOG(1) << "Resubscribing for updates, a new service got added";
-    mediator_thread_->SubscribeForUpdates(subscriptions_);
-  }
-}
-
-
-void TalkMediatorImpl::OnConnectionStateChange(bool logged_in) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  // 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;
-  state_.logged_in = logged_in;
-  if (logged_in) {
-    VLOG(1) << "P2P: Logged in.";
-    // ListenForUpdates enables the ListenTask.  This is done before
-    // SubscribeForUpdates.
-    mediator_thread_->ListenForUpdates();
-    // Now subscribe for updates to all the services we are interested in
-    mediator_thread_->SubscribeForUpdates(subscriptions_);
-  } else {
-    VLOG(1) << "P2P: Logged off.";
-    OnSubscriptionStateChange(false);
-  }
-}
-
-void TalkMediatorImpl::OnSubscriptionStateChange(bool subscribed) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  VLOG(1) << "P2P: " << (subscribed ? "subscribed" : "unsubscribed");
-  if (delegate_)
-    delegate_->OnNotificationStateChange(subscribed);
-}
-
-void TalkMediatorImpl::OnIncomingNotification(
-    const Notification& notification) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  VLOG(1) << "P2P: Updates are available on the server.";
-  if (delegate_)
-    delegate_->OnIncomingNotification(notification);
-}
-
-void TalkMediatorImpl::OnOutgoingNotification() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  VLOG(1) << "P2P: Peers were notified that updates are available on the "
-             "server.";
-  if (delegate_)
-    delegate_->OnOutgoingNotification();
-}
-
-}  // namespace notifier
diff --git a/jingle/notifier/listener/talk_mediator_impl.h b/jingle/notifier/listener/talk_mediator_impl.h
deleted file mode 100644
index d49e054..0000000
--- a/jingle/notifier/listener/talk_mediator_impl.h
+++ /dev/null
@@ -1,111 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-//
-// This class is the interface between talk code and the client code proper
-// It will manage all aspects of the connection and call back into the client
-// when it needs attention (for instance if updates are available for syncing).
-
-#ifndef JINGLE_NOTIFIER_LISTENER_TALK_MEDIATOR_IMPL_H_
-#define JINGLE_NOTIFIER_LISTENER_TALK_MEDIATOR_IMPL_H_
-
-#include <string>
-#include <vector>
-
-#include "base/compiler_specific.h"
-#include "base/gtest_prod_util.h"
-#include "base/memory/scoped_ptr.h"
-#include "base/threading/non_thread_safe.h"
-#include "jingle/notifier/base/notifier_options.h"
-#include "jingle/notifier/listener/mediator_thread.h"
-#include "jingle/notifier/listener/talk_mediator.h"
-#include "talk/xmpp/xmppclientsettings.h"
-
-class MessageLoop;
-
-namespace notifier {
-
-class TalkMediatorImpl
-    : public TalkMediator, public MediatorThread::Observer {
- public:
-  // Takes ownership of |mediator_thread|. It is guaranteed that
-  // |mediator_thread| is destroyed only when this object is destroyed.
-  // This means that you can store a pointer to mediator_thread separately
-  // and use it until this object is destroyed.
-  TalkMediatorImpl(
-      MediatorThread* mediator_thread,
-      const NotifierOptions& notifier_options);
-  virtual ~TalkMediatorImpl();
-
-  // TalkMediator implementation.
-
-  // Should be called on the same thread as the constructor.
-  virtual void SetDelegate(TalkMediator::Delegate* delegate) OVERRIDE;
-
-  // 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) OVERRIDE;
-  virtual bool Login() OVERRIDE;
-  // Users must call Logout once Login is called.
-  virtual bool Logout() OVERRIDE;
-
-  virtual void SendNotification(const Notification& data) OVERRIDE;
-
-  virtual void AddSubscription(const Subscription& subscription) OVERRIDE;
-
-  // MediatorThread::Delegate implementation.
-
-  virtual void OnConnectionStateChange(bool logged_in) OVERRIDE;
-
-  virtual void OnSubscriptionStateChange(bool subscribed) OVERRIDE;
-
-  virtual void OnIncomingNotification(
-      const Notification& notification) OVERRIDE;
-
-  virtual void OnOutgoingNotification() OVERRIDE;
-
- private:
-  struct TalkMediatorState {
-    TalkMediatorState()
-        : started(0), initialized(0), logging_in(0), logged_in(0) {
-    }
-
-    unsigned int started : 1;      // Background thread has started.
-    unsigned int initialized : 1;  // Initialized with login information.
-    unsigned int logging_in : 1;   // Logging in to the mediator's
-                                   // authenticator.
-    unsigned int logged_in : 1;    // Logged in the mediator's authenticator.
-  };
-
-  // Delegate, which we don't own.  May be NULL.
-  TalkMediator::Delegate* delegate_;
-
-  // Internal state.
-  TalkMediatorState state_;
-
-  // Cached and verfied from the SetAuthToken method.
-  buzz::XmppClientSettings xmpp_settings_;
-
-  // The worker thread through which talk events are posted and received.
-  scoped_ptr<MediatorThread> mediator_thread_;
-
-  const NotifierOptions notifier_options_;
-
-  SubscriptionList subscriptions_;
-
-  MessageLoop* parent_message_loop_;
-
-  FRIEND_TEST_ALL_PREFIXES(TalkMediatorImplTest, SetAuthToken);
-  FRIEND_TEST_ALL_PREFIXES(TalkMediatorImplTest, SendNotification);
-  FRIEND_TEST_ALL_PREFIXES(TalkMediatorImplTest, MediatorThreadCallbacks);
-
-  DISALLOW_COPY_AND_ASSIGN(TalkMediatorImpl);
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_TALK_MEDIATOR_IMPL_H_
diff --git a/jingle/notifier/listener/talk_mediator_unittest.cc b/jingle/notifier/listener/talk_mediator_unittest.cc
deleted file mode 100644
index c72e0e9..0000000
--- a/jingle/notifier/listener/talk_mediator_unittest.cc
+++ /dev/null
@@ -1,171 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include <string>
-
-#include "base/basictypes.h"
-#include "base/message_loop.h"
-#include "jingle/notifier/base/notifier_options.h"
-#include "jingle/notifier/listener/mediator_thread_mock.h"
-#include "jingle/notifier/listener/mediator_thread_impl.h"
-#include "jingle/notifier/listener/talk_mediator_impl.h"
-#include "talk/xmpp/xmppengine.h"
-#include "testing/gmock/include/gmock/gmock.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-namespace notifier {
-
-using ::testing::_;
-
-class MockTalkMediatorDelegate : public TalkMediator::Delegate {
- public:
-  MockTalkMediatorDelegate() {}
-  virtual ~MockTalkMediatorDelegate() {}
-
-  MOCK_METHOD1(OnNotificationStateChange,
-               void(bool notification_changed));
-  MOCK_METHOD1(OnIncomingNotification,
-               void(const Notification& data));
-  MOCK_METHOD0(OnOutgoingNotification, void());
-
- private:
-  DISALLOW_COPY_AND_ASSIGN(MockTalkMediatorDelegate);
-};
-
-class TalkMediatorImplTest : public testing::Test {
- protected:
-  TalkMediatorImplTest() {}
-  virtual ~TalkMediatorImplTest() {}
-
-  TalkMediatorImpl* NewMockedTalkMediator(
-      MockMediatorThread* mock_mediator_thread) {
-    return new TalkMediatorImpl(mock_mediator_thread,
-                                NotifierOptions());
-  }
-
-  int last_message_;
-
- private:
-  // TalkMediatorImpl expects a message loop.
-  MessageLoop message_loop_;
-
-  DISALLOW_COPY_AND_ASSIGN(TalkMediatorImplTest);
-};
-
-TEST_F(TalkMediatorImplTest, SetAuthToken) {
-  scoped_ptr<TalkMediatorImpl> talk1(
-      NewMockedTalkMediator(new MockMediatorThread()));
-  talk1->SetAuthToken("chromium@gmail.com", "token", "fake_service");
-  EXPECT_TRUE(talk1->state_.initialized);
-  talk1->Logout();
-
-  scoped_ptr<TalkMediatorImpl> talk2(
-      NewMockedTalkMediator(new MockMediatorThread()));
-  talk2->SetAuthToken("chromium@mail.google.com", "token", "fake_service");
-  EXPECT_TRUE(talk2->state_.initialized);
-  talk2->Logout();
-
-  scoped_ptr<TalkMediatorImpl> talk3(
-      NewMockedTalkMediator(new MockMediatorThread()));
-  talk3->SetAuthToken("chromium@mail.google.com", "token", "fake_service");
-  EXPECT_TRUE(talk3->state_.initialized);
-  talk3->Logout();
-}
-
-TEST_F(TalkMediatorImplTest, LoginWiring) {
-  // The TalkMediatorImpl owns the mock.
-  MockMediatorThread* mock = new MockMediatorThread();
-  scoped_ptr<TalkMediatorImpl> talk1(NewMockedTalkMediator(mock));
-
-  // Login checks states for initialization.
-  EXPECT_FALSE(talk1->Login());
-  EXPECT_EQ(0, mock->login_calls);
-
-  talk1->SetAuthToken("chromium@gmail.com", "token", "fake_service");
-  EXPECT_EQ(0, mock->update_settings_calls);
-
-  EXPECT_TRUE(talk1->Login());
-  EXPECT_EQ(1, mock->login_calls);
-
-  // We call SetAuthToken again to update the settings after an update.
-  talk1->SetAuthToken("chromium@gmail.com", "token", "fake_service");
-  EXPECT_EQ(1, mock->update_settings_calls);
-
-  // Successive calls to login will fail.  One needs to create a new talk
-  // mediator object.
-  EXPECT_FALSE(talk1->Login());
-  EXPECT_EQ(1, mock->login_calls);
-
-  EXPECT_TRUE(talk1->Logout());
-  EXPECT_EQ(1, mock->logout_calls);
-
-  // Successive logout calls do nothing.
-  EXPECT_FALSE(talk1->Logout());
-  EXPECT_EQ(1, mock->logout_calls);
-}
-
-TEST_F(TalkMediatorImplTest, SendNotification) {
-  // The TalkMediatorImpl owns the mock.
-  MockMediatorThread* mock = new MockMediatorThread();
-  scoped_ptr<TalkMediatorImpl> talk1(NewMockedTalkMediator(mock));
-
-  Notification data;
-  talk1->SendNotification(data);
-  EXPECT_EQ(1, mock->send_calls);
-
-  talk1->SetAuthToken("chromium@gmail.com", "token", "fake_service");
-  EXPECT_TRUE(talk1->Login());
-  talk1->OnConnectionStateChange(true);
-  EXPECT_EQ(1, mock->login_calls);
-
-  talk1->SendNotification(data);
-  EXPECT_EQ(2, mock->send_calls);
-  talk1->SendNotification(data);
-  EXPECT_EQ(3, mock->send_calls);
-
-  EXPECT_TRUE(talk1->Logout());
-  EXPECT_EQ(1, mock->logout_calls);
-
-  talk1->SendNotification(data);
-  EXPECT_EQ(4, mock->send_calls);
-}
-
-TEST_F(TalkMediatorImplTest, MediatorThreadCallbacks) {
-  // The TalkMediatorImpl owns the mock.
-  MockMediatorThread* mock = new MockMediatorThread();
-  scoped_ptr<TalkMediatorImpl> talk1(NewMockedTalkMediator(mock));
-
-  MockTalkMediatorDelegate mock_delegate;
-  EXPECT_CALL(mock_delegate, OnNotificationStateChange(true));
-  EXPECT_CALL(mock_delegate, OnIncomingNotification(_));
-  EXPECT_CALL(mock_delegate, OnOutgoingNotification());
-
-  talk1->SetDelegate(&mock_delegate);
-
-  talk1->SetAuthToken("chromium@gmail.com", "token", "fake_service");
-  EXPECT_TRUE(talk1->Login());
-  EXPECT_EQ(1, mock->login_calls);
-
-  // The message triggers calls to listen and subscribe.
-  EXPECT_EQ(1, mock->listen_calls);
-  EXPECT_EQ(1, mock->subscribe_calls);
-
-  // After subscription success is receieved, the talk mediator will allow
-  // sending of notifications.
-  Notification outgoing_data;
-  talk1->SendNotification(outgoing_data);
-  EXPECT_EQ(1, mock->send_calls);
-
-  Notification incoming_data;
-  incoming_data.channel = "service_url";
-  incoming_data.data = "service_data";
-  mock->ReceiveNotification(incoming_data);
-
-  // Shouldn't trigger a call to the delegate since we disconnect
-  // it before we logout the mediator thread.
-  talk1->Logout();
-  talk1.reset();
-}
-
-}  // namespace notifier