[Sync] Make InvalidationNotifier use PushClient

This removes a lot of duplicated code and unifies the two main
SyncNotifier implementations (P2P and Invalidation) to use the same
underlying transport layer.

Replace CacheInvalidationPacketHandler and ChromeNetwork with PushClientChannel.

Add PushClient::CreateDefaultOnIOThread which creates
an XmppPushClient directly.

BUG=76764
TEST=


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

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

21 files changed, 594 insertions, 492 deletions

diff --git a/jingle/notifier/communicator/single_login_attempt.cc b/jingle/notifier/communicator/single_login_attempt.cc
index b4cae89..0c789ee 100644
--- a/jingle/notifier/communicator/single_login_attempt.cc
+++ b/jingle/notifier/communicator/single_login_attempt.cc
@@ -41,6 +41,7 @@ SingleLoginAttempt::~SingleLoginAttempt() {}
 
 void SingleLoginAttempt::OnConnect(
     base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
+  DVLOG(1) << "Connected to " << current_settings_->ToString();
   delegate_->OnConnect(base_task);
 }
 
diff --git a/jingle/notifier/listener/push_client.cc b/jingle/notifier/listener/push_client.cc
index d704264..f96e8f8 100644
--- a/jingle/notifier/listener/push_client.cc
+++ b/jingle/notifier/listener/push_client.cc
@@ -31,4 +31,11 @@ scoped_ptr<PushClient> PushClient::CreateDefault(
       base::Bind(&CreateXmppPushClient, notifier_options)));
 }
 
+scoped_ptr<PushClient> PushClient::CreateDefaultOnIOThread(
+    const NotifierOptions& notifier_options) {
+  CHECK(notifier_options.request_context_getter->GetIOMessageLoopProxy()->
+        BelongsToCurrentThread());
+  return CreateXmppPushClient(notifier_options);
+}
+
 }  // namespace notifier
