Revert 138216 - [Sync] Turn notifier::PushClient into an interface

Split the previous implementation into two pieces:
XmppPushClient and NonBlockingPushClient.

Add FakePushClient and FakePushClientObserver.

Remove use of ThreadSafeObserverList.

Add PushClient::CreateDefault() function, which creates a
NonBlockingPushClient for an XmppPushClient.

Dep-inject PushClient into P2PNotifier.

Add some helper functions to notification_defines.{h,cc}.

BUG=76764
TEST=

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

TBR=akalin@chromium.org
Review URL: https://chromiumcodereview.appspot.com/10388227

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

18 files changed, 442 insertions, 1141 deletions

diff --git a/jingle/jingle.gyp b/jingle/jingle.gyp
index 3d424a0..9f9af8b 100644
--- a/jingle/jingle.gyp
+++ b/jingle/jingle.gyp
@@ -75,16 +75,12 @@
         'notifier/communicator/login_settings.h',
         'notifier/communicator/single_login_attempt.cc',
         'notifier/communicator/single_login_attempt.h',
-        'notifier/listener/non_blocking_push_client.cc',
-        'notifier/listener/non_blocking_push_client.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',
         'notifier/listener/notification_defines.h',
-        'notifier/listener/push_client_observer.cc',
-        'notifier/listener/push_client_observer.h',
-        'notifier/listener/push_client.cc',
-        'notifier/listener/push_client.h',
         'notifier/listener/push_notifications_listen_task.cc',
         'notifier/listener/push_notifications_listen_task.h',
         'notifier/listener/push_notifications_send_update_task.cc',
@@ -93,8 +89,6 @@
         'notifier/listener/push_notifications_subscribe_task.h',
         'notifier/listener/xml_element_util.cc',
         'notifier/listener/xml_element_util.h',
