[Sync] Split SyncNotifierImpl into P2PNotifier and InvalidationNotifier

Now that we have a clean interface, there's no need to have these
implementations mashed together.

Streamline ServerNotifierThread now that we use it directly.

BUG=76764
TEST=

Review URL: http://codereview.chromium.org/6685076

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

16 files changed, 448 insertions, 374 deletions

diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index 591bb02..185c116 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -25,7 +25,6 @@
 #include "base/task.h"
 #include "base/utf_string_conversions.h"
 #include "base/values.h"
-#include "chrome/browser/sync/sync_constants.h"
 #include "chrome/browser/sync/engine/all_status.h"
 #include "chrome/browser/sync/engine/change_reorder_buffer.h"
 #include "chrome/browser/sync/engine/model_safe_worker.h"
diff --git a/chrome/browser/sync/notifier/cache_invalidation_packet_handler.cc b/chrome/browser/sync/notifier/cache_invalidation_packet_handler.cc
index 02e5bfa..de3a6d7 100644
--- a/chrome/browser/sync/notifier/cache_invalidation_packet_handler.cc
+++ b/chrome/browser/sync/notifier/cache_invalidation_packet_handler.cc
@@ -1,4 +1,4 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
@@ -12,7 +12,6 @@
 #include "base/logging.h"
 #include "base/rand_util.h"
 #include "base/string_number_conversions.h"
-#include "chrome/browser/sync/sync_constants.h"
 #include "google/cacheinvalidation/invalidation-client.h"
 #include "jingle/notifier/listener/xml_element_util.h"
 #include "talk/xmpp/constants.h"