diff --git a/jingle/notifier/listener/push_client.h b/jingle/notifier/listener/push_client.h
index ea975a8..f80f422 100644
--- a/jingle/notifier/listener/push_client.h
+++ b/jingle/notifier/listener/push_client.h
@@ -27,6 +27,12 @@ class PushClient {
   static scoped_ptr<PushClient> CreateDefault(
       const NotifierOptions& notifier_options);
 
+  // Creates a default blocking PushClient implementation from the
+  // given options.  Must be called from the IO thread (according to
+  // |notifier_options|).
+  static scoped_ptr<PushClient> CreateDefaultOnIOThread(
+      const NotifierOptions& notifier_options);
+
   // Manage the list of observers for incoming notifications.
   virtual void AddObserver(PushClientObserver* observer) = 0;
   virtual void RemoveObserver(PushClientObserver* observer) = 0;
diff --git a/jingle/notifier/listener/push_client_unittest.cc b/jingle/notifier/listener/push_client_unittest.cc
index fd45466..42daf2a 100644
--- a/jingle/notifier/listener/push_client_unittest.cc
+++ b/jingle/notifier/listener/push_client_unittest.cc
@@ -33,13 +33,13 @@ class PushClientTest : public testing::Test {
 };
 
 // Make sure calling CreateDefault on the IO thread doesn't blow up.
-TEST_F(PushClientTest, OnIOThread) {
+TEST_F(PushClientTest, CreateDefaultOnIOThread) {
   const scoped_ptr<PushClient> push_client(
       PushClient::CreateDefault(notifier_options_));
 }
 
 // Make sure calling CreateDefault on a non-IO thread doesn't blow up.
-TEST_F(PushClientTest, OffIOThread) {
+TEST_F(PushClientTest, CreateDefaultOffIOThread) {
   base::Thread thread("Non-IO thread");
   EXPECT_TRUE(thread.Start());
   thread.message_loop()->PostTask(
@@ -49,6 +49,13 @@ TEST_F(PushClientTest, OffIOThread) {
   thread.Stop();
 }
 
+// Make sure calling CreateDefaultOnIOThread on the IO thread doesn't
+// blow up.
+TEST_F(PushClientTest, CreateDefaultOnIOThreadOnIOThread) {
+  const scoped_ptr<PushClient> push_client(
+      PushClient::CreateDefaultOnIOThread(notifier_options_));
+}
+
 }  // namespace
 
 }  // namespace notifier
diff --git a/sync/notifier/cache_invalidation_packet_handler.cc b/sync/notifier/cache_invalidation_packet_handler.cc
deleted file mode 100644
index 5d68753..0000000
--- a/sync/notifier/cache_invalidation_packet_handler.cc
+++ /dev/null
@@ -1,126 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "sync/notifier/cache_invalidation_packet_handler.h"
-
-#include <string>
-
-#include "base/base64.h"
-#include "base/callback.h"
-#include "base/compiler_specific.h"
-#include "base/logging.h"
-#include "base/rand_util.h"
-#include "base/string_number_conversions.h"
-#include "google/cacheinvalidation/impl/constants.h"
-#include "google/cacheinvalidation/include/invalidation-client.h"
-#include "google/cacheinvalidation/include/system-resources.h"
-#include "google/cacheinvalidation/v2/client_gateway.pb.h"
-#include "jingle/notifier/listener/notification_constants.h"
-#include "jingle/notifier/listener/push_notifications_send_update_task.h"
-#include "jingle/notifier/listener/xml_element_util.h"
-#include "talk/xmpp/constants.h"
-#include "talk/xmpp/jid.h"
-#include "talk/xmpp/xmppclient.h"
-#include "talk/xmpp/xmpptask.h"
-
-namespace sync_notifier {
-
-namespace {
-
-const char kBotJid[] = "tango@bot.talk.google.com";
-const char kChannelName[] = "tango_raw";
-
-}  // namespace
-
-CacheInvalidationPacketHandler::CacheInvalidationPacketHandler(
-    base::WeakPtr<buzz::XmppTaskParentInterface> base_task)
-    : weak_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
-      base_task_(base_task),
-      seq_(0),
-      scheduling_hash_(0) {
-  CHECK(base_task_.get());
-  // Owned by base_task.  Takes ownership of the callback.
-  notifier::PushNotificationsListenTask* listen_task =
-      new notifier::PushNotificationsListenTask(base_task_, this);
-  listen_task->Start();
-}
-
-CacheInvalidationPacketHandler::~CacheInvalidationPacketHandler() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-}
-
-void CacheInvalidationPacketHandler::SendMessage(
-    const std::string& message) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  if (!base_task_.get()) {
-    return;
-  }
-  ipc::invalidation::ClientGatewayMessage envelope;
-  envelope.set_is_client_to_server(true);
-  if (!service_context_.empty()) {
-    envelope.set_service_context(service_context_);
-    envelope.set_rpc_scheduling_hash(scheduling_hash_);
-  }
-  envelope.set_network_message(message);
-
-  notifier::Recipient recipient;
-  recipient.to = kBotJid;
-  notifier::Notification notification;
-  notification.channel = kChannelName;
-  notification.recipients.push_back(recipient);
-  envelope.SerializeToString(&notification.data);
-
-  // Owned by base_task_.
-  notifier::PushNotificationsSendUpdateTask* send_message_task =
-      new notifier::PushNotificationsSendUpdateTask(base_task_, notification);
-  send_message_task->Start();
-}
-
-void CacheInvalidationPacketHandler::SetMessageReceiver(
-    invalidation::MessageCallback* incoming_receiver) {
-  incoming_receiver_.reset(incoming_receiver);
-}
-
-void CacheInvalidationPacketHandler::SendSubscriptionRequest() {
-  notifier::Subscription subscription;
-  subscription.channel = kChannelName;
-  subscription.from = "";
-  notifier::SubscriptionList subscription_list;
-  subscription_list.push_back(subscription);
-  // Owned by base_task_.
-  notifier::PushNotificationsSubscribeTask* push_subscription_task =
-      new notifier::PushNotificationsSubscribeTask(
-          base_task_, subscription_list, this);
-  push_subscription_task->Start();
-}
-
-void CacheInvalidationPacketHandler::OnSubscribed() {
-  // TODO(ghc): Consider whether we should do more here.
-}
-
-void CacheInvalidationPacketHandler::OnSubscriptionError() {
-  // TODO(ghc): Consider whether we should do more here.
-}
-
-void CacheInvalidationPacketHandler::OnNotificationReceived(
-    const notifier::Notification& notification) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  const std::string& decoded_message = notification.data;
-  ipc::invalidation::ClientGatewayMessage envelope;
-  envelope.ParseFromString(decoded_message);
-  if (!envelope.IsInitialized()) {
-    LOG(ERROR) << "Could not parse ClientGatewayMessage: "
-               << decoded_message;
-    return;
-  }
-  if (envelope.has_service_context()) {
-    service_context_ = envelope.service_context();
-  }
-  if (envelope.has_rpc_scheduling_hash()) {
-    scheduling_hash_ = envelope.rpc_scheduling_hash();
-  }
-  incoming_receiver_->Run(envelope.network_message());
-}
-
-}  // namespace sync_notifier
diff --git a/sync/notifier/cache_invalidation_packet_handler.h b/sync/notifier/cache_invalidation_packet_handler.h
deleted file mode 100644
index 655933d..0000000
--- a/sync/notifier/cache_invalidation_packet_handler.h
+++ /dev/null
@@ -1,86 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-//
-// Class that handles the details of sending and receiving client
-// invalidation packets.
-
-#ifndef SYNC_NOTIFIER_CACHE_INVALIDATION_PACKET_HANDLER_H_
-#define SYNC_NOTIFIER_CACHE_INVALIDATION_PACKET_HANDLER_H_
-#pragma once
-
-#include <string>
-
-#include "base/basictypes.h"
-#include "base/gtest_prod_util.h"
-#include "base/memory/scoped_ptr.h"
-#include "base/memory/weak_ptr.h"
-#include "base/threading/non_thread_safe.h"
-#include "google/cacheinvalidation/include/system-resources.h"
-#include "jingle/notifier/listener/push_notifications_listen_task.h"
-#include "jingle/notifier/listener/push_notifications_subscribe_task.h"
-
-namespace buzz {
-class XmppTaskParentInterface;
-}  // namespace buzz
-
-namespace sync_notifier {
-
-class CacheInvalidationPacketHandler
-    : public notifier::PushNotificationsSubscribeTaskDelegate,
-      public notifier::PushNotificationsListenTaskDelegate {
- public:
-  // Starts routing packets from |invalidation_client| using
-  // |base_task|.  |base_task.get()| must still be non-NULL.
-  // |invalidation_client| must not already be routing packets through
-  // something.  Does not take ownership of |invalidation_client|.
-  CacheInvalidationPacketHandler(
-      base::WeakPtr<buzz::XmppTaskParentInterface> base_task);
-
-  // Makes the invalidation client passed into the constructor not
-  // route packets through the XMPP client passed into the constructor
-  // anymore.
-  virtual ~CacheInvalidationPacketHandler();
-
-  // If |base_task| is non-NULL, sends the outgoing message.
-  virtual void SendMessage(const std::string& outgoing_message);
-
-  virtual void SetMessageReceiver(
-      invalidation::MessageCallback* incoming_receiver);
-
-  // Sends a message requesting a subscription to the notification channel.
-  virtual void SendSubscriptionRequest();
-
-  // PushNotificationsSubscribeTaskDelegate implementation.
-  virtual void OnSubscribed() OVERRIDE;
-  virtual void OnSubscriptionError() OVERRIDE;
-
-  // PushNotificationsListenTaskDelegate implementation.
-  virtual void OnNotificationReceived(
-      const notifier::Notification& notification) OVERRIDE;
-
- private:
-  FRIEND_TEST_ALL_PREFIXES(CacheInvalidationPacketHandlerTest, Basic);
-
-  base::NonThreadSafe non_thread_safe_;
-  base::WeakPtrFactory<CacheInvalidationPacketHandler> weak_factory_;
-
-  base::WeakPtr<buzz::XmppTaskParentInterface> base_task_;
-
-  scoped_ptr<invalidation::MessageCallback> incoming_receiver_;
-
-  // Parameters for sent messages.
-
-  // Monotonically increasing sequence number.
-  int seq_;
-  // Service context.
-  std::string service_context_;
-  // Scheduling hash.
-  int64 scheduling_hash_;
-
-  DISALLOW_COPY_AND_ASSIGN(CacheInvalidationPacketHandler);
-};
-
-}  // namespace sync_notifier
-
-#endif  // SYNC_NOTIFIER_CACHE_INVALIDATION_PACKET_HANDLER_H_
diff --git a/sync/notifier/cache_invalidation_packet_handler_unittest.cc b/sync/notifier/cache_invalidation_packet_handler_unittest.cc
deleted file mode 100644
index 184c030..0000000
--- a/sync/notifier/cache_invalidation_packet_handler_unittest.cc
+++ /dev/null
@@ -1,77 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "sync/notifier/cache_invalidation_packet_handler.h"
-
-#include "base/base64.h"
-#include "base/memory/weak_ptr.h"
-#include "base/message_loop.h"
-#include "google/cacheinvalidation/deps/callback.h"
-#include "google/cacheinvalidation/include/system-resources.h"
-#include "google/cacheinvalidation/v2/client_gateway.pb.h"
-#include "jingle/notifier/base/fake_base_task.h"
-#include "jingle/notifier/listener/notification_defines.h"
-#include "talk/base/task.h"
-#include "talk/xmpp/asyncsocket.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-namespace sync_notifier {
-
-class MockMessageCallback {
- public:
-  void StoreMessage(const std::string& message) {
-    last_message = message;
-  }
-
-  std::string last_message;
-};
-
-class CacheInvalidationPacketHandlerTest : public testing::Test {
- public:
-  virtual ~CacheInvalidationPacketHandlerTest() {}
-
-  notifier::Notification MakeNotification(const std::string& data) {
-    notifier::Notification notification;
-    notification.channel = "tango_raw";
-    notification.data = data;
-    return notification;
-  }
-};
-
-TEST_F(CacheInvalidationPacketHandlerTest, Basic) {
-  MessageLoop message_loop;
-
-  notifier::FakeBaseTask fake_base_task;
-
-  std::string last_message;
-  MockMessageCallback callback;
-  invalidation::MessageCallback* mock_message_callback =
-      invalidation::NewPermanentCallback(
-          &callback, &MockMessageCallback::StoreMessage);
-
-  const char kInboundMessage[] = "non-bogus";
-  ipc::invalidation::ClientGatewayMessage envelope;
-  envelope.set_network_message(kInboundMessage);
-  std::string serialized;
-  envelope.SerializeToString(&serialized);
-  {
-    CacheInvalidationPacketHandler handler(fake_base_task.AsWeakPtr());
-    handler.SetMessageReceiver(mock_message_callback);
-
-    // Take care of any tasks posted by the constructor.
-    message_loop.RunAllPending();
-
-    {
-      handler.OnNotificationReceived(MakeNotification("bogus"));
-      handler.OnNotificationReceived(MakeNotification(serialized));
-    }
-
-    // Take care of any tasks posted by HandleOutboundPacket().
-    message_loop.RunAllPending();
-
-    EXPECT_EQ(callback.last_message, kInboundMessage);
-  }
-}
-
-}  // namespace sync_notifier
diff --git a/sync/notifier/chrome_invalidation_client.cc b/sync/notifier/chrome_invalidation_client.cc
index efa45e1..00e7b98 100644
--- a/sync/notifier/chrome_invalidation_client.cc
+++ b/sync/notifier/chrome_invalidation_client.cc
@@ -14,7 +14,7 @@
 #include "google/cacheinvalidation/include/invalidation-client.h"
 #include "google/cacheinvalidation/include/types.h"
 #include "google/cacheinvalidation/v2/types.pb.h"
