[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=

Committed: https://src.chromium.org/viewvc/chrome?view=rev&revision=138216

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

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

18 files changed, 1126 insertions, 442 deletions

diff --git a/jingle/jingle.gyp b/jingle/jingle.gyp
index 9f9af8b..3d424a0 100644
--- a/jingle/jingle.gyp
+++ b/jingle/jingle.gyp
@@ -75,12 +75,16 @@
         'notifier/communicator/login_settings.h',
         'notifier/communicator/single_login_attempt.cc',
         'notifier/communicator/single_login_attempt.h',
-        'notifier/listener/push_client.cc',
-        'notifier/listener/push_client.h',
+        'notifier/listener/non_blocking_push_client.cc',
+        'notifier/listener/non_blocking_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',
@@ -89,6 +93,8 @@
         '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',
@@ -120,6 +126,10 @@
         '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',
@@ -160,10 +170,12 @@
         '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
new file mode 100644
index 0000000..cfe4a99
--- /dev/null
+++ b/jingle/notifier/listener/fake_push_client.cc
@@ -0,0 +1,67 @@
+// 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
new file mode 100644
index 0000000..61ba72b
--- /dev/null
+++ b/jingle/notifier/listener/fake_push_client.h
@@ -0,0 +1,55 @@
+// 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
new file mode 100644
index 0000000..ec1b9e1
--- /dev/null
+++ b/jingle/notifier/listener/fake_push_client_observer.cc
@@ -0,0 +1,34 @@
+// 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
new file mode 100644
index 0000000..b8ee6d6
--- /dev/null
+++ b/jingle/notifier/listener/fake_push_client_observer.h
@@ -0,0 +1,35 @@
+// 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
new file mode 100644
index 0000000..23a8402
--- /dev/null
+++ b/jingle/notifier/listener/non_blocking_push_client.cc
@@ -0,0 +1,203 @@
+// 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/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 scoped_refptr<base::SingleThreadTaskRunner>&
+          delegate_task_runner,
+      const base::WeakPtr<NonBlockingPushClient>& parent_push_client);
+
+  // Must be called after being created.
+  //
+  // This is separated out from the constructor since posting tasks
+  // from the constructor is dangerous.
+  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 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) {}
+
+NonBlockingPushClient::Core::~Core() {
+  DCHECK(parent_task_runner_->BelongsToCurrentThread() ||
+         delegate_task_runner_->BelongsToCurrentThread());
+  DCHECK(!delegate_push_client_.get());
+}
+
+void NonBlockingPushClient::Core::CreateOnDelegateThread(
+    const CreateBlockingPushClientCallback&
+        create_blocking_push_client_callback) {
+  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
+  DCHECK(!delegate_push_client_.get());
+  delegate_push_client_ = create_blocking_push_client_callback.Run();
+  delegate_push_client_->AddObserver(this);
+}
+
+void NonBlockingPushClient::Core::DestroyOnDelegateThread() {
+  DCHECK(delegate_task_runner_->BelongsToCurrentThread());
+  DCHECK(delegate_push_client_.get());
+  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(delegate_task_runner_,
+                     weak_ptr_factory_.GetWeakPtr())) {
+  delegate_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&NonBlockingPushClient::Core::CreateOnDelegateThread,
+                 core_.get(), create_blocking_push_client_callback));
+}
+
+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
new file mode 100644
index 0000000..ca57229
--- /dev/null
+++ b/jingle/notifier/listener/non_blocking_push_client.h
@@ -0,0 +1,70 @@
+// 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
new file mode 100644
index 0000000..c415225
--- /dev/null
+++ b/jingle/notifier/listener/non_blocking_push_client_unittest.cc
@@ -0,0 +1,139 @@
+// 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 6d2041e..2f65775 100644
--- a/jingle/notifier/listener/notification_defines.cc
+++ b/jingle/notifier/listener/notification_defines.cc
@@ -4,11 +4,61 @@
 
 #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 f85288f..5d4a0c8 100644
--- a/jingle/notifier/listener/notification_defines.h
+++ b/jingle/notifier/listener/notification_defines.h
@@ -11,6 +11,10 @@
 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;
