[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

chrome/service/cloud_print/cloud_print_proxy_backend.cc

@@ -26,6 +26,7 @@
 #include "grit/generated_resources.h"
 #include "jingle/notifier/base/notifier_options.h"
 #include "jingle/notifier/listener/push_client.h"
+#include "jingle/notifier/listener/push_client_observer.h"
 #include "ui/base/l10n/l10n_util.h"
 
 // The real guts of CloudPrintProxyBackend, to keep the public client API clean.
@@ -33,7 +34,7 @@ class CloudPrintProxyBackend::Core
     : public base::RefCountedThreadSafe<CloudPrintProxyBackend::Core>,
       public CloudPrintAuth::Client,
       public CloudPrintConnector::Client,
-      public notifier::PushClient::Observer {
+      public notifier::PushClientObserver {
  public:
   // It is OK for print_server_url to be empty. In this case system should
   // use system default (local) print server.
@@ -86,7 +87,7 @@ class CloudPrintProxyBackend::Core
   // CloudPrintConnector::Client implementation.
   virtual void OnAuthFailed() OVERRIDE;
 
-  // notifier::PushClient::Delegate implementation.
+  // notifier::PushClientObserver implementation.
   virtual void OnNotificationStateChange(
       bool notifications_enabled) OVERRIDE;
   virtual void OnIncomingNotification(
@@ -390,7 +391,7 @@ void CloudPrintProxyBackend::Core::InitNotifications(
   notifier_options.request_context_getter =
       g_service_process->GetServiceURLRequestContextGetter();
   notifier_options.auth_mechanism = "X-OAUTH2";
-  push_client_.reset(new notifier::PushClient(notifier_options));
+  push_client_ = notifier::PushClient::CreateDefault(notifier_options);
   push_client_->AddObserver(this);
   notifier::Subscription subscription;
   subscription.channel = kCloudPrintPushNotificationsSource;

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': [

jingle/notifier/listener/fake_push_client.cc

+// 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
+

jingle/notifier/listener/fake_push_client.h

+// 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_

jingle/notifier/listener/fake_push_client_observer.cc

+// 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
+

jingle/notifier/listener/fake_push_client_observer.h

+// 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_

jingle/notifier/listener/non_blocking_push_client.cc

+// 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

jingle/notifier/listener/non_blocking_push_client.h

+// 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_

jingle/notifier/listener/non_blocking_push_client_unittest.cc

+// 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

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 + "\" }";
 }

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;
 };
 

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;
+  virtual ~PushClient();
 
-    // 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();
-
-  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

jingle/notifier/listener/push_client_observer.cc

+// 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

jingle/notifier/listener/push_client_observer.h

+// 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_

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

jingle/notifier/listener/xmpp_push_client.cc

+// 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

jingle/notifier/listener/xmpp_push_client.h

+// 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_

jingle/notifier/listener/xmpp_push_client_unittest.cc

+// 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

sync/notifier/p2p_notifier.cc

@@ -9,8 +9,8 @@
 #include "base/json/json_reader.h"
 #include "base/json/json_writer.h"
 #include "base/logging.h"
-#include "base/message_loop_proxy.h"
 #include "base/values.h"
+#include "jingle/notifier/listener/push_client.h"
 #include "sync/notifier/sync_notifier_observer.h"
 #include "sync/syncable/model_type_payload_map.h"
 
@@ -140,63 +140,62 @@ bool P2PNotificationData::ResetFromString(const std::string& str) {
   return true;
 }
 
-P2PNotifier::P2PNotifier(const notifier::NotifierOptions& notifier_options,
+P2PNotifier::P2PNotifier(scoped_ptr<notifier::PushClient> push_client,
                          P2PNotificationTarget send_notification_target)
-    : push_client_(notifier_options),
+    : push_client_(push_client.Pass()),
       logged_in_(false),
       notifications_enabled_(false),
-      send_notification_target_(send_notification_target),
-      parent_message_loop_proxy_(base::MessageLoopProxy::current()) {
+      send_notification_target_(send_notification_target) {
   DCHECK(send_notification_target_ == NOTIFY_OTHERS ||
          send_notification_target_ == NOTIFY_ALL);
-  push_client_.AddObserver(this);
+  push_client_->AddObserver(this);
 }
 
 P2PNotifier::~P2PNotifier() {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  push_client_.RemoveObserver(this);
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  push_client_->RemoveObserver(this);
 }
 
 void P2PNotifier::AddObserver(SyncNotifierObserver* observer) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   observer_list_.AddObserver(observer);
 }
 
 void P2PNotifier::RemoveObserver(SyncNotifierObserver* observer) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   observer_list_.RemoveObserver(observer);
 }
 
 void P2PNotifier::SetUniqueId(const std::string& unique_id) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   unique_id_ = unique_id;
 }
 
 void P2PNotifier::SetState(const std::string& state) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  // Do nothing.
 }
 
 void P2PNotifier::UpdateCredentials(
     const std::string& email, const std::string& token) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   notifier::Subscription subscription;
   subscription.channel = kSyncP2PNotificationChannel;
   // There may be some subtle issues around case sensitivity of the
   // from field, but it doesn't matter too much since this is only
   // used in p2p mode (which is only used in testing).
   subscription.from = email;
-  push_client_.UpdateSubscriptions(
+  push_client_->UpdateSubscriptions(
       notifier::SubscriptionList(1, subscription));
-
   // If already logged in, the new credentials will take effect on the
   // next reconnection.
-  push_client_.UpdateCredentials(email, token);
+  push_client_->UpdateCredentials(email, token);
   logged_in_ = true;
 }
 
 void P2PNotifier::UpdateEnabledTypes(
     syncable::ModelTypeSet enabled_types) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   const syncable::ModelTypeSet new_enabled_types =
       Difference(enabled_types, enabled_types_);
   enabled_types_ = enabled_types;
@@ -207,14 +206,14 @@ void P2PNotifier::UpdateEnabledTypes(
 
 void P2PNotifier::SendNotification(
     syncable::ModelTypeSet changed_types) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   const P2PNotificationData notification_data(
       unique_id_, send_notification_target_, changed_types);
   SendNotificationData(notification_data);
 }
 
 void P2PNotifier::OnNotificationStateChange(bool notifications_enabled) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   bool disabled_to_enabled = notifications_enabled && !notifications_enabled_;
   notifications_enabled_ = notifications_enabled;
   FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
@@ -228,7 +227,7 @@ void P2PNotifier::OnNotificationStateChange(bool notifications_enabled) {
 
 void P2PNotifier::OnIncomingNotification(
     const notifier::Notification& notification) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   DVLOG(1) << "Received notification " << notification.ToString();
   if (!logged_in_) {
     DVLOG(1) << "Not logged in yet -- not emitting notification";
@@ -265,31 +264,20 @@ void P2PNotifier::OnIncomingNotification(
                     OnIncomingNotification(type_payloads, REMOTE_NOTIFICATION));
 }
 
-void P2PNotifier::SimulateConnectForTest(
-    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  push_client_.SimulateConnectAndSubscribeForTest(base_task);
-}
-
-void P2PNotifier::ReflectSentNotificationsForTest() {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  push_client_.ReflectSentNotificationsForTest();
-}
-
 void P2PNotifier::SendNotificationDataForTest(
     const P2PNotificationData& notification_data) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   SendNotificationData(notification_data);
 }
 
 void P2PNotifier::SendNotificationData(
     const P2PNotificationData& notification_data) {
-  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  DCHECK(non_thread_safe_.CalledOnValidThread());
   notifier::Notification notification;
   notification.channel = kSyncP2PNotificationChannel;
   notification.data = notification_data.ToString();
   DVLOG(1) << "Sending XMPP notification: " << notification.ToString();
-  push_client_.SendNotification(notification);
+  push_client_->SendNotification(notification);
 }
 
 }  // namespace sync_notifier