@@ -25,6 +24,7 @@ namespace sync_notifier {
 namespace {
 
 const char kBotJid[] = "tango@bot.talk.google.com";
+const char kServiceUrl[] = "http://www.google.com/chrome/sync";
 
 const buzz::QName kQnData("google:notifier", "data");
 const buzz::QName kQnSeq("", "seq");
@@ -167,8 +167,7 @@ class CacheInvalidationSendMessageTask : public buzz::XmppTask {
     iq->AddElement(cache_invalidation_iq_packet);
     cache_invalidation_iq_packet->SetAttr(kQnSeq, base::IntToString(seq));
     cache_invalidation_iq_packet->SetAttr(kQnSid, sid);
-    cache_invalidation_iq_packet->SetAttr(
-        kQnServiceUrl, browser_sync::kSyncNotificationChannel);
+    cache_invalidation_iq_packet->SetAttr(kQnServiceUrl, kServiceUrl);
     cache_invalidation_iq_packet->SetBodyText(msg);
     return iq;
   }
diff --git a/chrome/browser/sync/notifier/invalidation_notifier.cc b/chrome/browser/sync/notifier/invalidation_notifier.cc
new file mode 100644
index 0000000..f1e40f3
--- /dev/null
+++ b/chrome/browser/sync/notifier/invalidation_notifier.cc
@@ -0,0 +1,128 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/notifier/invalidation_notifier.h"
+
+#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
+#include "chrome/browser/sync/protocol/service_constants.h"
+#include "chrome/browser/sync/sessions/session_state.h"
+#include "jingle/notifier/listener/notification_constants.h"
+#include "talk/xmpp/jid.h"
+#include "talk/xmpp/xmppclientsettings.h"
+
+namespace sync_notifier {
+
+InvalidationNotifier::InvalidationNotifier(
+    const notifier::NotifierOptions& notifier_options,
+    const std::string& client_info)
+    : notifier_options_(notifier_options),
+      server_notifier_thread_(notifier_options, client_info, this),
+      logged_in_(false) {
+  server_notifier_thread_.AddObserver(this);
+  server_notifier_thread_.Start();
+}
+
+InvalidationNotifier::~InvalidationNotifier() {
+  if (logged_in_) {
+    server_notifier_thread_.Logout();
+  }
+  server_notifier_thread_.RemoveObserver(this);
+}
+
+void InvalidationNotifier::AddObserver(SyncNotifierObserver* observer) {
+  observer_list_.AddObserver(observer);
+}
+
+void InvalidationNotifier::RemoveObserver(SyncNotifierObserver* observer) {
+  observer_list_.RemoveObserver(observer);
+}
+
+void InvalidationNotifier::SetState(const std::string& state) {
+  server_notifier_thread_.SetState(state);
+}
+
+namespace {
+
+// TODO(akalin): Figure out a clean way to consolidate all
+// XmppClientSettings code, e.g., the code in
+// TalkMediatorImpl::SetAuthToken().
+buzz::XmppClientSettings MakeXmppClientSettings(
+    const std::string& email,
+    const std::string& token,
+    const notifier::NotifierOptions& notifier_options) {
+  buzz::Jid jid = buzz::Jid(email);
+  DCHECK(!jid.node().empty() && jid.IsValid());
+
+  buzz::XmppClientSettings xmpp_client_settings;
+  xmpp_client_settings.set_user(jid.node());
+  xmpp_client_settings.set_resource("chrome-sync");
+  xmpp_client_settings.set_host(jid.domain());
+  xmpp_client_settings.set_use_tls(true);
+  xmpp_client_settings.set_auth_cookie(
+      notifier_options.invalidate_xmpp_login ?
+      token + "bogus" : token);
+  xmpp_client_settings.set_token_service(SYNC_SERVICE_NAME);
+  if (notifier_options.allow_insecure_connection) {
+    xmpp_client_settings.set_allow_plain(true);
+    xmpp_client_settings.set_use_tls(false);
+  }
+  return xmpp_client_settings;
+}
+
+}  // namespace
+
+void InvalidationNotifier::UpdateCredentials(
+    const std::string& email, const std::string& token) {
+  buzz::XmppClientSettings xmpp_client_settings =
+      MakeXmppClientSettings(email, token, notifier_options_);
+  if (logged_in_) {
+    server_notifier_thread_.UpdateXmppSettings(xmpp_client_settings);
+  } else {
+    server_notifier_thread_.Login(xmpp_client_settings);
+    logged_in_ = true;
+  }
+}
+
+void InvalidationNotifier::UpdateEnabledTypes(
+    const syncable::ModelTypeSet& types) {
+  server_notifier_thread_.UpdateEnabledTypes(types);
+}
+
+void InvalidationNotifier::SendNotification() {}
+
+void InvalidationNotifier::WriteState(const std::string& state) {
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
+      StoreState(state));
+}
+
+void InvalidationNotifier::OnConnectionStateChange(bool logged_in) {
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
+                    OnNotificationStateChange(logged_in));
+}
+
+void InvalidationNotifier::OnSubscriptionStateChange(bool subscribed) {}
+
+void InvalidationNotifier::OnIncomingNotification(
+    const notifier::Notification& notification) {
+  VLOG(1) << "Sync received server notification from "
+          << notification.channel << ": " << notification.data;
+  const std::string& model_type_list = notification.channel;
+  const std::string& notification_payload = notification.data;
+
+  syncable::ModelTypeBitSet model_types;
+  if (!syncable::ModelTypeBitSetFromString(model_type_list, &model_types)) {
+    LOG(DFATAL) << "Could not extract model types from server data.";
+    model_types.set();
+  }
+
+  browser_sync::sessions::TypePayloadMap type_payloads =
+      browser_sync::sessions::MakeTypePayloadMapFromBitSet(
+          model_types, notification_payload);
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
+                    OnIncomingNotification(type_payloads));
+}
+
+void InvalidationNotifier::OnOutgoingNotification() {}
+
+}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/invalidation_notifier.h b/chrome/browser/sync/notifier/invalidation_notifier.h
new file mode 100644
index 0000000..6ee27e9
--- /dev/null
+++ b/chrome/browser/sync/notifier/invalidation_notifier.h
@@ -0,0 +1,63 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+//
+// A notifier that communicates to sync servers for server-issued
+// notifications.
+
+#ifndef CHROME_BROWSER_SYNC_NOTIFIER_INVALIDATION_NOTIFIER_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_INVALIDATION_NOTIFIER_H_
+
+#include <string>
+
+#include "base/observer_list.h"
+#include "chrome/browser/sync/notifier/server_notifier_thread.h"
+#include "chrome/browser/sync/notifier/state_writer.h"
+#include "chrome/browser/sync/notifier/sync_notifier.h"
+#include "chrome/browser/sync/syncable/model_type.h"
+#include "jingle/notifier/base/notifier_options.h"
+
+namespace sync_notifier {
+
+class InvalidationNotifier
+    : public SyncNotifier,
+      public StateWriter,
+      public ServerNotifierThread::Observer {
+ public:
+  InvalidationNotifier(const notifier::NotifierOptions& notifier_options,
+                       const std::string& client_info);
+
+  virtual ~InvalidationNotifier();
+
+  // SyncNotifier implementation.
+  virtual void AddObserver(SyncNotifierObserver* observer);
+  virtual void RemoveObserver(SyncNotifierObserver* observer);
+  virtual void SetState(const std::string& state);
+  virtual void UpdateCredentials(
+      const std::string& email, const std::string& token);
+  virtual void UpdateEnabledTypes(const syncable::ModelTypeSet& types);
+  virtual void SendNotification();
+
+  // StateWriter implementation.
+  virtual void WriteState(const std::string& state);
+
+  // ServerNotifierThread::Observer implementation.
+  virtual void OnConnectionStateChange(bool logged_in);
+  virtual void OnSubscriptionStateChange(bool subscribed);
+  virtual void OnIncomingNotification(
+      const notifier::Notification& notification);
+  virtual void OnOutgoingNotification();
+
+ private:
+  const notifier::NotifierOptions notifier_options_;
+  // The actual notification listener.
+  ServerNotifierThread server_notifier_thread_;
+  // Whether we called Login() on |server_notifier_thread_| yet.
+  bool logged_in_;
+
+  ObserverList<SyncNotifierObserver> observer_list_;
+  syncable::ModelTypeSet enabled_types_;
+};
+
+}
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_INVALIDATION_NOTIFIER_H_
diff --git a/chrome/browser/sync/notifier/p2p_notifier.cc b/chrome/browser/sync/notifier/p2p_notifier.cc
new file mode 100644
index 0000000..2215d1b
--- /dev/null
+++ b/chrome/browser/sync/notifier/p2p_notifier.cc
@@ -0,0 +1,127 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/notifier/p2p_notifier.h"
+
+#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
+#include "chrome/browser/sync/protocol/service_constants.h"
+#include "chrome/browser/sync/sessions/session_state.h"
+#include "jingle/notifier/listener/mediator_thread_impl.h"
+#include "jingle/notifier/listener/talk_mediator_impl.h"
+
+namespace sync_notifier {
+
+namespace {
+const char kSyncNotificationChannel[] = "http://www.google.com/chrome/sync";
+const char kSyncNotificationData[] = "sync-ping-p2p";
+}  // namespace
+
+P2PNotifier::P2PNotifier(
+    const notifier::NotifierOptions& notifier_options)
+    : talk_mediator_(
+        new notifier::TalkMediatorImpl(
+            new notifier::MediatorThreadImpl(notifier_options),
+            notifier_options.invalidate_xmpp_login,
+            notifier_options.allow_insecure_connection)),
+      logged_in_(false),
+      notifications_enabled_(false) {
+  talk_mediator_->SetDelegate(this);
+}
+
+P2PNotifier::~P2PNotifier() {}
+
+void P2PNotifier::AddObserver(SyncNotifierObserver* observer) {
+  observer_list_.AddObserver(observer);
+}
+
+void P2PNotifier::RemoveObserver(SyncNotifierObserver* observer) {
+  observer_list_.RemoveObserver(observer);
+}
+
+void P2PNotifier::SetState(const std::string& state) {}
+
+void P2PNotifier::UpdateCredentials(
+    const std::string& email, const std::string& token) {
+  // If already logged in, the new credentials will take effect on the
+  // next reconnection.
+  if (!talk_mediator_->SetAuthToken(email, token, SYNC_SERVICE_NAME)) {
+    LOG(DFATAL) << "Could not set auth token for " << email;
+    return;
+  }
+  if (!logged_in_) {
+    if (!talk_mediator_->Login()) {
+      LOG(DFATAL) << "Could not login for " << email;
+      return;
+    }
+
+    notifier::Subscription subscription;
+    subscription.channel = kSyncNotificationChannel;
+    // 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;
+    talk_mediator_->AddSubscription(subscription);
+
+    logged_in_ = true;
+  }
+}
+
+void P2PNotifier::UpdateEnabledTypes(const syncable::ModelTypeSet& types) {
+  enabled_types_ = types;
+  MaybeEmitNotification();
+}
+
+void P2PNotifier::SendNotification() {
+  VLOG(1) << "Sending XMPP notification...";
+  notifier::Notification notification;
+  notification.channel = kSyncNotificationChannel;
+  notification.data = kSyncNotificationData;
+  bool success = talk_mediator_->SendNotification(notification);
+  if (success) {
+    VLOG(1) << "Sent XMPP notification";
+  } else {
+    VLOG(1) << "Could not send XMPP notification";
+  }
+}
+
+void P2PNotifier::OnNotificationStateChange(bool notifications_enabled) {
+  notifications_enabled_ = notifications_enabled;
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
+      OnNotificationStateChange(notifications_enabled_));
+  MaybeEmitNotification();
+}
+
+void P2PNotifier::OnIncomingNotification(
+    const notifier::Notification& notification) {
+  VLOG(1) << "Sync received P2P notification.";
+  if (notification.channel != kSyncNotificationChannel) {
+    LOG(WARNING) << "Notification fron unexpected source: "
+                 << notification.channel;
+  }
+  MaybeEmitNotification();
+}
+
+void P2PNotifier::OnOutgoingNotification() {}
+
+void P2PNotifier::MaybeEmitNotification() {
+  if (!logged_in_) {
+    VLOG(1) << "Not logged in yet -- not emitting notification";
+    return;
+  }
+  if (!notifications_enabled_) {
+    VLOG(1) << "Notifications not enabled -- not emitting notification";
+    return;
+  }
+  if (enabled_types_.empty()) {
+    VLOG(1) << "No enabled types -- not emitting notification";
+    return;
+  }
+  browser_sync::sessions::TypePayloadMap type_payloads =
+      browser_sync::sessions::MakeTypePayloadMapFromBitSet(
+          syncable::ModelTypeBitSetFromSet(enabled_types_), std::string());
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
+                    OnIncomingNotification(type_payloads));
+}
+
+}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/sync_notifier_impl.h b/chrome/browser/sync/notifier/p2p_notifier.h
index c9151bc..572c222 100644
--- a/chrome/browser/sync/notifier/sync_notifier_impl.h
+++ b/chrome/browser/sync/notifier/p2p_notifier.h
@@ -2,71 +2,66 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 //
-// Talk Mediator based implementation of the sync notifier interface.
-// Initializes the talk mediator and registers itself as the delegate.
+// A notifier that uses p2p notifications based on XMPP push
+// notifications.  Used only for sync integration tests.
 