-        'notifier/listener/xmpp_push_client.cc',
-        'notifier/listener/xmpp_push_client.h',
       ],
       'defines' : [
         '_CRT_SECURE_NO_WARNINGS',
@@ -126,10 +120,6 @@
         'notifier/base/fake_base_task.h',
         'notifier/base/mock_task.cc',
         'notifier/base/mock_task.h',
-        'notifier/listener/fake_push_client.cc',
-        'notifier/listener/fake_push_client.h',
-        'notifier/listener/fake_push_client_observer.cc',
-        'notifier/listener/fake_push_client_observer.h',
       ],
       'dependencies': [
         'notifier',
@@ -170,12 +160,10 @@
         'notifier/communicator/connection_settings_unittest.cc',
         'notifier/communicator/login_settings_unittest.cc',
         'notifier/communicator/single_login_attempt_unittest.cc',
-        'notifier/listener/non_blocking_push_client_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/xml_element_util_unittest.cc',
-        'notifier/listener/xmpp_push_client_unittest.cc',
         'run_all_unittests.cc',
       ],
       'conditions': [
diff --git a/jingle/notifier/listener/fake_push_client.cc b/jingle/notifier/listener/fake_push_client.cc
deleted file mode 100644
index cfe4a99..0000000
--- a/jingle/notifier/listener/fake_push_client.cc
+++ /dev/null
@@ -1,67 +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/fake_push_client.h"
-
-#include "jingle/notifier/listener/push_client_observer.h"
-
-namespace notifier {
-
-FakePushClient::FakePushClient() {}
-
-FakePushClient::~FakePushClient() {}
-
-void FakePushClient::AddObserver(PushClientObserver* observer) {
-  observers_.AddObserver(observer);
-}
-
-void FakePushClient::RemoveObserver(PushClientObserver* observer) {
-  observers_.RemoveObserver(observer);
-}
-
-void FakePushClient::UpdateSubscriptions(
-    const SubscriptionList& subscriptions) {
-  subscriptions_ = subscriptions;
-}
-
-void FakePushClient::UpdateCredentials(
-    const std::string& email, const std::string& token) {
-  email_ = email;
-  token_ = token;
-}
-
-void FakePushClient::SendNotification(const Notification& notification) {
-  sent_notifications_.push_back(notification);
-}
-
-void FakePushClient::SimulateNotificationStateChange(
-    bool notifications_enabled) {
-  FOR_EACH_OBSERVER(PushClientObserver, observers_,
-                    OnNotificationStateChange(notifications_enabled));
-}
-
-void FakePushClient::SimulateIncomingNotification(
-    const Notification& notification) {
-  FOR_EACH_OBSERVER(PushClientObserver, observers_,
-                    OnIncomingNotification(notification));
-}
-
-const SubscriptionList& FakePushClient::subscriptions() const {
-  return subscriptions_;
-}
-
-const std::string& FakePushClient::email() const {
-  return email_;
-}
-
-const std::string& FakePushClient::token() const {
-  return token_;
-}
-
-const std::vector<Notification>& FakePushClient::sent_notifications() const {
-  return sent_notifications_;
-}
-
-}  // namespace notifier
-
diff --git a/jingle/notifier/listener/fake_push_client.h b/jingle/notifier/listener/fake_push_client.h
deleted file mode 100644
index 61ba72b..0000000
--- a/jingle/notifier/listener/fake_push_client.h
+++ /dev/null
@@ -1,55 +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.
-
-#ifndef JINGLE_NOTIFIER_LISTENER_FAKE_PUSH_CLIENT_H_
-#define JINGLE_NOTIFIER_LISTENER_FAKE_PUSH_CLIENT_H_
-
-#include <vector>
-
-#include "base/basictypes.h"
-#include "base/compiler_specific.h"
-#include "base/observer_list.h"
-#include "jingle/notifier/listener/push_client.h"
-
-namespace notifier {
-
-// PushClient implementation that can be used for testing.
-class FakePushClient : public PushClient {
- public:
-  FakePushClient();
-  virtual ~FakePushClient();
-
-  // PushClient implementation.
-  virtual void AddObserver(PushClientObserver* observer) OVERRIDE;
-  virtual void RemoveObserver(PushClientObserver* observer) OVERRIDE;
-  virtual void UpdateSubscriptions(
-      const SubscriptionList& subscriptions) OVERRIDE;
-  virtual void UpdateCredentials(
-      const std::string& email, const std::string& token) OVERRIDE;
-  virtual void SendNotification(const Notification& notification) OVERRIDE;
-
-  // Triggers OnNotificationStateChange on all observers.
-  void SimulateNotificationStateChange(bool notifications_enabled);
-
-  // Triggers OnIncomingNotification on all observers.
-  void SimulateIncomingNotification(const Notification& notification);
-
-  const SubscriptionList& subscriptions() const;
-  const std::string& email() const;
-  const std::string& token() const;
-  const std::vector<Notification>& sent_notifications() const;
-
- private:
-  ObserverList<PushClientObserver> observers_;
-  SubscriptionList subscriptions_;
-  std::string email_;
-  std::string token_;
-  std::vector<Notification> sent_notifications_;
-
-  DISALLOW_COPY_AND_ASSIGN(FakePushClient);
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_FAKE_PUSH_CLIENT_H_
diff --git a/jingle/notifier/listener/fake_push_client_observer.cc b/jingle/notifier/listener/fake_push_client_observer.cc
deleted file mode 100644
index ec1b9e1..0000000
--- a/jingle/notifier/listener/fake_push_client_observer.cc
+++ /dev/null
@@ -1,34 +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/fake_push_client_observer.h"
-
-namespace notifier {
-
-FakePushClientObserver::FakePushClientObserver()
-    : notifications_enabled_(false) {}
-
-FakePushClientObserver::~FakePushClientObserver() {}
-
-void FakePushClientObserver::OnNotificationStateChange(
-    bool notifications_enabled) {
-  notifications_enabled_ = notifications_enabled;
-}
-
-void FakePushClientObserver::OnIncomingNotification(
-    const Notification& notification) {
-  last_incoming_notification_ = notification;
-}
-
-bool FakePushClientObserver::notifications_enabled() const {
-  return notifications_enabled_;
-}
-
-const Notification&
-FakePushClientObserver::last_incoming_notification() const {
-  return last_incoming_notification_;
-}
-
-}  // namespace notifier
-
diff --git a/jingle/notifier/listener/fake_push_client_observer.h b/jingle/notifier/listener/fake_push_client_observer.h
deleted file mode 100644
index b8ee6d6..0000000
--- a/jingle/notifier/listener/fake_push_client_observer.h
+++ /dev/null
@@ -1,35 +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.
-
-#ifndef JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_FAKE_PUSH_CLIENT_OBSERVER_H_
-#define JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_FAKE_PUSH_CLIENT_OBSERVER_H_
-
-#include "base/compiler_specific.h"
-#include "jingle/notifier/listener/push_client_observer.h"
-
-namespace notifier {
-
-// PushClientObserver implementation that can be used for testing.
-class FakePushClientObserver : public PushClientObserver {
- public:
-  FakePushClientObserver();
-  virtual ~FakePushClientObserver();
-
-  // PushClientObserver implementation.
-  virtual void OnNotificationStateChange(
-      bool notifications_enabled) OVERRIDE;
-  virtual void OnIncomingNotification(
-      const Notification& notification) OVERRIDE;
-
-  bool notifications_enabled() const;
-  const Notification& last_incoming_notification() const;
-
- private:
-  bool notifications_enabled_;
-  Notification last_incoming_notification_;
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_FAKE_PUSH_CLIENT_OBSERVER_H_
diff --git a/jingle/notifier/listener/non_blocking_push_client.cc b/jingle/notifier/listener/non_blocking_push_client.cc
deleted file mode 100644
index 45d64be..0000000
--- a/jingle/notifier/listener/non_blocking_push_client.cc
+++ /dev/null
@@ -1,218 +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/non_blocking_push_client.h"
-
-#include "base/bind.h"
-#include "base/compiler_specific.h"
-#include "base/message_loop_proxy.h"
-#include "base/location.h"
-#include "base/logging.h"
-#include "jingle/notifier/listener/push_client_observer.h"
-
-namespace notifier {
-
-// All methods are called on the delegate thread unless specified
-// otherwise.
-class NonBlockingPushClient::Core
-    : public base::RefCountedThreadSafe<NonBlockingPushClient::Core>,
-      public PushClientObserver {
- public:
-  // Called on the parent thread.
-  explicit Core(
-      const CreateBlockingPushClientCallback&
-          create_blocking_push_client_callback,
-      const scoped_refptr<base::SingleThreadTaskRunner>&
-          delegate_task_runner,
-      const base::WeakPtr<NonBlockingPushClient>& parent_push_client);
-
-  void CreateOnDelegateThread(
-      const CreateBlockingPushClientCallback&
-          create_blocking_push_client_callback);
-
-  // Must be called before being destroyed.
-  void DestroyOnDelegateThread();
-
-  void UpdateSubscriptions(const SubscriptionList& subscriptions);
-  void UpdateCredentials(const std::string& email, const std::string& token);
-  void SendNotification(const Notification& data);
-
-  virtual void OnNotificationStateChange(
-      bool notifications_enabled) OVERRIDE;
-  virtual void OnIncomingNotification(
-      const Notification& notification) OVERRIDE;
-
- private:
-  friend class base::RefCountedThreadSafe<NonBlockingPushClient::Core>;
-
-  // Called on either the parent thread or the delegate thread.
-  virtual ~Core();
-
-  const scoped_refptr<base::SingleThreadTaskRunner> parent_task_runner_;
-  const scoped_refptr<base::SingleThreadTaskRunner> delegate_task_runner_;
-
-  const base::WeakPtr<NonBlockingPushClient> parent_push_client_;
-  scoped_ptr<PushClient> delegate_push_client_;
-
-  DISALLOW_COPY_AND_ASSIGN(Core);
-};
-
-NonBlockingPushClient::Core::Core(
-    const CreateBlockingPushClientCallback&
-        create_blocking_push_client_callback,
-    const scoped_refptr<base::SingleThreadTaskRunner>& delegate_task_runner,
-    const base::WeakPtr<NonBlockingPushClient>& parent_push_client)
-    : parent_task_runner_(base::MessageLoopProxy::current()),
-      delegate_task_runner_(delegate_task_runner),
-      parent_push_client_(parent_push_client) {
-  delegate_task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&NonBlockingPushClient::Core::CreateOnDelegateThread,
-                 this, create_blocking_push_client_callback));
-}
-
-NonBlockingPushClient::Core::~Core() {
-  // We should be destroyed on the parent or delegate thread.
-  // However, BelongsToCurrentThread() may return false for both if
-  // we're destroyed late in shutdown, e.g. after the BrowserThread
-  // structures have been destroyed.
-
-  // DestroyOnDelegateThread() may not have been run.  One
-  // particular case is when the DestroyOnDelegateThread closure is
-  // destroyed by the delegate thread's MessageLoop, which then
-  // releases the last reference to this object (although the last
-  // reference may end up being released on the parent thread after
-  // this); this case occasionally happens on the sync integration
-  // tests.
-  //
-  // In any case, even though nothing may be using
-  // |delegate_push_client_| anymore, we can't assume it's safe to
-  // delete it, nor can we assume that it's safe to remove ourselves
-  // as an observer.  So we just leak it.
-  ignore_result(delegate_push_client_.release());
-}
-
-void NonBlockingPushClient::Core::CreateOnDelegateThread(
-    const CreateBlockingPushClientCallback&
-        create_blocking_push_client_callback) {
-  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
-  delegate_push_client_ = create_blocking_push_client_callback.Run();
-  delegate_push_client_->AddObserver(this);
-}
-
-void NonBlockingPushClient::Core::DestroyOnDelegateThread() {
-  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
-  delegate_push_client_->RemoveObserver(this);
-  delegate_push_client_.reset();
-}
-
-void NonBlockingPushClient::Core::UpdateSubscriptions(
-    const SubscriptionList& subscriptions) {
-  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
-  DCHECK(delegate_push_client_.get());
-  delegate_push_client_->UpdateSubscriptions(subscriptions);
-}
-
-void NonBlockingPushClient::Core::UpdateCredentials(
-      const std::string& email, const std::string& token) {
-  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
-  DCHECK(delegate_push_client_.get());
-  delegate_push_client_->UpdateCredentials(email, token);
-}
-
-void NonBlockingPushClient::Core::SendNotification(
-    const Notification& notification) {
-  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
-  DCHECK(delegate_push_client_.get());
-  delegate_push_client_->SendNotification(notification);
-}
-
-void NonBlockingPushClient::Core::OnNotificationStateChange(
-    bool notifications_enabled) {
-  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
-  parent_task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&NonBlockingPushClient::OnNotificationStateChange,
-                 parent_push_client_, notifications_enabled));
-}
-
-void NonBlockingPushClient::Core::OnIncomingNotification(
-    const Notification& notification) {
-  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
-  parent_task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&NonBlockingPushClient::OnIncomingNotification,
-                 parent_push_client_, notification));
-}
-
-NonBlockingPushClient::NonBlockingPushClient(
-    const scoped_refptr<base::SingleThreadTaskRunner>& delegate_task_runner,
-    const CreateBlockingPushClientCallback&
-        create_blocking_push_client_callback)
-    : weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
-      delegate_task_runner_(delegate_task_runner),
-      core_(new Core(create_blocking_push_client_callback,
-                     delegate_task_runner_,
-                     weak_ptr_factory_.GetWeakPtr())) {}
-
-NonBlockingPushClient::~NonBlockingPushClient() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  delegate_task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&NonBlockingPushClient::Core::DestroyOnDelegateThread,
-                 core_.get()));
-}
-
-void NonBlockingPushClient::AddObserver(PushClientObserver* observer) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  observers_.AddObserver(observer);
-}
-
-void NonBlockingPushClient::RemoveObserver(PushClientObserver* observer) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  observers_.RemoveObserver(observer);
-}
-
-void NonBlockingPushClient::UpdateSubscriptions(
-    const SubscriptionList& subscriptions) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  delegate_task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&NonBlockingPushClient::Core::UpdateSubscriptions,
-                 core_.get(), subscriptions));
-}
-
-void NonBlockingPushClient::UpdateCredentials(
-    const std::string& email, const std::string& token) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  delegate_task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&NonBlockingPushClient::Core::UpdateCredentials,
-                 core_.get(), email, token));
-}
-
-void NonBlockingPushClient::SendNotification(
-    const Notification& notification) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  delegate_task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&NonBlockingPushClient::Core::SendNotification, core_.get(),
-                 notification));
-}
-
-void NonBlockingPushClient::OnNotificationStateChange(
-    bool notifications_enabled) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  FOR_EACH_OBSERVER(PushClientObserver, observers_,
-                    OnNotificationStateChange(notifications_enabled));
-}
-
-void NonBlockingPushClient::OnIncomingNotification(
-    const Notification& notification) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  FOR_EACH_OBSERVER(PushClientObserver, observers_,
-                    OnIncomingNotification(notification));
-}
-
-}  // namespace notifier
diff --git a/jingle/notifier/listener/non_blocking_push_client.h b/jingle/notifier/listener/non_blocking_push_client.h
deleted file mode 100644
index ca57229..0000000
--- a/jingle/notifier/listener/non_blocking_push_client.h
+++ /dev/null
@@ -1,70 +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.
-
-#ifndef JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_PUSH_CLIENT_H_
-#define JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_PUSH_CLIENT_H_
-
-#include "base/basictypes.h"
-#include "base/callback_forward.h"
-#include "base/compiler_specific.h"
-#include "base/memory/ref_counted.h"
-#include "base/memory/scoped_ptr.h"
-#include "base/memory/weak_ptr.h"
-#include "base/observer_list.h"
-#include "base/threading/non_thread_safe.h"
-#include "jingle/notifier/listener/push_client.h"
-
-namespace base {
-class SingleThreadTaskRunner;
-} // namespace base
-
-namespace notifier {
-
-// This class implements a PushClient that doesn't block; it delegates
-// to another blocking PushClient on a separate thread.
-//
-// This class must be used on a single thread.
-class NonBlockingPushClient : public PushClient {
- public:
-  // The type for a function that creates a (blocking) PushClient.
-  // Will be called on the delegate task runner.
-  typedef base::Callback<scoped_ptr<PushClient>()>
-      CreateBlockingPushClientCallback;
-
-  // Runs the given callback on the given task runner, and delegates
-  // to that PushClient.
-  explicit NonBlockingPushClient(
-      const scoped_refptr<base::SingleThreadTaskRunner>& delegate_task_runner,
-      const CreateBlockingPushClientCallback&
-          create_blocking_push_client_callback);
-  virtual ~NonBlockingPushClient();
-
-  // PushClient implementation.
-  virtual void AddObserver(PushClientObserver* observer) OVERRIDE;
-  virtual void RemoveObserver(PushClientObserver* observer) OVERRIDE;
-  virtual void UpdateSubscriptions(
-      const SubscriptionList& subscriptions) OVERRIDE;
-  virtual void UpdateCredentials(
-      const std::string& email, const std::string& token) OVERRIDE;
-  virtual void SendNotification(const Notification& notification) OVERRIDE;
-
- private:
-  class Core;
-
-  void OnNotificationStateChange(bool notifications_enabled);
-  void OnIncomingNotification(const Notification& notification);
-
-  base::NonThreadSafe non_thread_safe_;
-  base::WeakPtrFactory<NonBlockingPushClient> weak_ptr_factory_;
-  const scoped_refptr<base::SingleThreadTaskRunner> delegate_task_runner_;
-  const scoped_refptr<Core> core_;
-
-  ObserverList<PushClientObserver> observers_;
-
-  DISALLOW_COPY_AND_ASSIGN(NonBlockingPushClient);
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_PUSH_CLIENT_H_
diff --git a/jingle/notifier/listener/non_blocking_push_client_unittest.cc b/jingle/notifier/listener/non_blocking_push_client_unittest.cc
deleted file mode 100644
index c415225..0000000
--- a/jingle/notifier/listener/non_blocking_push_client_unittest.cc
+++ /dev/null
@@ -1,139 +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/non_blocking_push_client.h"
-
-#include <cstddef>
-
-#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/listener/fake_push_client.h"
-#include "jingle/notifier/listener/fake_push_client_observer.h"
-#include "jingle/notifier/listener/push_client_observer.h"
-#include "net/url_request/url_request_test_util.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-namespace notifier {
-
-namespace {
-
-class NonBlockingPushClientTest : public testing::Test {
- protected:
-  NonBlockingPushClientTest() : fake_push_client_(NULL) {}
-
-  virtual ~NonBlockingPushClientTest() {}
-
-  virtual void SetUp() OVERRIDE {
-    push_client_.reset(
-        new NonBlockingPushClient(
-            base::MessageLoopProxy::current(),
-            base::Bind(&NonBlockingPushClientTest::CreateFakePushClient,
-                       base::Unretained(this))));
-    push_client_->AddObserver(&fake_observer_);
-    // Pump message loop to run CreateFakePushClient.
-    message_loop_.RunAllPending();
-  }
-
-  virtual void TearDown() OVERRIDE {
-    // Clear out any pending notifications before removing observers.
-    message_loop_.RunAllPending();
-    push_client_->RemoveObserver(&fake_observer_);
-    push_client_.reset();
-    // Then pump message loop to run
-    // NonBlockingPushClient::DestroyOnDelegateThread().
-    message_loop_.RunAllPending();
-  }
-
-  scoped_ptr<PushClient> CreateFakePushClient() {
-    if (fake_push_client_) {
-      ADD_FAILURE();
-      return scoped_ptr<PushClient>();
-    }
-    fake_push_client_ = new FakePushClient();
-    return scoped_ptr<PushClient>(fake_push_client_);
-  }
-
-  MessageLoop message_loop_;
-  FakePushClientObserver fake_observer_;
-  scoped_ptr<NonBlockingPushClient> push_client_;
-  // Owned by |push_client_|.
-  FakePushClient* fake_push_client_;
-};
-
-// Make sure UpdateSubscriptions() gets delegated properly.
-TEST_F(NonBlockingPushClientTest, UpdateSubscriptions) {
-  SubscriptionList subscriptions(10);
-  subscriptions[0].channel = "channel";
-  subscriptions[9].from = "from";
-
-  push_client_->UpdateSubscriptions(subscriptions);
-  EXPECT_TRUE(fake_push_client_->subscriptions().empty());
-  message_loop_.RunAllPending();
-  EXPECT_TRUE(
-      SubscriptionListsEqual(
-          fake_push_client_->subscriptions(), subscriptions));
-}
-
-// Make sure UpdateCredentials() gets delegated properly.
-TEST_F(NonBlockingPushClientTest, UpdateCredentials) {
-  const char kEmail[] = "foo@bar.com";
-  const char kToken[] = "baz";
-
-  push_client_->UpdateCredentials(kEmail, kToken);
-  EXPECT_TRUE(fake_push_client_->email().empty());
-  EXPECT_TRUE(fake_push_client_->token().empty());
-  message_loop_.RunAllPending();
-  EXPECT_EQ(kEmail, fake_push_client_->email());
-  EXPECT_EQ(kToken, fake_push_client_->token());
-}
-
-Notification MakeTestNotification() {
-  Notification notification;
-  notification.channel = "channel";
-  notification.recipients.resize(10);
-  notification.recipients[0].to = "to";
-  notification.recipients[9].user_specific_data = "user_specific_data";
-  notification.data = "data";
-  return notification;
-}
-
-// Make sure SendNotification() gets delegated properly.
-TEST_F(NonBlockingPushClientTest, SendNotification) {
-  const Notification notification = MakeTestNotification();
-
-  push_client_->SendNotification(notification);
-  EXPECT_TRUE(fake_push_client_->sent_notifications().empty());
-  message_loop_.RunAllPending();
-  ASSERT_EQ(1u, fake_push_client_->sent_notifications().size());
-  EXPECT_TRUE(
-      fake_push_client_->sent_notifications()[0].Equals(notification));
-}
-
-// Make sure notification state changes get propagated back to the
-// parent.
-TEST_F(NonBlockingPushClientTest, NotificationStateChange) {
-  EXPECT_FALSE(fake_observer_.notifications_enabled());
-  fake_push_client_->SimulateNotificationStateChange(true);
-  message_loop_.RunAllPending();
-  EXPECT_TRUE(fake_observer_.notifications_enabled());
-  fake_push_client_->SimulateNotificationStateChange(false);
-  message_loop_.RunAllPending();
-  EXPECT_FALSE(fake_observer_.notifications_enabled());
-}
-
-// Make sure incoming notifications get propagated back to the parent.
-TEST_F(NonBlockingPushClientTest, OnIncomingNotification) {
-  const Notification notification = MakeTestNotification();
-
-  fake_push_client_->SimulateIncomingNotification(notification);
-  message_loop_.RunAllPending();
-  EXPECT_TRUE(
-      fake_observer_.last_incoming_notification().Equals(notification));
-}
-
-}  // namespace
-
-}  // namespace notifier
diff --git a/jingle/notifier/listener/notification_defines.cc b/jingle/notifier/listener/notification_defines.cc
index 2f65775..6d2041e 100644
--- a/jingle/notifier/listener/notification_defines.cc
+++ b/jingle/notifier/listener/notification_defines.cc
@@ -4,61 +4,11 @@
 
 #include "jingle/notifier/listener/notification_defines.h"
 
-#include <cstddef>
-
 namespace notifier {
 
-Subscription::Subscription() {}
-Subscription::~Subscription() {}
-
-bool Subscription::Equals(const Subscription& other) const {
-  return channel == other.channel && from == other.from;
-}
-
-namespace {
-
-template <typename T>
-bool ListsEqual(const T& t1, const T& t2) {
-  if (t1.size() != t2.size()) {
-    return false;
-  }
-  for (size_t i = 0; i < t1.size(); ++i) {
-    if (!t1[i].Equals(t2[i])) {
-      return false;
-    }
-  }
-  return true;
-}
-
-}  // namespace
-
-bool SubscriptionListsEqual(const SubscriptionList& subscriptions1,
-                            const SubscriptionList& subscriptions2) {
-  return ListsEqual(subscriptions1, subscriptions2);
-}
-
-Recipient::Recipient() {}
-Recipient::~Recipient() {}
-
-bool Recipient::Equals(const Recipient& other) const {
-  return to == other.to && user_specific_data == other.user_specific_data;
-}
-
-bool RecipientListsEqual(const RecipientList& recipients1,
-                         const RecipientList& recipients2) {
-  return ListsEqual(recipients1, recipients2);
-}
-
 Notification::Notification() {}
 Notification::~Notification() {}
 
-bool Notification::Equals(const Notification& other) const {
-  return
-      channel == other.channel &&
-      data == other.data &&
-      RecipientListsEqual(recipients, other.recipients);
-}
-
 std::string Notification::ToString() const {
   return "{ channel: \"" + channel + "\", data: \"" + data + "\" }";
 }
diff --git a/jingle/notifier/listener/notification_defines.h b/jingle/notifier/listener/notification_defines.h
index 5d4a0c8..f85288f 100644
--- a/jingle/notifier/listener/notification_defines.h
+++ b/jingle/notifier/listener/notification_defines.h
@@ -11,10 +11,6 @@
 namespace notifier {
 
 struct Subscription {
-  Subscription();
-  ~Subscription();
-  bool Equals(const Subscription& other) const;
-
   // The name of the channel to subscribe to; usually but not always
   // a URL.
   std::string channel;
@@ -25,15 +21,8 @@ struct Subscription {
 
 typedef std::vector<Subscription> SubscriptionList;
 
-bool SubscriptionListsEqual(const SubscriptionList& subscriptions1,
-                            const SubscriptionList& subscriptions2);
-
 // A structure representing a <recipient/> block within a push message.
 struct Recipient {
-  Recipient();
-  ~Recipient();
-  bool Equals(const Recipient& other) const;
-
   // The bare jid of the recipient.
   std::string to;
   // User-specific data for the recipient.
@@ -42,9 +31,6 @@ struct Recipient {
 
 typedef std::vector<Recipient> RecipientList;
 
-bool RecipientListsEqual(const RecipientList& recipients1,
-                         const RecipientList& recipients2);
-
 struct Notification {
   Notification();
   ~Notification();
@@ -56,7 +42,6 @@ struct Notification {
   // The notification data payload.
   std::string data;
 
-  bool Equals(const Notification& other) const;
   std::string ToString() const;
 };
 
diff --git a/jingle/notifier/listener/push_client.cc b/jingle/notifier/listener/push_client.cc
index d704264..8c82b44 100644
--- a/jingle/notifier/listener/push_client.cc
+++ b/jingle/notifier/listener/push_client.cc
@@ -4,31 +4,317 @@
 
 #include "jingle/notifier/listener/push_client.h"
 
-#include <cstddef>
-
 #include "base/bind.h"
-#include "base/single_thread_task_runner.h"
-#include "jingle/notifier/listener/non_blocking_push_client.h"
-#include "jingle/notifier/listener/xmpp_push_client.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::~PushClient() {}
+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()) {
+}
 
-namespace {
+PushClient::~PushClient() {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  io_message_loop_proxy_->PostTask(
+      FROM_HERE,
+      base::Bind(&PushClient::Core::DestroyOnIOThread, core_.get()));
+}
 
-scoped_ptr<PushClient> CreateXmppPushClient(
-    const NotifierOptions& notifier_options) {
-  return scoped_ptr<PushClient>(new XmppPushClient(notifier_options));
+void PushClient::AddObserver(Observer* observer) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  core_->AddObserver(observer);
 }
 
-}  // namespace
+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()));
+}
 
-scoped_ptr<PushClient> PushClient::CreateDefault(
-    const NotifierOptions& notifier_options) {
-  return scoped_ptr<PushClient>(new NonBlockingPushClient(
-      notifier_options.request_context_getter->GetIOMessageLoopProxy(),
-      base::Bind(&CreateXmppPushClient, notifier_options)));
+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
index ea975a8..fdac8ba 100644
--- a/jingle/notifier/listener/push_client.h
+++ b/jingle/notifier/listener/push_client.h
@@ -6,44 +6,91 @@
 #define JINGLE_NOTIFIER_LISTENER_PUSH_CLIENT_H_
 
 #include <string>
+#include <vector>
 
-#include "base/memory/scoped_ptr.h"
+#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 notifier {
+namespace base {
+class MessageLoopProxy;
+} // namespace base
+
+namespace buzz {
+class XmppTaskParentInterface;
+}  // namespace buzz
 
-struct NotifierOptions;
-class PushClientObserver;
+namespace notifier {
 
-// A PushClient is an interface for classes that implement a push
-// mechanism, where a client can push notifications to and receive
-// notifications from other clients.
+// This class implements a client for the XMPP google:push protocol.
+//
+// This class must be used on a single thread.
 class PushClient {
  public:
-  virtual ~PushClient();
+  // 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;
 
-  // Creates a default non-blocking PushClient implementation from the
-  // given options.
-  static scoped_ptr<PushClient> CreateDefault(
-      const NotifierOptions& notifier_options);
+   protected:
+    virtual ~Observer();
+  };
 
-  // Manage the list of observers for incoming notifications.
-  virtual void AddObserver(PushClientObserver* observer) = 0;
-  virtual void RemoveObserver(PushClientObserver* observer) = 0;
+  explicit PushClient(const NotifierOptions& notifier_options);
+  ~PushClient();
 
-  // Implementors are required to have this take effect only on the
-  // next (re-)connection.  Therefore, clients should call this before
-  // UpdateCredentials().
-  virtual void UpdateSubscriptions(const SubscriptionList& subscriptions) = 0;
+  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.
-  virtual void UpdateCredentials(
-      const std::string& email, const std::string& token) = 0;
+  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_;
 
-  // Sends a notification (with no reliability guarantees).
-  virtual void SendNotification(const Notification& notification) = 0;
+  DISALLOW_COPY_AND_ASSIGN(PushClient);
 };
 
 }  // namespace notifier
diff --git a/jingle/notifier/listener/push_client_observer.cc b/jingle/notifier/listener/push_client_observer.cc
deleted file mode 100644
index 04a877b..0000000
--- a/jingle/notifier/listener/push_client_observer.cc
+++ /dev/null
@@ -1,11 +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/push_client_observer.h"
-
-namespace notifier {
-
-PushClientObserver::~PushClientObserver() {}
-
-}  // namespace notifier
diff --git a/jingle/notifier/listener/push_client_observer.h b/jingle/notifier/listener/push_client_observer.h
deleted file mode 100644
index 0fe7d46..0000000
--- a/jingle/notifier/listener/push_client_observer.h
+++ /dev/null
@@ -1,32 +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.
-
-#ifndef JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_PUSH_CLIENT_OBSERVER_H_
-#define JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_PUSH_CLIENT_OBSERVER_H_
-
-#include "jingle/notifier/listener/notification_defines.h"
-
-namespace notifier {
-
-// A PushClientObserver is notified whenever an incoming notification
-// is received or when the state of the push client changes.
-class PushClientObserver {
- protected:
-  virtual ~PushClientObserver();
-
- 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;
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_NON_BLOCKING_PUSH_CLIENT_OBSERVER_H_
diff --git a/jingle/notifier/listener/push_client_unittest.cc b/jingle/notifier/listener/push_client_unittest.cc
index fd45466..a5663d7 100644
--- a/jingle/notifier/listener/push_client_unittest.cc
+++ b/jingle/notifier/listener/push_client_unittest.cc
@@ -4,20 +4,31 @@
 
 #include "jingle/notifier/listener/push_client.h"
 
-#include "base/bind_helpers.h"
 #include "base/compiler_specific.h"
-#include "base/location.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop.h"
-#include "base/threading/thread.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() {
@@ -27,28 +38,68 @@ class PushClientTest : public testing::Test {
 
   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_;
 };
 
-// Make sure calling CreateDefault on the IO thread doesn't blow up.
-TEST_F(PushClientTest, OnIOThread) {
-  const scoped_ptr<PushClient> push_client(
-      PushClient::CreateDefault(notifier_options_));
+TEST_F(PushClientTest, OnIncomingNotification) {
+  EXPECT_CALL(mock_observer_, OnIncomingNotification(_));
+  push_client_->SimulateOnNotificationReceivedForTest(Notification());
 }
 
-// Make sure calling CreateDefault on a non-IO thread doesn't blow up.
-TEST_F(PushClientTest, OffIOThread) {
-  base::Thread thread("Non-IO thread");
-  EXPECT_TRUE(thread.Start());
-  thread.message_loop()->PostTask(
-      FROM_HERE,
-      base::Bind(base::IgnoreResult(&PushClient::CreateDefault),
-                 notifier_options_));
-  thread.Stop();
+TEST_F(PushClientTest, ConnectAndSubscribe) {
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
+  push_client_->SimulateConnectAndSubscribeForTest(
+      fake_base_task_.AsWeakPtr());
 }
 
-}  // namespace
+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/xmpp_push_client.cc b/jingle/notifier/listener/xmpp_push_client.cc
deleted file mode 100644
index dcf828d8..0000000
--- a/jingle/notifier/listener/xmpp_push_client.cc
+++ /dev/null
@@ -1,138 +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/xmpp_push_client.h"
-
-#include "base/logging.h"
-#include "base/message_loop_proxy.h"
-#include "jingle/notifier/base/notifier_options_util.h"
-#include "jingle/notifier/listener/push_client_observer.h"
-#include "jingle/notifier/listener/push_notifications_send_update_task.h"
-
-namespace notifier {
-
-XmppPushClient::XmppPushClient(const NotifierOptions& notifier_options)
-    : notifier_options_(notifier_options) {
-  DCHECK(notifier_options_.request_context_getter->
-         GetIOMessageLoopProxy()->BelongsToCurrentThread());
-}
-
-XmppPushClient::~XmppPushClient() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-}
-
-void XmppPushClient::OnConnect(
-    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  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 XmppPushClient::OnDisconnect() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  base_task_.reset();
-  FOR_EACH_OBSERVER(PushClientObserver, observers_,
-                    OnNotificationStateChange(false));
-}
-
-void XmppPushClient::OnNotificationReceived(
-    const Notification& notification) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  FOR_EACH_OBSERVER(PushClientObserver, observers_,
-                    OnIncomingNotification(notification));
-}
-
-void XmppPushClient::OnSubscribed() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  FOR_EACH_OBSERVER(PushClientObserver, observers_,
-                    OnNotificationStateChange(true));
-}
-
-void XmppPushClient::OnSubscriptionError() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  FOR_EACH_OBSERVER(PushClientObserver, observers_,
-                    OnNotificationStateChange(false));
-}
-
-void XmppPushClient::AddObserver(PushClientObserver* observer) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  observers_.AddObserver(observer);
-}
-
-void XmppPushClient::RemoveObserver(PushClientObserver* observer) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  observers_.RemoveObserver(observer);
-}
-
-void XmppPushClient::UpdateSubscriptions(
-    const SubscriptionList& subscriptions) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  subscriptions_ = subscriptions;
-}
-
-void XmppPushClient::UpdateCredentials(
-      const std::string& email, const std::string& token) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  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 XmppPushClient::SendNotification(const Notification& notification) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  if (!base_task_.get()) {
-    // TODO(akalin): Figure out whether we really need to do this.
-    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();
-}
-
-}  // namespace notifier
diff --git a/jingle/notifier/listener/xmpp_push_client.h b/jingle/notifier/listener/xmpp_push_client.h
deleted file mode 100644
index 9c4a6f3..0000000
--- a/jingle/notifier/listener/xmpp_push_client.h
+++ /dev/null
@@ -1,87 +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.
-
-#ifndef JINGLE_NOTIFIER_LISTENER_XMPP_PUSH_CLIENT_H_
-#define JINGLE_NOTIFIER_LISTENER_XMPP_PUSH_CLIENT_H_
-
-#include <string>
-#include <vector>
-
-#include "base/basictypes.h"
-#include "base/compiler_specific.h"
-#include "base/memory/scoped_ptr.h"
-#include "base/memory/ref_counted.h"
-#include "base/memory/weak_ptr.h"
-#include "base/observer_list.h"
-#include "base/threading/non_thread_safe.h"
-#include "jingle/notifier/base/notifier_options.h"
-#include "jingle/notifier/communicator/login.h"
-#include "jingle/notifier/listener/notification_defines.h"
-#include "jingle/notifier/listener/push_client.h"
-#include "jingle/notifier/listener/push_notifications_listen_task.h"
-#include "jingle/notifier/listener/push_notifications_subscribe_task.h"
-#include "talk/xmpp/xmppclientsettings.h"
-
-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 XmppPushClient :
-      public PushClient,
-      public LoginDelegate,
-      public PushNotificationsListenTaskDelegate,
-      public PushNotificationsSubscribeTaskDelegate {
- public:
-  explicit XmppPushClient(const NotifierOptions& notifier_options);
-  virtual ~XmppPushClient();
-
-  // PushClient implementation.
-  virtual void AddObserver(PushClientObserver* observer) OVERRIDE;
-  virtual void RemoveObserver(PushClientObserver* observer) OVERRIDE;
-  virtual void UpdateSubscriptions(
-      const SubscriptionList& subscriptions) OVERRIDE;
-  virtual void UpdateCredentials(
-      const std::string& email, const std::string& token) OVERRIDE;
-  virtual void SendNotification(const Notification& notification) OVERRIDE;
-
-  // Login::Delegate implementation.
-  virtual void OnConnect(
-      base::WeakPtr<buzz::XmppTaskParentInterface> base_task) OVERRIDE;
-  virtual void OnDisconnect() OVERRIDE;
-
-  // PushNotificationsListenTaskDelegate implementation.
-  virtual void OnNotificationReceived(
-      const Notification& notification) OVERRIDE;
-
-  // PushNotificationsSubscribeTaskDelegate implementation.
-  virtual void OnSubscribed() OVERRIDE;
-  virtual void OnSubscriptionError() OVERRIDE;
-
- private:
-  base::NonThreadSafe non_thread_safe_;
-  const NotifierOptions notifier_options_;
-  ObserverList<PushClientObserver> observers_;
-
-  // XMPP connection settings.
-  SubscriptionList subscriptions_;
-  buzz::XmppClientSettings xmpp_settings_;
-
-  scoped_ptr<notifier::Login> login_;
-
-  // The XMPP connection.
-  base::WeakPtr<buzz::XmppTaskParentInterface> base_task_;
-
-  std::vector<Notification> pending_notifications_to_send_;
-
-  DISALLOW_COPY_AND_ASSIGN(XmppPushClient);
-};
-
-}  // namespace notifier
-
-#endif  // JINGLE_NOTIFIER_LISTENER_XMPP_PUSH_CLIENT_H_
diff --git a/jingle/notifier/listener/xmpp_push_client_unittest.cc b/jingle/notifier/listener/xmpp_push_client_unittest.cc
deleted file mode 100644
index 6a92bcf..0000000
--- a/jingle/notifier/listener/xmpp_push_client_unittest.cc
+++ /dev/null
@@ -1,120 +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/xmpp_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 "jingle/notifier/listener/push_client_observer.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 PushClientObserver {
- public:
-  MOCK_METHOD1(OnNotificationStateChange, void(bool));
-  MOCK_METHOD1(OnIncomingNotification, void(const Notification&));
-};
-
-class XmppPushClientTest : public testing::Test {
- protected:
-  XmppPushClientTest() {
-    notifier_options_.request_context_getter =
-        new TestURLRequestContextGetter(message_loop_.message_loop_proxy());
-  }
-
-  virtual ~XmppPushClientTest() {}
-
-  virtual void SetUp() OVERRIDE {
-    xmpp_push_client_.reset(new XmppPushClient(notifier_options_));
-    xmpp_push_client_->AddObserver(&mock_observer_);
-  }
-
-  virtual void TearDown() OVERRIDE {
-    // Clear out any messages posted by XmppPushClient.
-    message_loop_.RunAllPending();
-    xmpp_push_client_->RemoveObserver(&mock_observer_);
-    xmpp_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<XmppPushClient> xmpp_push_client_;
-  FakeBaseTask fake_base_task_;
-};
-
-// Make sure the XMPP push client notifies its observers of incoming
-// notifications properly.
-TEST_F(XmppPushClientTest, OnIncomingNotification) {
-  EXPECT_CALL(mock_observer_, OnIncomingNotification(_));
-  xmpp_push_client_->OnNotificationReceived(Notification());
-}
-
-// Make sure the XMPP push client notifies its observers of a
-// successful connection properly.
-TEST_F(XmppPushClientTest, ConnectAndSubscribe) {
-  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
-  xmpp_push_client_->OnConnect(fake_base_task_.AsWeakPtr());
-  xmpp_push_client_->OnSubscribed();
-}
-
-// Make sure the XMPP push client notifies its observers of a
-// terminated connection properly.
-TEST_F(XmppPushClientTest, Disconnect) {
-  EXPECT_CALL(mock_observer_, OnNotificationStateChange(false));
-  xmpp_push_client_->OnDisconnect();
-}
-
-// Make sure the XMPP push client notifies its observers of a
-// subscription error properly.
-TEST_F(XmppPushClientTest, SubscriptionError) {
-  EXPECT_CALL(mock_observer_, OnNotificationStateChange(false));
-  xmpp_push_client_->OnSubscriptionError();
-}
-
-// Make sure nothing blows up when the XMPP push client sends a
-// notification.
-//
-// TODO(akalin): Figure out how to test that the notification was
-// actually sent.
-TEST_F(XmppPushClientTest, SendNotification) {
-  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
-
-  xmpp_push_client_->OnConnect(fake_base_task_.AsWeakPtr());
-  xmpp_push_client_->OnSubscribed();
-  xmpp_push_client_->SendNotification(Notification());
-}
-
-// Make sure nothing blows up when the XMPP push client sends a
-// notification when disconnected, and the client connects.
-//
-// TODO(akalin): Figure out how to test that the notification was
-// actually sent.
-TEST_F(XmppPushClientTest, SendNotificationPending) {
-  xmpp_push_client_->SendNotification(Notification());
-
-  Mock::VerifyAndClearExpectations(&mock_observer_);
-
-  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
-
-  xmpp_push_client_->OnConnect(fake_base_task_.AsWeakPtr());
-  xmpp_push_client_->OnSubscribed();
-}
-
-}  // namespace
-
-}  // namespace notifier