sync/notifier/p2p_notifier.h

@@ -15,18 +15,14 @@
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/observer_list.h"
-#include "jingle/notifier/base/notifier_options.h"
-#include "jingle/notifier/listener/push_client.h"
+#include "base/threading/non_thread_safe.h"
+#include "jingle/notifier/listener/push_client_observer.h"
 #include "sync/notifier/sync_notifier.h"
 #include "sync/syncable/model_type.h"
 
-namespace base {
-class MessageLoopProxy;
-}
-
-namespace buzz {
-class XmppTaskParentInterface;
-}  // namespace buzz
+namespace notifier {
+class PushClient;
+}  // namespace notifier
 
 namespace sync_notifier {
 
@@ -86,14 +82,14 @@ class P2PNotificationData {
 
 class P2PNotifier
     : public SyncNotifier,
-      public notifier::PushClient::Observer {
+      public notifier::PushClientObserver {
  public:
   // The |send_notification_target| parameter was added to allow us to send
   // self-notifications in some cases, but not others.  The value should be
   // either NOTIFY_ALL to send notifications to all clients, or NOTIFY_OTHERS
   // to send notifications to all clients except for the one that triggered the
   // notification.  See crbug.com/97780.
-  P2PNotifier(const notifier::NotifierOptions& notifier_options,
+  P2PNotifier(scoped_ptr<notifier::PushClient> push_client,
               P2PNotificationTarget send_notification_target);
 
   virtual ~P2PNotifier();
@@ -110,30 +106,23 @@ class P2PNotifier
   virtual void SendNotification(
       syncable::ModelTypeSet changed_types) OVERRIDE;
 
-  // PushClient::Delegate implementation.
+  // PushClientObserver implementation.
   virtual void OnNotificationStateChange(bool notifications_enabled) OVERRIDE;
   virtual void OnIncomingNotification(
       const notifier::Notification& notification) OVERRIDE;
 
-  // For testing.
-
-  void SimulateConnectForTest(
-      base::WeakPtr<buzz::XmppTaskParentInterface> base_task);
-
-  // Any notifications sent after this is called will be reflected,
-  // i.e. will be treated as an incoming notification also.
-  void ReflectSentNotificationsForTest();
-
   void SendNotificationDataForTest(
       const P2PNotificationData& notification_data);
 
  private:
   void SendNotificationData(const P2PNotificationData& notification_data);
 
+  base::NonThreadSafe non_thread_safe_;
+
   ObserverList<SyncNotifierObserver> observer_list_;
 
-  // The XMPP push client.
-  notifier::PushClient push_client_;
+  // The push client.
+  scoped_ptr<notifier::PushClient> push_client_;
   // Our unique ID.
   std::string unique_id_;
   // Whether we have called UpdateCredentials() yet.
@@ -145,8 +134,8 @@ class P2PNotifier
   P2PNotificationTarget send_notification_target_;
 
   syncable::ModelTypeSet enabled_types_;
-  scoped_refptr<base::MessageLoopProxy> parent_message_loop_proxy_;
 };
 
 }  // namespace sync_notifier
+
 #endif  // SYNC_NOTIFIER_P2P_NOTIFIER_H_

sync/notifier/sync_notifier_factory.cc

@@ -7,6 +7,7 @@
 #include <string>
 
 #include "base/logging.h"
+#include "jingle/notifier/listener/push_client.h"
 #include "sync/notifier/non_blocking_invalidation_notifier.h"
 #include "sync/notifier/p2p_notifier.h"
 #include "sync/notifier/sync_notifier.h"
@@ -25,7 +26,9 @@ SyncNotifier* CreateDefaultSyncNotifier(
     // NOTIFY_OTHERS.  There's no good reason to notify ourselves of our own
     // commits.  We self-notify for now only because the integration tests rely
     // on this behaviour.  See crbug.com/97780.
-    return new P2PNotifier(notifier_options, NOTIFY_ALL);
+    return new P2PNotifier(
+        notifier::PushClient::CreateDefault(notifier_options),
+        NOTIFY_ALL);
   }
 
   return new NonBlockingInvalidationNotifier(