-#ifndef CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_IMPL_H_
-#define CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_IMPL_H_
+#ifndef CHROME_BROWSER_SYNC_NOTIFIER_P2P_NOTIFIER_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_P2P_NOTIFIER_H_
 
 #include <string>
 
 #include "base/observer_list.h"
-#include "chrome/browser/sync/notifier/state_writer.h"
+#include "base/scoped_ptr.h"
 #include "chrome/browser/sync/notifier/sync_notifier.h"
 #include "chrome/browser/sync/syncable/model_type.h"
-#include "jingle/notifier/base/notifier_options.h"
 #include "jingle/notifier/listener/talk_mediator.h"
-#include "jingle/notifier/listener/talk_mediator_impl.h"
+
+namespace notifier {
+struct NotifierOptions;
+}  // namespace
 
 namespace sync_notifier {
-class ServerNotifierThread;
 
-class SyncNotifierImpl
+class P2PNotifier
     : public SyncNotifier,
-      public sync_notifier::StateWriter,
       public notifier::TalkMediator::Delegate {
  public:
-  SyncNotifierImpl(const notifier::NotifierOptions& notifier_options,
-                   const std::string& client_info);
-
-  virtual ~SyncNotifierImpl();
-
-  // TalkMediator::Delegate implementation.
-  virtual void OnNotificationStateChange(bool notifications_enabled);
-
-  virtual void OnIncomingNotification(
-      const notifier::Notification& notification);
+  explicit P2PNotifier(const notifier::NotifierOptions& notifier_options);
 
-  virtual void OnOutgoingNotification() {}
-
-  // sync_notifier::StateWriter implementation.
-  virtual void WriteState(const std::string& state);
+  virtual ~P2PNotifier();
 
   // SyncNotifier implementation
-  virtual void UpdateCredentials(
-      const std::string& email, const std::string& token);
-
-  virtual void SetState(const std::string& state);
-
   virtual void AddObserver(SyncNotifierObserver* observer);
   virtual void RemoveObserver(SyncNotifierObserver* observer);
-
+  virtual void SetState(const std::string& state);
+  virtual void UpdateCredentials(
+      const std::string& email, const std::string& token);
   virtual void UpdateEnabledTypes(const syncable::ModelTypeSet& types);
   virtual void SendNotification();
+
+  // TalkMediator::Delegate implementation.
+  virtual void OnNotificationStateChange(bool notifications_enabled);
+  virtual void OnIncomingNotification(
+      const notifier::Notification& notification);
+  virtual void OnOutgoingNotification();
+
  private:
-  // Login to the talk mediator with the given credentials.
-  void TalkMediatorLogin(
-      const std::string& email, const std::string& token);
+  // Call OnIncomingNotification() on observers if we have a non-empty
+  // set of enabled types.
+  void MaybeEmitNotification();
+
+  ObserverList<SyncNotifierObserver> observer_list_;
 
-  // Notification (xmpp) handler.
+  // The actual notification listener.
   scoped_ptr<notifier::TalkMediator> talk_mediator_;
-  syncable::ModelTypeSet enabled_types_;
-  std::string state_;
+  // Whether we called Login() on |talk_mediator_| yet.
+  bool logged_in_;
+  // Whether |talk_mediator_| has notified us that notifications are
+  // enabled.
+  bool notifications_enabled_;
 
-  notifier::NotifierOptions notifier_options_;
-  const std::string client_info_;
-  ServerNotifierThread* server_notifier_thread_;
-  ObserverList<SyncNotifierObserver> observer_list_;
+  syncable::ModelTypeSet enabled_types_;
 };
+
 }