@@ -21,8 +25,15 @@ 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.
@@ -31,6 +42,9 @@ struct Recipient {
 
 typedef std::vector<Recipient> RecipientList;
 
+bool RecipientListsEqual(const RecipientList& recipients1,
+                         const RecipientList& recipients2);
+
 struct Notification {
   Notification();
   ~Notification();
@@ -42,6 +56,7 @@ 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 8c82b44..d704264 100644
--- a/jingle/notifier/listener/push_client.cc
+++ b/jingle/notifier/listener/push_client.cc
@@ -4,317 +4,31 @@
 
 #include "jingle/notifier/listener/push_client.h"
 
+#include <cstddef>
+
 #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"
+#include "base/single_thread_task_runner.h"
+#include "jingle/notifier/listener/non_blocking_push_client.h"
+#include "jingle/notifier/listener/xmpp_push_client.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() {}
 
-PushClient::~PushClient() {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&PushClient::Core::DestroyOnIOThread, core_.get()));
-}
+namespace {
 
-void PushClient::AddObserver(Observer* observer) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  core_->AddObserver(observer);
+scoped_ptr<PushClient> CreateXmppPushClient(
+    const NotifierOptions& notifier_options) {
+  return scoped_ptr<PushClient>(new XmppPushClient(notifier_options));
 }
 
-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()));
-}
+}  // namespace
 
-void PushClient::ReflectSentNotificationsForTest() {
-  io_message_loop_proxy_->PostTask(
-      FROM_HERE,
-      base::Bind(&PushClient::Core::ReflectSentNotificationsForTest,
-                 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)));
 }
 
 }  // namespace notifier
diff --git a/jingle/notifier/listener/push_client.h b/jingle/notifier/listener/push_client.h
index fdac8ba..ea975a8 100644
--- a/jingle/notifier/listener/push_client.h
+++ b/jingle/notifier/listener/push_client.h
@@ -6,91 +6,44 @@
 #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 "base/memory/scoped_ptr.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.
+struct NotifierOptions;
+class PushClientObserver;
+
+// 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.
 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;
+  virtual ~PushClient();
 
-   protected:
-    virtual ~Observer();
-  };
+  // Creates a default non-blocking PushClient implementation from the
+  // given options.
+  static scoped_ptr<PushClient> CreateDefault(
+      const NotifierOptions& notifier_options);
 
-  explicit PushClient(const NotifierOptions& notifier_options);
-  ~PushClient();
+  // Manage the list of observers for incoming notifications.
+  virtual void AddObserver(PushClientObserver* observer) = 0;
+  virtual void RemoveObserver(PushClientObserver* observer) = 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);
+  // 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;
 
   // 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_;
+  virtual void UpdateCredentials(
+      const std::string& email, const std::string& token) = 0;
 
-  DISALLOW_COPY_AND_ASSIGN(PushClient);
+  // Sends a notification (with no reliability guarantees).
+  virtual void SendNotification(const Notification& notification) = 0;
 };
 
 }  // namespace notifier
diff --git a/jingle/notifier/listener/push_client_observer.cc b/jingle/notifier/listener/push_client_observer.cc
new file mode 100644
index 0000000..04a877b
--- /dev/null
+++ b/jingle/notifier/listener/push_client_observer.cc
@@ -0,0 +1,11 @@
+// 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
new file mode 100644
index 0000000..0fe7d46
--- /dev/null
+++ b/jingle/notifier/listener/push_client_observer.h
@@ -0,0 +1,32 @@
+// 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 a5663d7..fd45466 100644
--- a/jingle/notifier/listener/push_client_unittest.cc
+++ b/jingle/notifier/listener/push_client_unittest.cc
@@ -4,31 +4,20 @@
 
 #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 "jingle/notifier/base/fake_base_task.h"
+#include "base/threading/thread.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() {
@@ -38,68 +27,28 @@ 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_;
 };
 
-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());
+// 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, Disconnect) {
-  EXPECT_CALL(mock_observer_, OnNotificationStateChange(false));
-  push_client_->SimulateDisconnectForTest();
+// 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, 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
 
 }  // namespace notifier
diff --git a/jingle/notifier/listener/xmpp_push_client.cc b/jingle/notifier/listener/xmpp_push_client.cc
new file mode 100644
index 0000000..dcf828d8
--- /dev/null
+++ b/jingle/notifier/listener/xmpp_push_client.cc
@@ -0,0 +1,138 @@
+// 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
new file mode 100644
index 0000000..9c4a6f3
--- /dev/null
+++ b/jingle/notifier/listener/xmpp_push_client.h
@@ -0,0 +1,87 @@
+// 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
new file mode 100644
index 0000000..6a92bcf
--- /dev/null
+++ b/jingle/notifier/listener/xmpp_push_client_unittest.cc
@@ -0,0 +1,120 @@
+// 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