-#include "sync/notifier/cache_invalidation_packet_handler.h"
+#include "jingle/notifier/listener/push_client.h"
 #include "sync/notifier/invalidation_util.h"
 #include "sync/notifier/registration_manager.h"
 #include "sync/syncable/model_type.h"
@@ -23,14 +23,16 @@ namespace {
 
 const char kApplicationName[] = "chrome-sync";
 
-}  // anonymous namespace
+}  // namespace
 
 namespace sync_notifier {
 
 ChromeInvalidationClient::Listener::~Listener() {}
 
-ChromeInvalidationClient::ChromeInvalidationClient()
-    : chrome_system_resources_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
+ChromeInvalidationClient::ChromeInvalidationClient(
+    scoped_ptr<notifier::PushClient> push_client)
+    : chrome_system_resources_(push_client.Pass(),
+                               ALLOW_THIS_IN_INITIALIZER_LIST(this)),
       listener_(NULL),
       state_writer_(NULL),
       ticl_ready_(false) {
@@ -51,8 +53,7 @@ void ChromeInvalidationClient::Start(
     const browser_sync::WeakHandle<InvalidationStateTracker>&
         invalidation_state_tracker,
     Listener* listener,
-    StateWriter* state_writer,
-    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
+    StateWriter* state_writer) {
   DCHECK(non_thread_safe_.CalledOnValidThread());
   Stop();
 
@@ -91,32 +92,25 @@ void ChromeInvalidationClient::Start(
       invalidation::CreateInvalidationClient(
           &chrome_system_resources_, client_type, client_id,
           kApplicationName, this));
-  ChangeBaseTask(base_task);
   invalidation_client_->Start();
 
   registration_manager_.reset(
       new RegistrationManager(invalidation_client_.get()));
 }
 
-void ChromeInvalidationClient::ChangeBaseTask(
-    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
-  DCHECK(invalidation_client_.get());
-  DCHECK(base_task.get());
-  cache_invalidation_packet_handler_.reset(
-      new CacheInvalidationPacketHandler(base_task));
-  chrome_system_resources_.network()->UpdatePacketHandler(
-      cache_invalidation_packet_handler_.get());
+void ChromeInvalidationClient::UpdateCredentials(
+    const std::string& email, const std::string& token) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  chrome_system_resources_.network()->UpdateCredentials(email, token);
 }
 
 void ChromeInvalidationClient::Stop() {
   DCHECK(non_thread_safe_.CalledOnValidThread());
   if (!invalidation_client_.get()) {
-    DCHECK(!cache_invalidation_packet_handler_.get());
     return;
   }
 
   registration_manager_.reset();
-  cache_invalidation_packet_handler_.reset();
   chrome_system_resources_.Stop();
   invalidation_client_->Stop();
 
diff --git a/sync/notifier/chrome_invalidation_client.h b/sync/notifier/chrome_invalidation_client.h
index 45f5037..0d6ac8a 100644
--- a/sync/notifier/chrome_invalidation_client.h
+++ b/sync/notifier/chrome_invalidation_client.h
@@ -28,11 +28,14 @@ namespace buzz {
 class XmppTaskParentInterface;
 }  // namespace buzz
 
+namespace notifier {
+class PushClient;
+}  // namespace notifier
+
 namespace sync_notifier {
 
 using invalidation::InvalidationListener;
 
-class CacheInvalidationPacketHandler;
 class RegistrationManager;
 
 // ChromeInvalidationClient is not thread-safe and lives on the sync
@@ -51,14 +54,14 @@ class ChromeInvalidationClient
     virtual void OnSessionStatusChanged(bool has_session) = 0;
   };
 
-  ChromeInvalidationClient();
+  explicit ChromeInvalidationClient(
+      scoped_ptr<notifier::PushClient> push_client);
 
   // Calls Stop().
   virtual ~ChromeInvalidationClient();
 
   // Does not take ownership of |listener| or |state_writer|.
   // |invalidation_state_tracker| must be initialized.
-  // |base_task| must still be non-NULL.
   void Start(
       const std::string& client_id, const std::string& client_info,
       const std::string& state,
@@ -66,15 +69,9 @@ class ChromeInvalidationClient
       const browser_sync::WeakHandle<InvalidationStateTracker>&
           invalidation_state_tracker,
       Listener* listener,
-      StateWriter* state_writer,
-      base::WeakPtr<buzz::XmppTaskParentInterface> base_task);
-
-  void Stop();
+      StateWriter* state_writer);
 
-  // Changes the task used to |base_task|, which must still be
-  // non-NULL.  Must only be called between calls to Start() and
-  // Stop().
-  void ChangeBaseTask(base::WeakPtr<buzz::XmppTaskParentInterface> base_task);
+  void UpdateCredentials(const std::string& email, const std::string& token);
 
   // Register the sync types that we're interested in getting
   // notifications for.  May be called at any time.
@@ -116,6 +113,8 @@ class ChromeInvalidationClient
  private:
   friend class ChromeInvalidationClientTest;
 
+  void Stop();
+
   void EmitInvalidation(
       syncable::ModelTypeSet types, const std::string& payload);
 
@@ -127,8 +126,6 @@ class ChromeInvalidationClient
   Listener* listener_;
   StateWriter* state_writer_;
   scoped_ptr<invalidation::InvalidationClient> invalidation_client_;
-  scoped_ptr<CacheInvalidationPacketHandler>
-      cache_invalidation_packet_handler_;
   scoped_ptr<RegistrationManager> registration_manager_;
   // Stored to pass to |registration_manager_| on start.
   syncable::ModelTypeSet registered_types_;
diff --git a/sync/notifier/chrome_invalidation_client_unittest.cc b/sync/notifier/chrome_invalidation_client_unittest.cc
index 2892491..09d100f 100644
--- a/sync/notifier/chrome_invalidation_client_unittest.cc
+++ b/sync/notifier/chrome_invalidation_client_unittest.cc
@@ -8,7 +8,7 @@
 #include "google/cacheinvalidation/include/invalidation-client.h"
 #include "google/cacheinvalidation/include/types.h"
 #include "google/cacheinvalidation/v2/types.pb.h"
-#include "jingle/notifier/base/fake_base_task.h"
+#include "jingle/notifier/listener/fake_push_client.h"
 #include "sync/notifier/chrome_invalidation_client.h"
 #include "sync/notifier/state_writer.h"
 #include "sync/syncable/model_type.h"
@@ -65,13 +65,17 @@ class MockStateWriter : public StateWriter {
 
 class ChromeInvalidationClientTest : public testing::Test {
  protected:
+  ChromeInvalidationClientTest()
+      : client_(
+          scoped_ptr<notifier::PushClient>(
+              new notifier::FakePushClient())) {}
+
   virtual void SetUp() {
     client_.Start(kClientId, kClientInfo, kState,
                   InvalidationVersionMap(),
                   browser_sync::MakeWeakHandle(
                       mock_invalidation_state_tracker_.AsWeakPtr()),
-                  &mock_listener_, &mock_state_writer_,
-                  fake_base_task_.AsWeakPtr());
+                  &mock_listener_, &mock_state_writer_);
   }
 
   virtual void TearDown() {
@@ -127,7 +131,6 @@ class ChromeInvalidationClientTest : public testing::Test {
       mock_invalidation_state_tracker_;
   StrictMock<MockStateWriter> mock_state_writer_;
   StrictMock<MockInvalidationClient> mock_invalidation_client_;
-  notifier::FakeBaseTask fake_base_task_;
   ChromeInvalidationClient client_;
 };
 
diff --git a/sync/notifier/chrome_system_resources.cc b/sync/notifier/chrome_system_resources.cc
index 498f5fd..f0c9348 100644
--- a/sync/notifier/chrome_system_resources.cc
+++ b/sync/notifier/chrome_system_resources.cc
@@ -15,7 +15,7 @@
 #include "base/string_util.h"
 #include "base/stringprintf.h"
 #include "google/cacheinvalidation/include/types.h"
-#include "sync/notifier/cache_invalidation_packet_handler.h"
+#include "jingle/notifier/listener/push_client.h"
 #include "sync/notifier/invalidation_util.h"
 
 namespace sync_notifier {
@@ -192,60 +192,15 @@ void ChromeStorage::RunAndDeleteReadKeyCallback(
   delete callback;
 }
 
-ChromeNetwork::ChromeNetwork()
-    : packet_handler_(NULL),
-      ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)) {}
-
-ChromeNetwork::~ChromeNetwork() {
-  STLDeleteElements(&network_status_receivers_);
-}
-
-void ChromeNetwork::SendMessage(const std::string& outgoing_message) {
-  if (packet_handler_) {
-    packet_handler_->SendMessage(outgoing_message);
-  }
-}
-
-void ChromeNetwork::SetMessageReceiver(
-    invalidation::MessageCallback* incoming_receiver) {
-  incoming_receiver_.reset(incoming_receiver);
-}
-
-void ChromeNetwork::AddNetworkStatusReceiver(
-    invalidation::NetworkStatusCallback* network_status_receiver) {
-  network_status_receivers_.push_back(network_status_receiver);
-}
-
-void ChromeNetwork::SetSystemResources(
-    invalidation::SystemResources* resources) {
-  // Do nothing.
-}
-
-void ChromeNetwork::UpdatePacketHandler(
-    CacheInvalidationPacketHandler* packet_handler) {
-  packet_handler_ = packet_handler;
-  if (packet_handler_ != NULL) {
-    packet_handler_->SetMessageReceiver(
-        new invalidation::MessageCallback(
-            base::Bind(&ChromeNetwork::HandleInboundMessage,
-                       weak_factory_.GetWeakPtr())));
-  }
-  packet_handler_->SendSubscriptionRequest();
-}
-
-void ChromeNetwork::HandleInboundMessage(const std::string& incoming_message) {
-  if (incoming_receiver_.get()) {
-    incoming_receiver_->Run(incoming_message);
-  }
-}
-
-ChromeSystemResources::ChromeSystemResources(StateWriter* state_writer)
+ChromeSystemResources::ChromeSystemResources(
+    scoped_ptr<notifier::PushClient> push_client,
+    StateWriter* state_writer)
     : is_started_(false),
       logger_(new ChromeLogger()),
       internal_scheduler_(new ChromeScheduler()),
       listener_scheduler_(new ChromeScheduler()),
       storage_(new ChromeStorage(state_writer, internal_scheduler_.get())),
-      network_(new ChromeNetwork()) {
+      push_client_channel_(push_client.Pass()) {
 }
 
 ChromeSystemResources::~ChromeSystemResources() {
@@ -283,8 +238,8 @@ ChromeStorage* ChromeSystemResources::storage() {
   return storage_.get();
 }
 
-ChromeNetwork* ChromeSystemResources::network() {
-  return network_.get();
+PushClientChannel* ChromeSystemResources::network() {
+  return &push_client_channel_;
 }
 
 ChromeScheduler* ChromeSystemResources::internal_scheduler() {
diff --git a/sync/notifier/chrome_system_resources.h b/sync/notifier/chrome_system_resources.h
index 032e050..852b56c 100644
--- a/sync/notifier/chrome_system_resources.h
+++ b/sync/notifier/chrome_system_resources.h
@@ -20,11 +20,14 @@
 #include "base/message_loop.h"
 #include "base/threading/non_thread_safe.h"
 #include "google/cacheinvalidation/include/system-resources.h"
+#include "sync/notifier/push_client_channel.h"
 #include "sync/notifier/state_writer.h"
 
-namespace sync_notifier {
+namespace notifier {
+class PushClient;
+}  // namespace notifier
 
-class CacheInvalidationPacketHandler;
+namespace sync_notifier {
 
 class ChromeLogger : public invalidation::Logger {
  public:
@@ -114,38 +117,10 @@ class ChromeStorage : public invalidation::Storage {
   std::string cached_state_;
 };
 
-class ChromeNetwork : public invalidation::NetworkChannel {
- public:
-  ChromeNetwork();
-
-  virtual ~ChromeNetwork();
-
-  void UpdatePacketHandler(CacheInvalidationPacketHandler* packet_handler);
-
-  // invalidation::NetworkChannel implementation.
-  virtual void SendMessage(const std::string& outgoing_message) OVERRIDE;
-
-  virtual void SetMessageReceiver(
-      invalidation::MessageCallback* incoming_receiver) OVERRIDE;
-
-  virtual void AddNetworkStatusReceiver(
-      invalidation::NetworkStatusCallback* network_status_receiver) OVERRIDE;
-
-  virtual void SetSystemResources(
-      invalidation::SystemResources* resources) OVERRIDE;
-
- private:
-  void HandleInboundMessage(const std::string& incoming_message);
-
-  CacheInvalidationPacketHandler* packet_handler_;
-  scoped_ptr<invalidation::MessageCallback> incoming_receiver_;
-  std::vector<invalidation::NetworkStatusCallback*> network_status_receivers_;
-  base::WeakPtrFactory<ChromeNetwork> weak_factory_;
-};
-
 class ChromeSystemResources : public invalidation::SystemResources {
  public:
-  explicit ChromeSystemResources(StateWriter* state_writer);
+  ChromeSystemResources(scoped_ptr<notifier::PushClient> push_client,
+                        StateWriter* state_writer);
 
   virtual ~ChromeSystemResources();
 
@@ -157,7 +132,7 @@ class ChromeSystemResources : public invalidation::SystemResources {
   virtual std::string platform() const OVERRIDE;
   virtual ChromeLogger* logger() OVERRIDE;
   virtual ChromeStorage* storage() OVERRIDE;
-  virtual ChromeNetwork* network() OVERRIDE;
+  virtual PushClientChannel* network() OVERRIDE;
   virtual ChromeScheduler* internal_scheduler() OVERRIDE;
   virtual ChromeScheduler* listener_scheduler() OVERRIDE;
 
@@ -168,7 +143,7 @@ class ChromeSystemResources : public invalidation::SystemResources {
   scoped_ptr<ChromeScheduler> internal_scheduler_;
   scoped_ptr<ChromeScheduler> listener_scheduler_;
   scoped_ptr<ChromeStorage> storage_;
-  scoped_ptr<ChromeNetwork> network_;
+  PushClientChannel push_client_channel_;
 };
 
 }  // namespace sync_notifier
diff --git a/sync/notifier/chrome_system_resources_unittest.cc b/sync/notifier/chrome_system_resources_unittest.cc
index a61497f..bb969af 100644
--- a/sync/notifier/chrome_system_resources_unittest.cc
+++ b/sync/notifier/chrome_system_resources_unittest.cc
@@ -12,6 +12,7 @@
 #include "base/message_loop.h"
 
 #include "google/cacheinvalidation/include/types.h"
+#include "jingle/notifier/listener/fake_push_client.h"
 #include "sync/notifier/state_writer.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -48,7 +49,9 @@ class MockStorageCallback {
 class ChromeSystemResourcesTest : public testing::Test {
  protected:
   ChromeSystemResourcesTest()
-      : chrome_system_resources_(&mock_state_writer_) {}
+      : chrome_system_resources_(
+          scoped_ptr<notifier::PushClient>(new notifier::FakePushClient()),
+          &mock_state_writer_) {}
 
   virtual ~ChromeSystemResourcesTest() {}
 
diff --git a/sync/notifier/invalidation_notifier.cc b/sync/notifier/invalidation_notifier.cc
index 60f81cc..c86a502 100644
--- a/sync/notifier/invalidation_notifier.cc
+++ b/sync/notifier/invalidation_notifier.cc
@@ -6,8 +6,7 @@
 
 #include "base/logging.h"
 #include "base/message_loop_proxy.h"
-#include "jingle/notifier/base/const_communicator.h"
-#include "jingle/notifier/base/notifier_options_util.h"
+#include "jingle/notifier/listener/push_client.h"
 #include "net/url_request/url_request_context.h"
 #include "sync/notifier/sync_notifier_observer.h"
 #include "sync/syncable/model_type_payload_map.h"
@@ -17,22 +16,16 @@
 namespace sync_notifier {
 
 InvalidationNotifier::InvalidationNotifier(
-    const notifier::NotifierOptions& notifier_options,
+    scoped_ptr<notifier::PushClient> push_client,
     const InvalidationVersionMap& initial_max_invalidation_versions,
     const browser_sync::WeakHandle<InvalidationStateTracker>&
         invalidation_state_tracker,
     const std::string& client_info)
     : state_(STOPPED),
-      notifier_options_(notifier_options),
       initial_max_invalidation_versions_(initial_max_invalidation_versions),
       invalidation_state_tracker_(invalidation_state_tracker),
-      client_info_(client_info) {
-  DCHECK_EQ(notifier::NOTIFICATION_SERVER,
-            notifier_options.notification_method);
-  DCHECK(notifier_options_.request_context_getter);
-  // TODO(akalin): Replace NonThreadSafe checks with IO thread checks.
-  DCHECK(notifier_options_.request_context_getter->GetIOMessageLoopProxy()->
-      BelongsToCurrentThread());
+      client_info_(client_info),
+      invalidation_client_(push_client.Pass()) {
 }
 
 InvalidationNotifier::~InvalidationNotifier() {
@@ -64,25 +57,16 @@ void InvalidationNotifier::SetState(const std::string& state) {
 
 void InvalidationNotifier::UpdateCredentials(
     const std::string& email, const std::string& token) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  CHECK(!invalidation_client_id_.empty());
-  DVLOG(1) << "Updating credentials for " << email;
-  buzz::XmppClientSettings xmpp_client_settings =
-      notifier::MakeXmppClientSettings(notifier_options_, email, token);
-  if (state_ >= CONNECTING) {
-    login_->UpdateXmppSettings(xmpp_client_settings);
-  } else {
-    DVLOG(1) << "First time updating credentials: connecting";
-    login_.reset(
-        new notifier::Login(this,
-                            xmpp_client_settings,
-                            notifier_options_.request_context_getter,
-                            notifier::GetServerList(notifier_options_),
-                            notifier_options_.try_ssltcp_first,
-                            notifier_options_.auth_mechanism));
-    login_->StartConnection();
-    state_ = CONNECTING;
+  if (state_ == STOPPED) {
+    invalidation_client_.Start(
+        invalidation_client_id_, client_info_, invalidation_state_,
+        initial_max_invalidation_versions_,
+        invalidation_state_tracker_,
+        this, this);
+    invalidation_state_.clear();
+    state_ = STARTED;
   }
+  invalidation_client_.UpdateCredentials(email, token);
 }
 
 void InvalidationNotifier::UpdateEnabledTypes(
@@ -98,31 +82,6 @@ void InvalidationNotifier::SendNotification(
   // Do nothing.
 }
 
-void InvalidationNotifier::OnConnect(
-    base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  DVLOG(1) << "OnConnect";
-  if (state_ >= STARTED) {
-    invalidation_client_.ChangeBaseTask(base_task);
-  } else {
-    DVLOG(1) << "First time connecting: starting invalidation client with id "
-             << invalidation_client_id_ << " and client info "
-             << client_info_;
-    invalidation_client_.Start(
-        invalidation_client_id_, client_info_, invalidation_state_,
-        initial_max_invalidation_versions_,
-        invalidation_state_tracker_,
-        this, this, base_task);
-    invalidation_state_.clear();
-    state_ = STARTED;
-  }
-}
-
-void InvalidationNotifier::OnDisconnect() {
-  DCHECK(non_thread_safe_.CalledOnValidThread());
-  DVLOG(1) << "OnDisconnect";
-}
-
 void InvalidationNotifier::OnInvalidate(
     const syncable::ModelTypePayloadMap& type_payloads) {
   DCHECK(non_thread_safe_.CalledOnValidThread());
diff --git a/sync/notifier/invalidation_notifier.h b/sync/notifier/invalidation_notifier.h
index a287a6f..1771ab7 100644
--- a/sync/notifier/invalidation_notifier.h
+++ b/sync/notifier/invalidation_notifier.h
@@ -20,8 +20,6 @@
 #include "base/memory/scoped_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 "sync/notifier/chrome_invalidation_client.h"
 #include "sync/notifier/invalidation_state_tracker.h"
 #include "sync/notifier/state_writer.h"
@@ -29,18 +27,21 @@
 #include "sync/syncable/model_type.h"
 #include "sync/util/weak_handle.h"
 
+namespace notifier {
+class PushClient;
+}  // namespace notifier
+
 namespace sync_notifier {
 
 // This class must live on the IO thread.
 class InvalidationNotifier
     : public SyncNotifier,
-      public notifier::LoginDelegate,
       public ChromeInvalidationClient::Listener,
       public StateWriter {
  public:
   // |invalidation_state_tracker| must be initialized.
   InvalidationNotifier(
-      const notifier::NotifierOptions& notifier_options,
+      scoped_ptr<notifier::PushClient> push_client,
       const InvalidationVersionMap& initial_max_invalidation_versions,
       const browser_sync::WeakHandle<InvalidationStateTracker>&
           invalidation_state_tracker,
@@ -60,11 +61,6 @@ class InvalidationNotifier
   virtual void SendNotification(
       syncable::ModelTypeSet changed_types) OVERRIDE;
 
-  // notifier::LoginDelegate implementation.
-  virtual void OnConnect(
-      base::WeakPtr<buzz::XmppTaskParentInterface> base_task) OVERRIDE;
-  virtual void OnDisconnect() OVERRIDE;
-
   // ChromeInvalidationClient::Listener implementation.
   virtual void OnInvalidate(
       const syncable::ModelTypePayloadMap& type_payloads) OVERRIDE;
@@ -87,9 +83,6 @@ class InvalidationNotifier
   };
   State state_;
 
-  // Used to build parameters for |login_|.
-  const notifier::NotifierOptions notifier_options_;
-
   // Passed to |invalidation_client_|.
   const InvalidationVersionMap initial_max_invalidation_versions_;
 
@@ -109,10 +102,6 @@ class InvalidationNotifier
   // The state to pass to |chrome_invalidation_client_|.
   std::string invalidation_state_;
 
-  // The XMPP connection manager.
-  // TODO(akalin): Use PushClient instead.
-  scoped_ptr<notifier::Login> login_;
-
   // The invalidation client.
   ChromeInvalidationClient invalidation_client_;
 
diff --git a/sync/notifier/invalidation_notifier_unittest.cc b/sync/notifier/invalidation_notifier_unittest.cc
index 9c6d593..d3646e0 100644
--- a/sync/notifier/invalidation_notifier_unittest.cc
+++ b/sync/notifier/invalidation_notifier_unittest.cc
@@ -8,6 +8,7 @@
 #include "base/message_loop.h"
 #include "jingle/notifier/base/fake_base_task.h"
 #include "jingle/notifier/base/notifier_options.h"
+#include "jingle/notifier/listener/fake_push_client.h"
 #include "net/url_request/url_request_test_util.h"
 #include "sync/notifier/invalidation_state_tracker.h"
 #include "sync/notifier/mock_sync_notifier_observer.h"
@@ -33,7 +34,7 @@ class InvalidationNotifierTest : public testing::Test {
         new TestURLRequestContextGetter(message_loop_.message_loop_proxy());
     invalidation_notifier_.reset(
         new InvalidationNotifier(
-            notifier_options,
+            scoped_ptr<notifier::PushClient>(new notifier::FakePushClient()),
             InvalidationVersionMap(),
             browser_sync::MakeWeakHandle(
                 base::WeakPtr<InvalidationStateTracker>()),
@@ -77,17 +78,12 @@ TEST_F(InvalidationNotifierTest, Basic) {
   invalidation_notifier_->SetUniqueId("fake_id");
   invalidation_notifier_->UpdateCredentials("foo@bar.com", "fake_token");
 
-  invalidation_notifier_->OnConnect(fake_base_task_.AsWeakPtr());
   invalidation_notifier_->OnSessionStatusChanged(true);
 
   invalidation_notifier_->WriteState("new_fake_state");
 
   invalidation_notifier_->OnInvalidate(type_payloads);
 
-  // Shouldn't trigger notification state change.
-  invalidation_notifier_->OnDisconnect();
-  invalidation_notifier_->OnConnect(fake_base_task_.AsWeakPtr());
-
   invalidation_notifier_->OnSessionStatusChanged(false);
 }
 
diff --git a/sync/notifier/non_blocking_invalidation_notifier.cc b/sync/notifier/non_blocking_invalidation_notifier.cc
index 8830a23..3c4fb13 100644
--- a/sync/notifier/non_blocking_invalidation_notifier.cc
+++ b/sync/notifier/non_blocking_invalidation_notifier.cc
@@ -8,6 +8,7 @@
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop.h"
 #include "base/threading/thread.h"
+#include "jingle/notifier/listener/push_client.h"
 #include "sync/notifier/invalidation_notifier.h"
 
 namespace sync_notifier {
@@ -82,7 +83,7 @@ void NonBlockingInvalidationNotifier::Core::Initialize(
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
   invalidation_notifier_.reset(
       new InvalidationNotifier(
-          notifier_options,
+          notifier::PushClient::CreateDefaultOnIOThread(notifier_options),
           initial_max_invalidation_versions,
           invalidation_state_tracker,
           client_info));
diff --git a/sync/notifier/push_client_channel.cc b/sync/notifier/push_client_channel.cc
new file mode 100644
index 0000000..cfd3129
--- /dev/null
+++ b/sync/notifier/push_client_channel.cc
@@ -0,0 +1,151 @@
+// 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 "sync/notifier/push_client_channel.h"
+
+#include "base/stl_util.h"
+#include "google/cacheinvalidation/v2/client_gateway.pb.h"
+#include "jingle/notifier/listener/push_client.h"
+
+namespace sync_notifier {
+
+namespace {
+
+const char kBotJid[] = "tango@bot.talk.google.com";
+const char kChannelName[] = "tango_raw";
+
+}  // namespace
+
+PushClientChannel::PushClientChannel(
+    scoped_ptr<notifier::PushClient> push_client)
+    : push_client_(push_client.Pass()),
+      notifications_enabled_(false),
+      scheduling_hash_(0) {
+  push_client_->AddObserver(this);
+  notifier::Subscription subscription;
+  subscription.channel = kChannelName;
+  subscription.from = "";
+  notifier::SubscriptionList subscriptions;
+  subscriptions.push_back(subscription);
+  push_client_->UpdateSubscriptions(subscriptions);
+}
+
+PushClientChannel::~PushClientChannel() {
+  push_client_->RemoveObserver(this);
+  STLDeleteElements(&network_status_receivers_);
+}
+
+void PushClientChannel::UpdateCredentials(
+    const std::string& email, const std::string& token) {
+  push_client_->UpdateCredentials(email, token);
+}
+
+void PushClientChannel::SendMessage(const std::string& outgoing_message) {
+  push_client_->SendNotification(
+      EncodeMessage(outgoing_message, service_context_, scheduling_hash_));
+}
+
+void PushClientChannel::SetMessageReceiver(
+    invalidation::MessageCallback* incoming_receiver) {
+  incoming_receiver_.reset(incoming_receiver);
+}
+
+void PushClientChannel::AddNetworkStatusReceiver(
+    invalidation::NetworkStatusCallback* network_status_receiver) {
+  network_status_receiver->Run(notifications_enabled_);
+  network_status_receivers_.push_back(network_status_receiver);
+}
+
+void PushClientChannel::SetSystemResources(
+    invalidation::SystemResources* resources) {
+  // Do nothing.
+}
+
+void PushClientChannel::OnNotificationStateChange(
+    bool notifications_enabled) {
+  for (NetworkStatusReceiverList::const_iterator it =
+           network_status_receivers_.begin();
+       it != network_status_receivers_.end(); ++it) {
+    (*it)->Run(notifications_enabled);
+  }
+}
+
+void PushClientChannel::OnIncomingNotification(
+    const notifier::Notification& notification) {
+  if (!incoming_receiver_.get()) {
+    DLOG(ERROR) << "No receiver for incoming notification";
+    return;
+  }
+  std::string message;
+  if (!DecodeMessage(notification,
+                     &message, &service_context_, &scheduling_hash_)) {
+    DLOG(ERROR) << "Could not parse ClientGatewayMessage from: "
+                << notification.ToString();
+  }
+  incoming_receiver_->Run(message);
+}
+
+const std::string& PushClientChannel::GetServiceContextForTest() const {
+  return service_context_;
+}
+
+int64 PushClientChannel::GetSchedulingHashForTest() const {
+  return scheduling_hash_;
+}
+
+notifier::Notification PushClientChannel::EncodeMessageForTest(
+    const std::string& message, const std::string& service_context,
+    int64 scheduling_hash) {
+  return EncodeMessage(message, service_context, scheduling_hash);
+}
+
+bool PushClientChannel::DecodeMessageForTest(
+    const notifier::Notification& notification,
+    std::string* message,
+    std::string* service_context,
+    int64* scheduling_hash) {
+  return DecodeMessage(
+      notification, message, service_context, scheduling_hash);
+}
+
+notifier::Notification PushClientChannel::EncodeMessage(
+    const std::string& message, const std::string& service_context,
+    int64 scheduling_hash) {
+  ipc::invalidation::ClientGatewayMessage envelope;
+  envelope.set_is_client_to_server(true);
+  if (!service_context.empty()) {
+    envelope.set_service_context(service_context);
+    envelope.set_rpc_scheduling_hash(scheduling_hash);
+  }
+  envelope.set_network_message(message);
+
+  notifier::Recipient recipient;
+  recipient.to = kBotJid;
+  notifier::Notification notification;
+  notification.channel = kChannelName;
+  notification.recipients.push_back(recipient);
+  envelope.SerializeToString(&notification.data);
+  return notification;
+}
+
+bool PushClientChannel::DecodeMessage(
+    const notifier::Notification& notification,
+    std::string* message,
+    std::string* service_context,
+    int64* scheduling_hash) {
+  ipc::invalidation::ClientGatewayMessage envelope;
+  if (!envelope.ParseFromString(notification.data)) {
+    return false;
+  }
+  *message = envelope.network_message();
+  if (envelope.has_service_context()) {
+    *service_context = envelope.service_context();
+  }
+  if (envelope.has_rpc_scheduling_hash()) {
+    *scheduling_hash = envelope.rpc_scheduling_hash();
+  }
+  return true;
+}
+
+}  // namespace sync_notifier
diff --git a/sync/notifier/push_client_channel.h b/sync/notifier/push_client_channel.h
new file mode 100644
index 0000000..1721def
--- /dev/null
+++ b/sync/notifier/push_client_channel.h
@@ -0,0 +1,100 @@
+// 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 SYNC_NOTIFIER_PUSH_CLIENT_CHANNEL_H_
+#define SYNC_NOTIFIER_PUSH_CLIENT_CHANNEL_H_
+#pragma once
+
+#include <string>
+#include <vector>
+
+#include "base/basictypes.h"
+#include "base/compiler_specific.h"
+#include "base/memory/scoped_ptr.h"
+#include "google/cacheinvalidation/include/system-resources.h"
+#include "jingle/notifier/listener/push_client_observer.h"
+
+namespace notifier {
+class PushClient;
+}  // namespace notifier
+
+namespace sync_notifier {
+
+// A PushClientChannel is an implementation of NetworkChannel that
+// routes messages through a PushClient.
+class PushClientChannel
+    : public invalidation::NetworkChannel,
+      public notifier::PushClientObserver {
+ public:
+  explicit PushClientChannel(scoped_ptr<notifier::PushClient> push_client);
+
+  virtual ~PushClientChannel();
+
+  // 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);
+
+  // invalidation::NetworkChannel implementation.
+  virtual void SendMessage(const std::string& outgoing_message) OVERRIDE;
+  virtual void SetMessageReceiver(
+      invalidation::MessageCallback* incoming_receiver) OVERRIDE;
+  virtual void AddNetworkStatusReceiver(
+      invalidation::NetworkStatusCallback* network_status_receiver) OVERRIDE;
+  virtual void SetSystemResources(
+      invalidation::SystemResources* resources) OVERRIDE;
+
+  // notifier::PushClient::Observer implementation.
+  virtual void OnNotificationStateChange(bool notifications_enabled) OVERRIDE;
+  virtual void OnIncomingNotification(
+      const notifier::Notification& notification) OVERRIDE;
+
+  const std::string& GetServiceContextForTest() const;
+
+  int64 GetSchedulingHashForTest() const;
+
+  static notifier::Notification EncodeMessageForTest(
+      const std::string& message,
+      const std::string& service_context,
+      int64 scheduling_hash);
+
+  static bool DecodeMessageForTest(
+      const notifier::Notification& notification,
+      std::string* message,
+      std::string* service_context,
+      int64* scheduling_hash);
+
+ private:
+  typedef std::vector<invalidation::NetworkStatusCallback*>
+      NetworkStatusReceiverList;
+
+  static notifier::Notification EncodeMessage(
+      const std::string& message,
+      const std::string& service_context,
+      int64 scheduling_hash);
+
+  static bool DecodeMessage(
+      const notifier::Notification& notification,
+      std::string* message,
+      std::string* service_context,
+      int64* scheduling_hash);
+
+  scoped_ptr<notifier::PushClient> push_client_;
+  scoped_ptr<invalidation::MessageCallback> incoming_receiver_;
+  NetworkStatusReceiverList network_status_receivers_;
+
+  bool notifications_enabled_;
+
+  // Service context.
+  std::string service_context_;
+
+  // Scheduling hash.
+  int64 scheduling_hash_;
+
+  DISALLOW_COPY_AND_ASSIGN(PushClientChannel);
+};
+
+}  // namespace sync_notifier
+
+#endif  // SYNC_NOTIFIER_PUSH_CLIENT_CHANNEL_H_
diff --git a/sync/notifier/push_client_channel_unittest.cc b/sync/notifier/push_client_channel_unittest.cc
new file mode 100644
index 0000000..e179109
--- /dev/null
+++ b/sync/notifier/push_client_channel_unittest.cc
@@ -0,0 +1,247 @@
+// 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 "sync/notifier/push_client_channel.h"
+
+#include <cstddef>
+#include <string>
+
+#include "base/compiler_specific.h"
+#include "jingle/notifier/listener/fake_push_client.h"
+#include "jingle/notifier/listener/notification_defines.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace sync_notifier {
+namespace {
+
+class PushClientChannelTest : public ::testing::Test {
+ protected:
+  PushClientChannelTest()
+      : fake_push_client_(new notifier::FakePushClient()),
+        push_client_channel_(
+            scoped_ptr<notifier::PushClient>(fake_push_client_)),
+        connected_(false) {
+    push_client_channel_.SetMessageReceiver(
+        invalidation::NewPermanentCallback(
+            this, &PushClientChannelTest::OnIncomingMessage));
+    push_client_channel_.AddNetworkStatusReceiver(
+        invalidation::NewPermanentCallback(
+            this, &PushClientChannelTest::OnNetworkStatusChange));
+    push_client_channel_.SetSystemResources(NULL);
+  }
+
+  virtual ~PushClientChannelTest() {}
+
+  void OnIncomingMessage(const std::string& incoming_message) {
+    last_message_ = incoming_message;
+  }
+
+  void OnNetworkStatusChange(bool connected) {
+    connected_ = connected;
+  }
+
+  notifier::FakePushClient* fake_push_client_;
+  PushClientChannel push_client_channel_;
+  std::string last_message_;
+  bool connected_;
+};
+
+const char kMessage[] = "message";
+const char kServiceContext[] = "service context";
+const int64 kSchedulingHash = 100;
+
+// Encode a message with some context into a notification and then
+// decode it.  The decoded info should match the original info.
+TEST_F(PushClientChannelTest, EncodeDecode) {
+  const notifier::Notification& notification =
+      PushClientChannel::EncodeMessageForTest(
+          kMessage, kServiceContext, kSchedulingHash);
+  std::string message;
+  std::string service_context;
+  int64 scheduling_hash = 0LL;
+  EXPECT_TRUE(PushClientChannel::DecodeMessageForTest(
+      notification, &message, &service_context, &scheduling_hash));
+  EXPECT_EQ(kMessage, message);
+  EXPECT_EQ(kServiceContext, service_context);
+  EXPECT_EQ(kSchedulingHash, scheduling_hash);
+}
+
+// Encode a message with no context into a notification and then
+// decode it.  The decoded message should match the original message,
+// but the context and hash should be untouched.
+TEST_F(PushClientChannelTest, EncodeDecodeNoContext) {
+  const notifier::Notification& notification =
+      PushClientChannel::EncodeMessageForTest(
+          kMessage, "", kSchedulingHash);
+  std::string message;
+  std::string service_context = kServiceContext;
+  int64 scheduling_hash = kSchedulingHash + 1;
+  EXPECT_TRUE(PushClientChannel::DecodeMessageForTest(
+      notification, &message, &service_context, &scheduling_hash));
+  EXPECT_EQ(kMessage, message);
+  EXPECT_EQ(kServiceContext, service_context);
+  EXPECT_EQ(kSchedulingHash + 1, scheduling_hash);
+}
+
+// Decode an empty notification.  It should result in an empty message
+// but should leave the context and hash untouched.
+TEST_F(PushClientChannelTest, DecodeEmpty) {
+  std::string message = kMessage;
+  std::string service_context = kServiceContext;
+  int64 scheduling_hash = kSchedulingHash;
+  EXPECT_TRUE(PushClientChannel::DecodeMessageForTest(
+      notifier::Notification(),
+      &message, &service_context, &scheduling_hash));
+  EXPECT_TRUE(message.empty());
+  EXPECT_EQ(kServiceContext, service_context);
+  EXPECT_EQ(kSchedulingHash, scheduling_hash);
+}
+
+// Try to decode a garbage notification.  It should leave all its
+// arguments untouched and return false.
+TEST_F(PushClientChannelTest, DecodeGarbage) {
+  notifier::Notification notification;
+  notification.data = "garbage";
+  std::string message = kMessage;
+  std::string service_context = kServiceContext;
+  int64 scheduling_hash = kSchedulingHash;
+  EXPECT_FALSE(PushClientChannel::DecodeMessageForTest(
+      notification, &message, &service_context, &scheduling_hash));
+  EXPECT_EQ(kMessage, message);
+  EXPECT_EQ(kServiceContext, service_context);
+  EXPECT_EQ(kSchedulingHash, scheduling_hash);
+}
+
+// Make sure the channel subscribes to the correct notifications
+// channel on construction.
+TEST_F(PushClientChannelTest, Subscriptions) {
+  notifier::Subscription expected_subscription;
+  expected_subscription.channel = "tango_raw";
+  EXPECT_TRUE(notifier::SubscriptionListsEqual(
+      fake_push_client_->subscriptions(),
+      notifier::SubscriptionList(1, expected_subscription)));
+}
+
+// Call UpdateCredentials on the channel.  It should propagate it to
+// the push client.
+TEST_F(PushClientChannelTest, UpdateCredentials) {
+  const char kEmail[] = "foo@bar.com";
+  const char kToken[] = "token";
+  EXPECT_TRUE(fake_push_client_->email().empty());
+  EXPECT_TRUE(fake_push_client_->token().empty());
+  push_client_channel_.UpdateCredentials(kEmail, kToken);
+  EXPECT_EQ(kEmail, fake_push_client_->email());
+  EXPECT_EQ(kToken, fake_push_client_->token());
+}
+
+// Call SendMessage on the channel.  It should propagate it to the
+// push client.
+TEST_F(PushClientChannelTest, SendMessage) {
+  EXPECT_TRUE(fake_push_client_->sent_notifications().empty());
+  push_client_channel_.SendMessage(kMessage);
+  const notifier::Notification expected_notification =
+      PushClientChannel::EncodeMessageForTest(
+          kMessage,
+          push_client_channel_.GetServiceContextForTest(),
+          push_client_channel_.GetSchedulingHashForTest());
+  ASSERT_EQ(1u, fake_push_client_->sent_notifications().size());
+  EXPECT_TRUE(
+      fake_push_client_->sent_notifications()[0].Equals(
+          expected_notification));
+}
+
+// Simulate notification state changes on the push client.  It should
+// propagate to the channel.
+TEST_F(PushClientChannelTest, OnNotificationStateChange) {
+  EXPECT_FALSE(connected_);
+  fake_push_client_->SimulateNotificationStateChange(true);
+  EXPECT_TRUE(connected_);
+  fake_push_client_->SimulateNotificationStateChange(false);
+  EXPECT_FALSE(connected_);
+}
+
+// Simulate an incoming notification.  It should be decoded properly
+// by the channel.
+TEST_F(PushClientChannelTest, OnIncomingNotification) {
+  const notifier::Notification notification =
+      PushClientChannel::EncodeMessageForTest(
+          kMessage, kServiceContext, kSchedulingHash);
+
+  fake_push_client_->SimulateIncomingNotification(notification);
+  EXPECT_EQ(kServiceContext,
+            push_client_channel_.GetServiceContextForTest());
+  EXPECT_EQ(kSchedulingHash,
+            push_client_channel_.GetSchedulingHashForTest());
+  EXPECT_EQ(kMessage, last_message_);
+}
+
+// Simulate an incoming notification with no receiver.  It should be
+// dropped by the channel.
+TEST_F(PushClientChannelTest, OnIncomingNotificationNoReceiver) {
+  const notifier::Notification notification =
+      PushClientChannel::EncodeMessageForTest(
+          kMessage, kServiceContext, kSchedulingHash);
+
+  push_client_channel_.SetMessageReceiver(NULL);
+  fake_push_client_->SimulateIncomingNotification(notification);
+  EXPECT_TRUE(push_client_channel_.GetServiceContextForTest().empty());
+  EXPECT_EQ(static_cast<int64>(0),
+            push_client_channel_.GetSchedulingHashForTest());
+  EXPECT_TRUE(last_message_.empty());
+}
+
+// Simulate an incoming garbage notification.  It should be dropped by
+// the channel.
+TEST_F(PushClientChannelTest, OnIncomingNotificationGarbage) {
+  notifier::Notification notification;
+  notification.data = "garbage";
+
+  fake_push_client_->SimulateIncomingNotification(notification);
+  EXPECT_TRUE(push_client_channel_.GetServiceContextForTest().empty());
+  EXPECT_EQ(static_cast<int64>(0),
+            push_client_channel_.GetSchedulingHashForTest());
+  EXPECT_TRUE(last_message_.empty());
+}
+
+// Send a message, simulate an incoming message with context, and then
+// send the same message again.  The first sent message should not
+// have any context, but the second sent message should have the
+// context from the incoming emssage.
+TEST_F(PushClientChannelTest, PersistedMessageState) {
+  push_client_channel_.SendMessage(kMessage);
+  ASSERT_EQ(1u, fake_push_client_->sent_notifications().size());
+  {
+    std::string message;
+    std::string service_context;
+    int64 scheduling_hash = 0LL;
+    EXPECT_TRUE(PushClientChannel::DecodeMessageForTest(
+        fake_push_client_->sent_notifications()[0],
+        &message, &service_context, &scheduling_hash));
+    EXPECT_EQ(kMessage, message);
+    EXPECT_TRUE(service_context.empty());
+    EXPECT_EQ(0LL, scheduling_hash);
+  }
+
+  const notifier::Notification notification =
+      PushClientChannel::EncodeMessageForTest(
+          kMessage, kServiceContext, kSchedulingHash);
+  fake_push_client_->SimulateIncomingNotification(notification);
+
+  push_client_channel_.SendMessage(kMessage);
+  ASSERT_EQ(2u, fake_push_client_->sent_notifications().size());
+  {
+    std::string message;
+    std::string service_context;
+    int64 scheduling_hash = 0LL;
+    EXPECT_TRUE(PushClientChannel::DecodeMessageForTest(
+        fake_push_client_->sent_notifications()[1],
+        &message, &service_context, &scheduling_hash));
+    EXPECT_EQ(kMessage, message);
+    EXPECT_EQ(kServiceContext, service_context);
+    EXPECT_EQ(kSchedulingHash, scheduling_hash);
+  }
+}
+
+}  // namespace
+}  // namespace sync_notifier
diff --git a/sync/sync.gyp b/sync/sync.gyp
index 8bb5c2c..7a46b9d 100644
--- a/sync/sync.gyp
+++ b/sync/sync.gyp
@@ -213,8 +213,6 @@
       'conditions': [
         ['OS != "android"', {
           'sources': [
-            'notifier/cache_invalidation_packet_handler.cc',
-            'notifier/cache_invalidation_packet_handler.h',
             'notifier/chrome_invalidation_client.cc',
             'notifier/chrome_invalidation_client.h',
             'notifier/chrome_system_resources.cc',
@@ -228,6 +226,8 @@
             'notifier/non_blocking_invalidation_notifier.cc',
             'notifier/p2p_notifier.h',
             'notifier/p2p_notifier.cc',
+            'notifier/push_client_channel.cc',
+            'notifier/push_client_channel.h',
             'notifier/registration_manager.cc',
             'notifier/registration_manager.h',
             'notifier/state_writer.h',
@@ -586,12 +586,12 @@
         'conditions': [
           ['OS != "android"', {
             'sources': [
-              'notifier/cache_invalidation_packet_handler_unittest.cc',
               'notifier/chrome_invalidation_client_unittest.cc',
               'notifier/chrome_system_resources_unittest.cc',
               'notifier/invalidation_notifier_unittest.cc',
               'notifier/non_blocking_invalidation_notifier_unittest.cc',
               'notifier/p2p_notifier_unittest.cc',
+              'notifier/push_client_channel_unittest.cc',
               'notifier/registration_manager_unittest.cc',
             ],
           }],