-#endif  // CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_IMPL_H_
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_P2P_NOTIFIER_H_
diff --git a/chrome/browser/sync/notifier/server_notifier_thread.cc b/chrome/browser/sync/notifier/server_notifier_thread.cc
index 2476710..9c980275 100644
--- a/chrome/browser/sync/notifier/server_notifier_thread.cc
+++ b/chrome/browser/sync/notifier/server_notifier_thread.cc
@@ -9,19 +9,19 @@
 
 #include "base/logging.h"
 #include "chrome/browser/sync/notifier/chrome_invalidation_client.h"
+#include "chrome/browser/sync/protocol/service_constants.h"
 #include "jingle/notifier/base/notifier_options.h"
 #include "jingle/notifier/listener/notification_defines.h"
 #include "talk/xmpp/xmppclient.h"
+#include "talk/xmpp/xmppclientsettings.h"
 
 namespace sync_notifier {
 
 ServerNotifierThread::ServerNotifierThread(
     const notifier::NotifierOptions& notifier_options,
-    const std::string& client_info, const std::string& state,
-    StateWriter* state_writer)
+    const std::string& client_info, StateWriter* state_writer)
     : notifier::MediatorThreadImpl(notifier_options),
       client_info_(client_info),
-      state_(state),
       state_writers_(new ObserverListThreadSafe<StateWriter>()),
       state_writer_(state_writer) {
   DCHECK_EQ(notifier::NOTIFICATION_SERVER,
@@ -32,41 +32,24 @@ ServerNotifierThread::ServerNotifierThread(
 
 ServerNotifierThread::~ServerNotifierThread() {}
 
-void ServerNotifierThread::ListenForUpdates() {
+void ServerNotifierThread::SetState(const std::string& state) {
   DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  worker_message_loop()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(this, &ServerNotifierThread::DoListenForUpdates));
-}
-
-void ServerNotifierThread::SubscribeForUpdates(
-    const notifier::SubscriptionList& subscriptions) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  worker_message_loop()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this, &ServerNotifierThread::RegisterTypes));
-
-  worker_message_loop()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this, &ServerNotifierThread::SignalSubscribed));
+  state_ = state;
 }
 
 void ServerNotifierThread::UpdateEnabledTypes(
     const syncable::ModelTypeSet& types) {
   DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+
   worker_message_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(
-          this,
-          &ServerNotifierThread::SetRegisteredTypes,
-          types));
+          this, &ServerNotifierThread::DoUpdateEnabledTypes, types));
 
   worker_message_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(
-          this, &ServerNotifierThread::RegisterTypes));
+          this, &ServerNotifierThread::RegisterEnabledTypes));
 }
 
 void ServerNotifierThread::Logout() {
@@ -80,11 +63,32 @@ void ServerNotifierThread::Logout() {
   MediatorThreadImpl::Logout();
 }
 
-void ServerNotifierThread::SendNotification(
-    const notifier::Notification& notification) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  NOTREACHED() << "Shouldn't send notifications if ServerNotifierThread is "
-                  "used";
+void ServerNotifierThread::OnConnect(
+    base::WeakPtr<talk_base::Task> base_task) {
+  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
+  // Sets |base_task_|.
+  MediatorThreadImpl::OnConnect(base_task);
+  if (chrome_invalidation_client_.get()) {
+    // If we already have an invalidation client, simply change the
+    // base task.
+    chrome_invalidation_client_->ChangeBaseTask(base_task_);
+  } else {
+    // Otherwise, create the invalidation client.
+    chrome_invalidation_client_.reset(new ChromeInvalidationClient());
+
+    // TODO(akalin): Make cache_guid() part of the client ID.  If we do
+    // so and we somehow propagate it up to the server somehow, we can
+    // make it so that we won't receive any notifications that were
+    // generated from our own changes.
+    const std::string kClientId = "server_notifier_thread";
+    // It's okay to read/write |state_| here, since it was written
+    // before the worker thread was created and is never read/written
+    // to by the parent thread after that.
+    chrome_invalidation_client_->Start(
+        kClientId, client_info_, state_, this, this, base_task_);
+    state_.clear();
+    RegisterEnabledTypes();
+  }
 }
 
 void ServerNotifierThread::OnInvalidate(
@@ -121,42 +125,23 @@ void ServerNotifierThread::WriteState(const std::string& state) {
   state_writers_->Notify(&StateWriter::WriteState, state);
 }
 
-void ServerNotifierThread::DoListenForUpdates() {
-  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  if (!base_task_.get()) {
-    return;
-  }
-
-  if (chrome_invalidation_client_.get()) {
-    // If we already have an invalidation client, simply change the
-    // base task.
-    chrome_invalidation_client_->ChangeBaseTask(base_task_);
-  } else {
-    // Otherwise, create the invalidation client.
-    chrome_invalidation_client_.reset(new ChromeInvalidationClient());
+void ServerNotifierThread::SendNotification(
+    const notifier::Notification& notification) {
+  NOTREACHED();
+}
 
-    // TODO(akalin): Make cache_guid() part of the client ID.  If we do
-    // so and we somehow propagate it up to the server somehow, we can
-    // make it so that we won't receive any notifications that were
-    // generated from our own changes.
-    const std::string kClientId = "server_notifier_thread";
-    chrome_invalidation_client_->Start(
-        kClientId, client_info_, state_, this, this, base_task_);
-    RegisterTypes();
-    state_.clear();
-  }
+void ServerNotifierThread::DoUpdateEnabledTypes(
+    const syncable::ModelTypeSet& types) {
+  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
+  enabled_types_ = types;
 }
 
-void ServerNotifierThread::RegisterTypes() {
+void ServerNotifierThread::RegisterEnabledTypes() {
   DCHECK_EQ(MessageLoop::current(), worker_message_loop());
   if (!chrome_invalidation_client_.get()) {
     return;
   }
-  chrome_invalidation_client_->RegisterTypes(registered_types_);
-}
-
-void ServerNotifierThread::SignalSubscribed() {
-  observers_->Notify(&Observer::OnSubscriptionStateChange, true);
+  chrome_invalidation_client_->RegisterTypes(enabled_types_);
 }
 
 void ServerNotifierThread::StopInvalidationListener() {
@@ -164,9 +149,4 @@ void ServerNotifierThread::StopInvalidationListener() {
   chrome_invalidation_client_.reset();
 }
 
-void ServerNotifierThread::SetRegisteredTypes(
-    const syncable::ModelTypeSet& types) {
-  registered_types_ = types;
-}
-
 }  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/server_notifier_thread.h b/chrome/browser/sync/notifier/server_notifier_thread.h
index 1d4af47..e09811e 100644
--- a/chrome/browser/sync/notifier/server_notifier_thread.h
+++ b/chrome/browser/sync/notifier/server_notifier_thread.h
@@ -5,9 +5,8 @@
 // This class is the (hackish) way to use the XMPP parts of
 // MediatorThread for server-issued notifications.
 //
-// TODO(akalin): Decomp MediatorThread into an XMPP service part and a
-// notifications-specific part and use the XMPP service part for
-// server-issued notifications.
+// TODO(akalin): Separate this code from notifier::MediatorThreadImpl
+// and combine it with InvalidationNotifier.
 
 #ifndef CHROME_BROWSER_SYNC_NOTIFIER_SERVER_NOTIFIER_THREAD_H_
 #define CHROME_BROWSER_SYNC_NOTIFIER_SERVER_NOTIFIER_THREAD_H_
@@ -40,23 +39,22 @@ class ServerNotifierThread
   explicit ServerNotifierThread(
       const notifier::NotifierOptions& notifier_options,
       const std::string& client_info,
-      const std::string& state, StateWriter* state_writer);
+      StateWriter* state_writer);
 
   virtual ~ServerNotifierThread();
 
-  // Overridden to start listening to server notifications.
-  virtual void ListenForUpdates();
+  // Should be called exactly once before Login().
+  void SetState(const std::string& state);
 
-  // Overridden to immediately notify the delegate that subscriptions
-  // (i.e., notifications) are on.  Must be called only after a call
-  // to ListenForUpdates().
-  virtual void SubscribeForUpdates(
-      const notifier::SubscriptionList& subscriptions);
+  // Must be called after Start() is called and before Logout() is
+  // called.
+  void UpdateEnabledTypes(const syncable::ModelTypeSet& types);
 
-  // Overridden to stop listening to server notifications.
+  // Overridden to stop the invalidation listener.
   virtual void Logout();
 
-  virtual void SendNotification(const notifier::Notification& data);
+  // Overridden to do actual login.
+  virtual void OnConnect(base::WeakPtr<talk_base::Task> base_task);
 
   // ChromeInvalidationClient::Listener implementation.
   // We pass on two pieces of information to observers through the
@@ -65,6 +63,9 @@ class ServerNotifierThread
   //       |channel| member.
   // - the invalidation payload for that model type, through the
   //       Notification's |data| member.
+  //
+  // TODO(akalin): Remove this hack once we merge with
+  // InvalidationNotifier.
   virtual void OnInvalidate(syncable::ModelType model_type,
                             const std::string& payload);
   virtual void OnInvalidateAll();
@@ -72,22 +73,21 @@ class ServerNotifierThread
   // StateWriter implementation.
   virtual void WriteState(const std::string& state);
 
-  virtual void UpdateEnabledTypes(const syncable::ModelTypeSet& types);
-
  private:
-  // Posted to the worker thread by ListenForUpdates().
-  void DoListenForUpdates();
+  // Made private because this function shouldn't be called.
+  virtual void SendNotification(const notifier::Notification& data);
 
-  // Posted to the worker thread by SubscribeForUpdates().
-  void RegisterTypes();
+  // Posted to the worker thread by UpdateEnabledTypes().
+  void DoUpdateEnabledTypes(const syncable::ModelTypeSet& types);
 
-  void SignalSubscribed();
+  // Posted to the worker thread by UpdateEnabledTypes() and also
+  // called by OnConnect().
+  void RegisterEnabledTypes();
 
   // Posted to the worker thread by Logout().
   void StopInvalidationListener();
 
   const std::string client_info_;
-  std::string state_;
   // Hack to get the nice thread-safe behavior for |state_writer_|.
   scoped_refptr<ObserverListThreadSafe<StateWriter> > state_writers_;
   // We still need to keep |state_writer_| around to remove it from
@@ -95,9 +95,12 @@ class ServerNotifierThread
   StateWriter* state_writer_;
   scoped_ptr<ChromeInvalidationClient> chrome_invalidation_client_;
 
-  syncable::ModelTypeSet registered_types_;
+  // The state to pass to |chrome_invalidation_client_|.
+  std::string state_;
 
-  void SetRegisteredTypes(const syncable::ModelTypeSet& types);
+  // The types to register.  Should only be accessed on the worker
+  // thread.
+  syncable::ModelTypeSet enabled_types_;
 };
 
 }  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/server_notifier_thread_unittest.cc b/chrome/browser/sync/notifier/server_notifier_thread_unittest.cc
index bb35411..15072bd 100644
--- a/chrome/browser/sync/notifier/server_notifier_thread_unittest.cc
+++ b/chrome/browser/sync/notifier/server_notifier_thread_unittest.cc
@@ -22,7 +22,7 @@ class FakeServerNotifierThread : public ServerNotifierThread {
  public:
    FakeServerNotifierThread()
       : ServerNotifierThread(notifier::NotifierOptions(),
-                             "fake client info", "fake state",
+                             "fake client info",
                              ALLOW_THIS_IN_INITIALIZER_LIST(this)) {}
 
   virtual ~FakeServerNotifierThread() {}
@@ -129,30 +129,24 @@ TEST_F(ServerNotifierThreadTest, Basic) {
   server_notifier_thread.Start();
   server_notifier_thread.Login(buzz::XmppClientSettings());
   server_notifier_thread.SimulateConnect();
-  server_notifier_thread.ListenForUpdates();
-  server_notifier_thread.SubscribeForUpdates(notifier::SubscriptionList());
   server_notifier_thread.Logout();
 }
 
-TEST_F(ServerNotifierThreadTest, DisconnectBeforeListen) {
+TEST_F(ServerNotifierThreadTest, LoginLogout) {
   FakeServerNotifierThread server_notifier_thread;
 
   server_notifier_thread.Start();
   server_notifier_thread.Login(buzz::XmppClientSettings());
-  server_notifier_thread.ListenForUpdates();
-  server_notifier_thread.SubscribeForUpdates(notifier::SubscriptionList());
   server_notifier_thread.Logout();
 }
 
-TEST_F(ServerNotifierThreadTest, Disconnected) {
+TEST_F(ServerNotifierThreadTest, DisconnectLogout) {
   FakeServerNotifierThread server_notifier_thread;
 
   server_notifier_thread.Start();
   server_notifier_thread.Login(buzz::XmppClientSettings());
   server_notifier_thread.SimulateConnect();
   server_notifier_thread.SimulateDisconnect();
-  server_notifier_thread.ListenForUpdates();
-  server_notifier_thread.SubscribeForUpdates(notifier::SubscriptionList());
   server_notifier_thread.Logout();
 }
 
diff --git a/chrome/browser/sync/notifier/sync_notifier_factory.cc b/chrome/browser/sync/notifier/sync_notifier_factory.cc
index 3603d5a..03ca84c 100644
--- a/chrome/browser/sync/notifier/sync_notifier_factory.cc
+++ b/chrome/browser/sync/notifier/sync_notifier_factory.cc
@@ -9,8 +9,9 @@
 #include "base/command_line.h"
 #include "base/string_number_conversions.h"
 #include "base/string_util.h"
+#include "chrome/browser/sync/notifier/invalidation_notifier.h"
+#include "chrome/browser/sync/notifier/p2p_notifier.h"
 #include "chrome/browser/sync/notifier/sync_notifier.h"
-#include "chrome/browser/sync/notifier/sync_notifier_impl.h"
 #include "chrome/common/chrome_switches.h"
 #include "jingle/notifier/base/notifier_options.h"
 #include "jingle/notifier/communicator/const_communicator.h"
@@ -83,7 +84,11 @@ SyncNotifier* CreateDefaultSyncNotifier(const CommandLine& command_line,
         notifier::StringToNotificationMethod(notification_method_str);
   }
 
-  return new SyncNotifierImpl(notifier_options, client_info);
+  if (notifier_options.notification_method == notifier::NOTIFICATION_P2P) {
+    return new P2PNotifier(notifier_options);
+  }
+
+  return new InvalidationNotifier(notifier_options, client_info);
 }
 }  // namespace
 
diff --git a/chrome/browser/sync/notifier/sync_notifier_impl.cc b/chrome/browser/sync/notifier/sync_notifier_impl.cc
deleted file mode 100644
index ac501f1..0000000
--- a/chrome/browser/sync/notifier/sync_notifier_impl.cc
+++ /dev/null
@@ -1,193 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "chrome/browser/sync/notifier/sync_notifier_impl.h"
-
-#include "chrome/browser/sync/notifier/server_notifier_thread.h"
-#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
-#include "chrome/browser/sync/protocol/service_constants.h"
-#include "chrome/browser/sync/sessions/session_state.h"
-#include "chrome/browser/sync/sync_constants.h"
-#include "chrome/browser/sync/syncable/model_type.h"
-#include "jingle/notifier/listener/mediator_thread_impl.h"
-#include "jingle/notifier/listener/notification_constants.h"
-
-// TODO(akalin): Split this class into two implementations - one for p2p and
-// one for server issued notifications.
-using notifier::TalkMediator;
-using notifier::TalkMediatorImpl;
-namespace sync_notifier {
-
-SyncNotifierImpl::SyncNotifierImpl(
-    const notifier::NotifierOptions& notifier_options,
-    const std::string& client_info)
-    : notifier_options_(notifier_options),
-      client_info_(client_info),
-      server_notifier_thread_(NULL) { }
-
-SyncNotifierImpl::~SyncNotifierImpl() {
-  scoped_ptr<TalkMediator> talk_mediator(talk_mediator_.release());
-
-  // Shutdown the xmpp buzz connection.
-  if (talk_mediator.get()) {
-    VLOG(1) << "P2P: Mediator logout started.";
-    talk_mediator->Logout();
-    VLOG(1) << "P2P: Mediator logout completed.";
-    talk_mediator.reset();
-
-    // |server_notifier_thread_| is owned by |talk_mediator|. We NULL
-    // it out here so as to not have a dangling pointer.
-    server_notifier_thread_= NULL;
-    VLOG(1) << "P2P: Mediator destroyed.";
-  }
-}
-
-void SyncNotifierImpl::AddObserver(SyncNotifierObserver* observer) {
-  observer_list_.AddObserver(observer);
-}
-
-void SyncNotifierImpl::RemoveObserver(SyncNotifierObserver* observer) {
-  observer_list_.RemoveObserver(observer);
-}
-
-void SyncNotifierImpl::OnNotificationStateChange(bool notifications_enabled) {
-  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
-      OnNotificationStateChange(notifications_enabled));
-
-  // If using p2p notifications, generate a notification for all enabled types.
-  // Used only for tests.
-  if ((notifier_options_.notification_method !=
-      notifier::NOTIFICATION_SERVER) && notifications_enabled) {
-    browser_sync::sessions::TypePayloadMap model_types_with_payloads =
-        browser_sync::sessions::MakeTypePayloadMapFromBitSet(
-            syncable::ModelTypeBitSetFromSet(enabled_types_), std::string());
-    FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
-        OnIncomingNotification(model_types_with_payloads));
-  }
-}
-
-void SyncNotifierImpl::OnIncomingNotification(
-    const notifier::Notification& notification) {
-  browser_sync::sessions::TypePayloadMap model_types_with_payloads;
-
-  // Check if the service url is a sync URL.  An empty service URL is
-  // treated as a legacy sync notification.  If we're listening to
-  // server-issued notifications, no need to check the service_url.
-  if (notifier_options_.notification_method ==
-      notifier::NOTIFICATION_SERVER) {
-    VLOG(1) << "Sync received server notification from " <<
-        notification.channel << ": " << notification.data;
-    syncable::ModelTypeBitSet model_types;
-    const std::string& model_type_list = notification.channel;
-    const std::string& notification_payload = notification.data;
-
-    if (!syncable::ModelTypeBitSetFromString(model_type_list, &model_types)) {
-      LOG(DFATAL) << "Could not extract model types from server data.";
-      model_types.set();
-    }
-
-    model_types_with_payloads =
-        browser_sync::sessions::MakeTypePayloadMapFromBitSet(model_types,
-            notification_payload);
-  } else if (notification.channel == browser_sync::kSyncNotificationChannel) {
-    VLOG(1) << "Sync received P2P notification.";
-
-    // Catch for sync integration tests (uses p2p). Just set all enabled
-    // datatypes.
-    model_types_with_payloads =
-        browser_sync::sessions::MakeTypePayloadMapFromBitSet(
-            syncable::ModelTypeBitSetFromSet(enabled_types_), std::string());
-  } else {
-    LOG(WARNING) << "Notification fron unexpected source: "
-                 << notification.channel;
-  }
-
-  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
-      OnIncomingNotification(model_types_with_payloads));
-}
-
-void SyncNotifierImpl::WriteState(const std::string& state) {
-  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
-      StoreState(state));
-}
-
-void SyncNotifierImpl::UpdateCredentials(
-    const std::string& email, const std::string& token) {
-  // Reset talk_mediator_ before creating ServerNotifierThread/MediatorThread,
-  // to avoid any problems with having two of those threads at the same time.
-  talk_mediator_.reset();
-
-  if (notifier_options_.notification_method ==
-      notifier::NOTIFICATION_SERVER) {
-    // |talk_mediator_| takes ownership of |sync_notifier_thread_|
-    // but it is guaranteed that |sync_notifier_thread_| is destroyed only
-    // when |talk_mediator_| is (see the comments in talk_mediator.h).
-    server_notifier_thread_ = new sync_notifier::ServerNotifierThread(
-        notifier_options_, client_info_, state_, this);
-    talk_mediator_.reset(
-        new TalkMediatorImpl(server_notifier_thread_,
-                             notifier_options_.invalidate_xmpp_login,
-                             notifier_options_.allow_insecure_connection));
-
-    // Since we may be initialized more than once, make sure that any
-    // newly created server notifier thread has the latest enabled types.
-    server_notifier_thread_->UpdateEnabledTypes(enabled_types_);
-  } else {
-    notifier::MediatorThread* mediator_thread =
-        new notifier::MediatorThreadImpl(notifier_options_);
-    talk_mediator_.reset(
-        new TalkMediatorImpl(mediator_thread,
-                             notifier_options_.invalidate_xmpp_login,
-                             notifier_options_.allow_insecure_connection));
-    notifier::Subscription subscription;
-    subscription.channel = browser_sync::kSyncNotificationChannel;
-    // 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;
-    talk_mediator_->AddSubscription(subscription);
-    server_notifier_thread_ = NULL;
-  }
-  talk_mediator_->SetDelegate(this);
-  if (!talk_mediator_->SetAuthToken(email, token, SYNC_SERVICE_NAME)) {
-    LOG(ERROR) << "Could not set auth info for " << email;
-  }
-  talk_mediator_->Login();
-}
-
-void SyncNotifierImpl::SetState(const std::string& state) {
-  state_ = state;
-}
-
-void SyncNotifierImpl::UpdateEnabledTypes(const syncable::ModelTypeSet& types) {
-  enabled_types_ = types;
-  if (server_notifier_thread_ != NULL) {
-    server_notifier_thread_->UpdateEnabledTypes(types);
-  }
-}
-
-void SyncNotifierImpl::SendNotification() {
-  // Do nothing if we are using server based notifications.
-  if (notifier_options_.notification_method ==
-      notifier::NOTIFICATION_SERVER) {
-    return;
-  }
-
-  if (!talk_mediator_.get()) {
-    NOTREACHED() << "Cannot send notification: talk_mediator_ is NULL";
-    return;
-  }
-
-  VLOG(1) << "Sending XMPP notification...";
-  notifier::Notification notification;
-  notification.channel = browser_sync::kSyncNotificationChannel;
-  notification.data = browser_sync::kSyncNotificationData;
-  bool success = talk_mediator_->SendNotification(notification);
-  if (success) {
-    VLOG(1) << "Sent XMPP notification";
-  } else {
-    VLOG(1) << "Could not send XMPP notification";
-  }
-}
-}  // namespace sync_notifier
diff --git a/chrome/browser/sync/profile_sync_service_unittest.cc b/chrome/browser/sync/profile_sync_service_unittest.cc
index 7c30844..f2d5a42 100644
--- a/chrome/browser/sync/profile_sync_service_unittest.cc
+++ b/chrome/browser/sync/profile_sync_service_unittest.cc
@@ -991,11 +991,13 @@ TEST_F(ProfileSyncServiceTest,
   EXPECT_CALL(event_handler,
               HandleJsEvent("onSyncServiceStateChanged",
                             HasArgs(JsArgList()))).Times(3);
-  // For some reason, these two events don't fire on Linux.
+  // For some reason, these events may or may not fire.
   EXPECT_CALL(event_handler, HandleJsEvent("onChangesApplied", _))
       .Times(AtMost(1));
   EXPECT_CALL(event_handler, HandleJsEvent("onChangesComplete", _))
       .Times(AtMost(1));
+  EXPECT_CALL(event_handler, HandleJsEvent("onSyncNotificationStateChange", _))
+      .Times(AtMost(1));
 
   EXPECT_EQ(NULL, service_->GetBackendForTest());
   EXPECT_FALSE(service_->sync_initialized());
@@ -1076,11 +1078,13 @@ TEST_F(ProfileSyncServiceTest,
   StartSyncServiceAndSetInitialSyncEnded(true, false);
 
   StrictMock<MockJsEventHandler> event_handler;
-  // For some reason, these two events don't fire on Linux.
+  // For some reason, these events may or may not fire.
   EXPECT_CALL(event_handler, HandleJsEvent("onChangesApplied", _))
       .Times(AtMost(1));
   EXPECT_CALL(event_handler, HandleJsEvent("onChangesComplete", _))
       .Times(AtMost(1));
+  EXPECT_CALL(event_handler, HandleJsEvent("onSyncNotificationStateChange", _))
+      .Times(AtMost(1));
 
   ListValue arg_list1;
   arg_list1.Append(Value::CreateBooleanValue(true));
diff --git a/chrome/browser/sync/sync_constants.cc b/chrome/browser/sync/sync_constants.cc
deleted file mode 100644
index 8cca452..0000000
--- a/chrome/browser/sync/sync_constants.cc
+++ /dev/null
@@ -1,13 +0,0 @@
-// Copyright (c) 2010 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 "chrome/browser/sync/sync_constants.h"
-
-namespace browser_sync {
-
-const char kSyncNotificationChannel[] = "http://www.google.com/chrome/sync";
-const char kSyncNotificationData[] = "sync-ping-p2p";
-
-}  // namespace browser_sync
-
diff --git a/chrome/browser/sync/sync_constants.h b/chrome/browser/sync/sync_constants.h
deleted file mode 100644
index 0076106..0000000
--- a/chrome/browser/sync/sync_constants.h
+++ /dev/null
@@ -1,17 +0,0 @@
-// Copyright (c) 2010 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 CHROME_BROWSER_SYNC_SYNC_CONSTANTS_H_
-#define CHROME_BROWSER_SYNC_SYNC_CONSTANTS_H_
-#pragma once
-
-namespace browser_sync {
-
-extern const char kSyncNotificationChannel[];
-extern const char kSyncNotificationData[];
-
-}  // namespace browser_sync
-
-#endif  // CHROME_BROWSER_SYNC_SYNC_CONSTANTS_H_
-
diff --git a/chrome/browser/sync/test_profile_sync_service.cc b/chrome/browser/sync/test_profile_sync_service.cc
index ed4668e..b2e4e44 100644
--- a/chrome/browser/sync/test_profile_sync_service.cc
+++ b/chrome/browser/sync/test_profile_sync_service.cc
@@ -96,7 +96,7 @@ sync_api::HttpPostProviderFactory*
 
 void SyncBackendHostForProfileSyncTest::InitCore(
     const Core::DoInitializeOptions& options) {
-  std::wstring user = L"testuser";
+  std::wstring user = L"testuser@gmail.com";
   core_loop()->PostTask(
       FROM_HERE,
       NewRunnableMethod(core_.get(),
diff --git a/chrome/chrome.gyp b/chrome/chrome.gyp
index d89f17f..0471d6e 100644
--- a/chrome/chrome.gyp
+++ b/chrome/chrome.gyp
@@ -744,8 +744,6 @@
         'browser/sync/sessions/sync_session.h',
         'browser/sync/sessions/sync_session_context.cc',
         'browser/sync/sessions/sync_session_context.h',
-        'browser/sync/sync_constants.cc',
-        'browser/sync/sync_constants.h',
         'browser/sync/syncable/autofill_migration.h',
         'browser/sync/syncable/blob.h',
         'browser/sync/syncable/dir_open_result.h',
@@ -839,8 +837,12 @@
         'browser/sync/notifier/chrome_invalidation_client.h',
         'browser/sync/notifier/chrome_system_resources.cc',
         'browser/sync/notifier/chrome_system_resources.h',
+        'browser/sync/notifier/invalidation_notifier.h',
+        'browser/sync/notifier/invalidation_notifier.cc',
         'browser/sync/notifier/invalidation_util.cc',
         'browser/sync/notifier/invalidation_util.h',
+        'browser/sync/notifier/p2p_notifier.h',
+        'browser/sync/notifier/p2p_notifier.cc',
         'browser/sync/notifier/registration_manager.cc',
         'browser/sync/notifier/registration_manager.h',
         'browser/sync/notifier/server_notifier_thread.cc',
@@ -849,8 +851,6 @@
         'browser/sync/notifier/sync_notifier.h',
         'browser/sync/notifier/sync_notifier_factory.h',
         'browser/sync/notifier/sync_notifier_factory.cc',
-        'browser/sync/notifier/sync_notifier_impl.h',
-        'browser/sync/notifier/sync_notifier_impl.cc',
         'browser/sync/notifier/sync_notifier_callback.h',
       ],
       'include_dirs': [