Refactor sync notifier out of sync api

We now have a clearly defined interface for notifier.

Old p2p notifications are still used by tests.

Original patch by nileshagrawal@google.com.

BUG=75831
TEST=

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

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

15 files changed, 549 insertions, 331 deletions

diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index 894bd1e..2668ec3 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -38,8 +38,9 @@
 #include "chrome/browser/sync/js_arg_list.h"
 #include "chrome/browser/sync/js_backend.h"
 #include "chrome/browser/sync/js_event_router.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/notifier/sync_notifier_factory.h"
+#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
 #include "chrome/browser/sync/protocol/app_specifics.pb.h"
 #include "chrome/browser/sync/protocol/autofill_specifics.pb.h"
 #include "chrome/browser/sync/protocol/bookmark_specifics.pb.h"
@@ -63,10 +64,6 @@
 #include "chrome/common/deprecated/event_sys.h"
 #include "chrome/common/net/gaia/gaia_authenticator.h"
 #include "content/browser/browser_thread.h"
-#include "jingle/notifier/listener/mediator_thread_impl.h"
-#include "jingle/notifier/listener/notification_constants.h"
-#include "jingle/notifier/listener/talk_mediator.h"
-#include "jingle/notifier/listener/talk_mediator_impl.h"
 #include "net/base/network_change_notifier.h"
 
 using browser_sync::AllStatus;
@@ -83,8 +80,6 @@ using browser_sync::SyncerThread;
 using browser_sync::SyncerThreadAdapter;
 using browser_sync::kNigoriTag;
 using browser_sync::sessions::SyncSessionContext;
-using notifier::TalkMediator;
-using notifier::TalkMediatorImpl;
 using std::list;
 using std::hex;
 using std::string;
@@ -93,6 +88,7 @@ using syncable::Directory;
 using syncable::DirectoryManager;
 using syncable::Entry;
 using syncable::SPECIFICS;
+using sync_notifier::SyncNotifierFactory;
 using sync_pb::AutofillProfileSpecifics;
 
 typedef GoogleServiceAuthError AuthError;
@@ -1113,8 +1109,7 @@ const sync_pb::PasswordSpecificsData&
 // SyncManager's implementation: SyncManager::SyncInternal
 class SyncManager::SyncInternal
     : public net::NetworkChangeNotifier::IPAddressObserver,
-      public TalkMediator::Delegate,
-      public sync_notifier::StateWriter,
+      public sync_notifier::SyncNotifierObserver,
       public browser_sync::ChannelEventHandler<syncable::DirectoryChangeEvent>,
       public browser_sync::JsBackend,
       public SyncEngineEventListener {
@@ -1126,10 +1121,8 @@ class SyncManager::SyncInternal
         parent_router_(NULL),
         sync_manager_(sync_manager),
         registrar_(NULL),
-        notification_pending_(false),
         initialized_(false),
-        ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)),
-        server_notifier_thread_(NULL) {
+        ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)) {
     DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
   }
 
@@ -1146,7 +1139,6 @@ class SyncManager::SyncInternal
             ModelSafeWorkerRegistrar* model_safe_worker_registrar,
             const char* user_agent,
             const SyncCredentials& credentials,
-            const notifier::NotifierOptions& notifier_options,
             const std::string& restored_key_for_bootstrapping,
             bool setup_for_test_mode);
 
@@ -1200,21 +1192,14 @@ class SyncManager::SyncInternal
   // Open the directory named with username_for_share
   bool OpenDirectory();
 
-  // Login to the talk mediator with the given credentials.
-  void TalkMediatorLogin(
-      const std::string& email, const std::string& token);
-
-  // TalkMediator::Delegate implementation.
+  // SyncNotifierObserver implementation.
   virtual void OnNotificationStateChange(
       bool notifications_enabled);
 
   virtual void OnIncomingNotification(
-      const IncomingNotificationData& notification_data);
-
-  virtual void OnOutgoingNotification();
+      const browser_sync::sessions::TypePayloadMap& type_payloads);
 
-  // sync_notifier::StateWriter implementation.
-  virtual void WriteState(const std::string& state);
+  virtual void StoreState(const std::string& cookie);
 
   void AddObserver(SyncManager::Observer* observer);
 
@@ -1226,7 +1211,6 @@ class SyncManager::SyncInternal
     return connection_manager_.get();
   }
   SyncerThreadAdapter* syncer_thread() { return syncer_thread_.get(); }
-  TalkMediator* talk_mediator() { return talk_mediator_.get(); }
   UserShare* GetUserShare() { return &share_; }
 
   // Return the currently active (validated) username for use with syncable
@@ -1334,10 +1318,6 @@ class SyncManager::SyncInternal
                               const browser_sync::JsEventHandler* sender);
 
  private:
-  // Helper to handle the details of initializing the TalkMediator.
-  // Must be called only after OpenDirectory() is called.
-  void InitializeTalkMediator();
-
   // Helper to call OnAuthError when no authentication credentials are
   // available.
   void RaiseAuthNeededEvent();
@@ -1347,10 +1327,8 @@ class SyncManager::SyncInternal
   // already initialized, this is a no-op.
   void MarkAndNotifyInitializationComplete();
 
-  // If there's a pending notification to be sent, either from the
-  // new_pending_notification flag or a previous unsuccessfully sent
-  // notification, tries to send a notification.
-  void SendPendingXMPPNotification(bool new_pending_notification);
+  // Sends notifications to peers.
+  void SendNotification();
 
   // Determine if the parents or predecessors differ between the old and new
   // versions of an entry stored in |a| and |b|.  Note that a node's index may
@@ -1467,8 +1445,8 @@ class SyncManager::SyncInternal
   // The thread that runs the Syncer. Needs to be explicitly Start()ed.
   scoped_ptr<SyncerThreadAdapter> syncer_thread_;
 
-  // Notification (xmpp) handler.
-  scoped_ptr<TalkMediator> talk_mediator_;
+  // The SyncNotifier which notifies us when updates need to be downloaded.
+  scoped_ptr<sync_notifier::SyncNotifier> sync_notifier_;
 
   // A multi-purpose status watch object that aggregates stats from various
   // sync components.
@@ -1503,9 +1481,6 @@ class SyncManager::SyncInternal
   // The instance is shared between the SyncManager and the Syncer.
   ModelSafeWorkerRegistrar* registrar_;
 
-  // True if the next SyncCycle should notify peers of an update.
-  bool notification_pending_;
-
   // Set to true once Init has been called, and we know of an authenticated
   // valid) username either from a fresh authentication attempt (as in
   // first-use case) or from a previous attempt stored in our UserSettings
@@ -1515,17 +1490,11 @@ class SyncManager::SyncInternal
   bool initialized_;
   mutable base::Lock initialized_mutex_;
 
-  notifier::NotifierOptions notifier_options_;
-
   // True if the SyncManager should be running in test mode (no syncer thread
   // actually communicating with the server).
   bool setup_for_test_mode_;
 
-  syncable::ModelTypeSet enabled_types_;
-
   ScopedRunnableMethodFactory<SyncManager::SyncInternal> method_factory_;
-
-  sync_notifier::ServerNotifierThread* server_notifier_thread_;
 };
 const int SyncManager::SyncInternal::kDefaultNudgeDelayMilliseconds = 200;
 const int SyncManager::SyncInternal::kPreferencesNudgeDelayMilliseconds = 2000;
@@ -1544,7 +1513,6 @@ bool SyncManager::Init(const FilePath& database_location,
                        ModelSafeWorkerRegistrar* registrar,
                        const char* user_agent,
                        const SyncCredentials& credentials,
-                       const notifier::NotifierOptions& notifier_options,
                        const std::string& restored_key_for_bootstrapping,
                        bool setup_for_test_mode) {
   DCHECK(post_factory);
@@ -1558,7 +1526,6 @@ bool SyncManager::Init(const FilePath& database_location,
                      registrar,
                      user_agent,
                      credentials,
-                     notifier_options,
                      restored_key_for_bootstrapping,
                      setup_for_test_mode);
 }
@@ -1660,7 +1627,6 @@ bool SyncManager::SyncInternal::Init(
     ModelSafeWorkerRegistrar* model_safe_worker_registrar,
     const char* user_agent,
     const SyncCredentials& credentials,
-    const notifier::NotifierOptions& notifier_options,
     const std::string& restored_key_for_bootstrapping,
     bool setup_for_test_mode) {
 
@@ -1668,7 +1634,6 @@ bool SyncManager::SyncInternal::Init(
 
   core_message_loop_ = MessageLoop::current();
   DCHECK(core_message_loop_);
-  notifier_options_ = notifier_options;
   registrar_ = model_safe_worker_registrar;
   setup_for_test_mode_ = setup_for_test_mode;
 
@@ -1782,38 +1747,14 @@ void SyncManager::SyncInternal::MarkAndNotifyInitializationComplete() {
                     OnInitializationComplete());
 }
 
-void SyncManager::SyncInternal::SendPendingXMPPNotification(
-    bool new_pending_notification) {
+void SyncManager::SyncInternal::SendNotification() {
   DCHECK_EQ(MessageLoop::current(), core_message_loop_);
-  DCHECK_NE(notifier_options_.notification_method,
-            notifier::NOTIFICATION_SERVER);
-  notification_pending_ = notification_pending_ || new_pending_notification;
-  if (!notification_pending_) {
-    VLOG(1) << "Not sending notification: no pending notification";
+  if (!sync_notifier_.get()) {
+    VLOG(1) << "Not sending notification: sync_notifier_ is NULL";
     return;
   }
-  if (!talk_mediator()) {
-    VLOG(1) << "Not sending notification: shutting down (talk_mediator_ is "
-               "NULL)";
-    return;
-  }
-  VLOG(1) << "Sending XMPP notification...";
-  OutgoingNotificationData notification_data;
-  notification_data.service_id = browser_sync::kSyncServiceId;
-  notification_data.service_url = browser_sync::kSyncServiceUrl;
-  notification_data.send_content = true;
-  notification_data.priority = browser_sync::kSyncPriority;
-  notification_data.write_to_cache_only = true;
-  notification_data.service_specific_data =
-      browser_sync::kSyncServiceSpecificData;
-  notification_data.require_subscription = true;
-  bool success = talk_mediator()->SendNotification(notification_data);
-  if (success) {
-    notification_pending_ = false;
-    VLOG(1) << "Sent XMPP notification";
-  } else {
-    VLOG(1) << "Could not send XMPP notification";
-  }
+  allstatus_.IncrementNotificationsSent();
+  sync_notifier_->SendNotification();
 }
 
 bool SyncManager::SyncInternal::OpenDirectory() {
@@ -1855,6 +1796,29 @@ bool SyncManager::SyncInternal::SignIn(const SyncCredentials& credentials) {
   if (!OpenDirectory())
     return false;
 
+  // Initialize the sync notifier. This should be done only after OpenDirectory
+  // is called, as we also need to set the state.
+  // TODO(nileshagrawal): Pass SyncNotifierFactory as an argument to Init
+  // to improve testability.
+  SyncNotifierFactory sync_notifier_factory;
+  sync_notifier_.reset(sync_notifier_factory.CreateSyncNotifier(
+      *CommandLine::ForCurrentProcess()));
+  sync_notifier_->AddObserver(this);
+
+  syncable::ScopedDirLookup lookup(dir_manager(), username_for_share());
+  std::string state;
+  if (lookup.good()) {
+    state = lookup->GetAndClearNotificationState();
+  } else {
+    LOG(ERROR) << "Could not read notification state";
+  }
+  if (VLOG_IS_ON(1)) {
+    std::string encoded_state;
+    base::Base64Encode(state, &encoded_state);
+    VLOG(1) << "Read notification state: " << encoded_state;
+  }
+  sync_notifier_->SetState(state);
+
   if (!setup_for_test_mode_) {
     UpdateCredentials(credentials);
   }
@@ -1865,8 +1829,11 @@ void SyncManager::SyncInternal::UpdateCredentials(
     const SyncCredentials& credentials) {
   DCHECK_EQ(MessageLoop::current(), core_message_loop_);
   DCHECK_EQ(credentials.email, share_.name);
+  DCHECK(!credentials.email.empty());
+  DCHECK(!credentials.sync_token.empty());
   connection_manager()->set_auth_token(credentials.sync_token);
-  TalkMediatorLogin(credentials.email, credentials.sync_token);
+  sync_notifier_->UpdateCredentials(
+      credentials.email, credentials.sync_token);
   CheckServerReachable();
 }
 
@@ -1874,51 +1841,7 @@ void SyncManager::SyncInternal::UpdateEnabledTypes(
   const syncable::ModelTypeSet& types) {
   DCHECK_EQ(MessageLoop::current(), core_message_loop_);
 
-  enabled_types_ = types;
-  if (server_notifier_thread_ != NULL) {
-    server_notifier_thread_->UpdateEnabledTypes(types);
-  }
-}
-
-void SyncManager::SyncInternal::InitializeTalkMediator() {
-  if (notifier_options_.notification_method ==
-      notifier::NOTIFICATION_SERVER) {
-    syncable::ScopedDirLookup lookup(dir_manager(), username_for_share());
-    std::string state;
-    if (lookup.good())
-      state = lookup->GetAndClearNotificationState();
-    else
-      LOG(ERROR) << "Could not read notification state";
-    if (VLOG_IS_ON(1)) {
-      std::string encoded_state;
-      base::Base64Encode(state, &encoded_state);
-      VLOG(1) << "Read notification state: " << encoded_state;
-    }
-
-    // |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_, 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));
-    talk_mediator_->AddSubscribedServiceUrl(browser_sync::kSyncServiceUrl);
-    server_notifier_thread_ = NULL;
-  }
-  talk_mediator()->SetDelegate(this);
+  sync_notifier_->UpdateEnabledTypes(types);
 }
 
 void SyncManager::SyncInternal::RaiseAuthNeededEvent() {
@@ -2136,12 +2059,6 @@ void SyncManager::Shutdown() {
 void SyncManager::SyncInternal::Shutdown() {
   method_factory_.RevokeAll();
 
-  // We NULL out talk_mediator_ so that any tasks pumped below do not
-  // trigger further XMPP actions.
-  //
-  // TODO(akalin): NULL the other member variables defensively, too.
-  scoped_ptr<TalkMediator> talk_mediator(talk_mediator_.release());
-
   if (syncer_thread()) {
     if (!syncer_thread()->Stop(kThreadExitTimeoutMsec)) {
       LOG(FATAL) << "Unable to stop the syncer, it won't be happy...";
@@ -2149,17 +2066,12 @@ void SyncManager::SyncInternal::Shutdown() {
     syncer_thread_.reset();
   }
 
-  // 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.";
+  // We NULL out sync_notifer_ so that any pending tasks do not
+  // trigger further notifications.
+  // TODO(akalin): NULL the other member variables defensively, too.
+  if (sync_notifier_.get()) {
+    sync_notifier_->RemoveObserver(this);
+    sync_notifier_.reset();
   }
 
   // Pump any messages the auth watcher, syncer thread, or talk
@@ -2487,18 +2399,19 @@ void SyncManager::SyncInternal::OnSyncEngineEvent(
                         OnSyncCycleCompleted(event.snapshot));
     }
 
-    if (notifier_options_.notification_method !=
-        notifier::NOTIFICATION_SERVER) {
-      // TODO(chron): Consider changing this back to track has_more_to_sync
-      // only notify peers if a successful commit has occurred.
-      bool new_pending_notification =
-          (event.snapshot->syncer_status.num_successful_commits > 0);
+    // This is here for tests, which are still using p2p notifications.
+    // SendNotification does not do anything if we are using server based
+    // notifications.
+    // TODO(chron): Consider changing this back to track has_more_to_sync
+    // only notify peers if a successful commit has occurred.
+    bool new_notification =
+        (event.snapshot->syncer_status.num_successful_commits > 0);
+    if (new_notification) {
       core_message_loop_->PostTask(
           FROM_HERE,
           NewRunnableMethod(
               this,
-              &SyncManager::SyncInternal::SendPendingXMPPNotification,
-              new_pending_notification));
+              &SyncManager::SyncInternal::SendNotification));
     }
   }
 
@@ -2646,87 +2559,16 @@ void SyncManager::SyncInternal::OnNotificationStateChange(
     parent_router_->RouteJsEvent("onSyncNotificationStateChange",
                                  browser_sync::JsArgList(args), NULL);
   }
-  if ((notifier_options_.notification_method !=
-       notifier::NOTIFICATION_SERVER) && notifications_enabled) {
-    // Nudge the syncer thread when notifications are enabled, in case there is
-    // any data that has not yet been synced. If we are listening to
-    // server-issued notifications, we are already guaranteed to receive a
-    // notification on a successful connection.
-    if (syncer_thread()) {
-      syncer_thread()->NudgeSyncer(0, SyncerThread::kLocal);
-    }
-
-    // Send a notification as soon as subscriptions are on
-    // (see http://code.google.com/p/chromium/issues/detail?id=38563 ).
-    core_message_loop_->PostTask(
-        FROM_HERE,
-        NewRunnableMethod(
-            this,
-            &SyncManager::SyncInternal::SendPendingXMPPNotification,
-            true));
-  }
-}
-
-void SyncManager::SyncInternal::TalkMediatorLogin(
-    const std::string& email, const std::string& token) {
-  DCHECK_EQ(MessageLoop::current(), core_message_loop_);
-  DCHECK(!email.empty());
-  DCHECK(!token.empty());
-  InitializeTalkMediator();
-  talk_mediator()->SetAuthToken(email, token, SYNC_SERVICE_NAME);
-  talk_mediator()->Login();
 }
 
 void SyncManager::SyncInternal::OnIncomingNotification(
-    const IncomingNotificationData& notification_data) {
-  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_data.service_url << ": " <<
-        notification_data.service_specific_data;
-    syncable::ModelTypeBitSet model_types;
-    const std::string& model_type_list = notification_data.service_url;
-    const std::string& notification_payload =
-        notification_data.service_specific_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_data.service_url.empty() ||
-             (notification_data.service_url ==
-              browser_sync::kSyncLegacyServiceUrl) ||
-             (notification_data.service_url ==
-              browser_sync::kSyncServiceUrl)) {
-    VLOG(1) << "Sync received P2P notification.";
-
-    // Catch for sync integration tests (uses p2p). Just set all enabled
-    // datatypes.
-    ModelSafeRoutingInfo routes;
-    registrar_->GetModelSafeRoutingInfo(&routes);
-    model_types_with_payloads =
-        browser_sync::sessions::MakeTypePayloadMapFromRoutingInfo(routes,
-            std::string());
-  } else {
-    LOG(WARNING) << "Notification fron unexpected source: "
-                 << notification_data.service_url;
-  }
-
-  if (!model_types_with_payloads.empty()) {
+    const browser_sync::sessions::TypePayloadMap& type_payloads) {
+  if (!type_payloads.empty()) {
     if (syncer_thread()) {
       syncer_thread()->NudgeSyncerWithPayloads(
           kSyncerThreadDelayMsec,
           SyncerThread::kNotification,
-          model_types_with_payloads);
+          type_payloads);
     }
     allstatus_.IncrementNotificationsReceived();
   } else {
@@ -2738,8 +2580,8 @@ void SyncManager::SyncInternal::OnIncomingNotification(
     ListValue* changed_types = new ListValue();
     args.Append(changed_types);
     for (browser_sync::sessions::TypePayloadMap::const_iterator
-             it = model_types_with_payloads.begin();
-         it != model_types_with_payloads.end(); ++it) {
+             it = type_payloads.begin();
+         it != type_payloads.end(); ++it) {
       const std::string& model_type_str =
           syncable::ModelTypeToString(it->first);
       changed_types->Append(Value::CreateStringValue(model_type_str));
@@ -2749,13 +2591,8 @@ void SyncManager::SyncInternal::OnIncomingNotification(
   }
 }
 
-void SyncManager::SyncInternal::OnOutgoingNotification() {
-  DCHECK_NE(notifier_options_.notification_method,
-            notifier::NOTIFICATION_SERVER);
-  allstatus_.IncrementNotificationsSent();
-}
-
-void SyncManager::SyncInternal::WriteState(const std::string& state) {
+void SyncManager::SyncInternal::StoreState(
+    const std::string& state) {
   syncable::ScopedDirLookup lookup(dir_manager(), username_for_share());
   if (!lookup.good()) {
     LOG(ERROR) << "Could not write notification state";
@@ -2837,10 +2674,11 @@ void SyncManager::TriggerOnNotificationStateChangeForTest(
 
 void SyncManager::TriggerOnIncomingNotificationForTest(
     const syncable::ModelTypeBitSet& model_types) {
-  IncomingNotificationData notification_data;
-  notification_data.service_url = model_types.to_string();
-  // Here we rely on the default notification method being SERVER.
-  data_->OnIncomingNotification(notification_data);
+  browser_sync::sessions::TypePayloadMap model_types_with_payloads =
+      browser_sync::sessions::MakeTypePayloadMapFromBitSet(model_types,
+          std::string());
+
+  data_->OnIncomingNotification(model_types_with_payloads);
 }
 
 }  // namespace sync_api
diff --git a/chrome/browser/sync/engine/syncapi.h b/chrome/browser/sync/engine/syncapi.h
index 3629793..830f077 100644
--- a/chrome/browser/sync/engine/syncapi.h
+++ b/chrome/browser/sync/engine/syncapi.h
@@ -65,10 +65,6 @@ struct SyncSessionSnapshot;
 }
 }
 
-namespace notifier {
-struct NotifierOptions;
-}
-
 // Forward declarations of internal class types so that sync API objects
 // may have opaque pointers to these types.
 namespace syncable {
@@ -853,7 +849,6 @@ class SyncManager {
             browser_sync::ModelSafeWorkerRegistrar* registrar,
             const char* user_agent,
             const SyncCredentials& credentials,
-            const notifier::NotifierOptions& notifier_options,
             const std::string& restored_key_for_bootstrapping,
             bool setup_for_test_mode);
 
diff --git a/chrome/browser/sync/engine/syncapi_unittest.cc b/chrome/browser/sync/engine/syncapi_unittest.cc
index 38b72ab..fec9975 100644
--- a/chrome/browser/sync/engine/syncapi_unittest.cc
+++ b/chrome/browser/sync/engine/syncapi_unittest.cc
@@ -631,8 +631,7 @@ class SyncManagerTest : public testing::Test,
     ASSERT_TRUE(temp_dir_.CreateUniqueTempDir());
     sync_manager_.Init(temp_dir_.path(), "bogus", 0, false,
                        new TestHttpPostProviderFactory(), this, "bogus",
-                       SyncCredentials(), notifier::NotifierOptions(),
-                       "", true /* setup_for_test_mode */);
+                       SyncCredentials(), "", true /* setup_for_test_mode */);
     sync_manager_.AddObserver(&observer_);
     ModelSafeRoutingInfo routes;
     GetModelSafeRoutingInfo(&routes);
diff --git a/chrome/browser/sync/engine/syncer_thread.cc b/chrome/browser/sync/engine/syncer_thread.cc
index d9240ed..8ad82e5 100644
--- a/chrome/browser/sync/engine/syncer_thread.cc
+++ b/chrome/browser/sync/engine/syncer_thread.cc
@@ -16,7 +16,6 @@
 #include "chrome/browser/sync/engine/net/server_connection_manager.h"
 #include "chrome/browser/sync/engine/syncer.h"
 #include "chrome/browser/sync/sessions/sync_session.h"
-#include "jingle/notifier/listener/notification_constants.h"
 
 #if defined(OS_MACOSX)
 #include <CoreFoundation/CFNumber.h>
diff --git a/chrome/browser/sync/glue/sync_backend_host.cc b/chrome/browser/sync/glue/sync_backend_host.cc
index 8cdef93..b552e6b 100644
--- a/chrome/browser/sync/glue/sync_backend_host.cc
+++ b/chrome/browser/sync/glue/sync_backend_host.cc
@@ -91,8 +91,7 @@ void SyncBackendHost::Initialize(
     const syncable::ModelTypeSet& types,
     URLRequestContextGetter* baseline_context_getter,
     const SyncCredentials& credentials,
-    bool delete_sync_data_folder,
-    const notifier::NotifierOptions& notifier_options) {
+    bool delete_sync_data_folder) {
   if (!core_thread_.Start())
     return;
 
@@ -145,7 +144,6 @@ void SyncBackendHost::Initialize(
       MakeHttpBridgeFactory(baseline_context_getter),
       credentials,
       delete_sync_data_folder,
-      notifier_options,
       RestoreEncryptionBootstrapToken(),
       false));
 }
@@ -591,14 +589,12 @@ SyncBackendHost::Core::DoInitializeOptions::DoInitializeOptions(
     sync_api::HttpPostProviderFactory* http_bridge_factory,
     const sync_api::SyncCredentials& credentials,
     bool delete_sync_data_folder,
-    const notifier::NotifierOptions& notifier_options,
     std::string restored_key_for_bootstrapping,
     bool setup_for_test_mode)
     : service_url(service_url),
       http_bridge_factory(http_bridge_factory),
       credentials(credentials),
       delete_sync_data_folder(delete_sync_data_folder),
-      notifier_options(notifier_options),
       restored_key_for_bootstrapping(restored_key_for_bootstrapping),
       setup_for_test_mode(setup_for_test_mode) {
 }
@@ -716,7 +712,6 @@ void SyncBackendHost::Core::DoInitialize(const DoInitializeOptions& options) {
       host_,  // ModelSafeWorkerRegistrar.
       MakeUserAgentForSyncapi().c_str(),
       options.credentials,
-      options.notifier_options,
       options.restored_key_for_bootstrapping,
       options.setup_for_test_mode);
   DCHECK(success) << "Syncapi initialization failed!";
diff --git a/chrome/browser/sync/glue/sync_backend_host.h b/chrome/browser/sync/glue/sync_backend_host.h
index a822994..2204733 100644
--- a/chrome/browser/sync/glue/sync_backend_host.h
+++ b/chrome/browser/sync/glue/sync_backend_host.h
@@ -28,15 +28,10 @@
 #include "chrome/common/net/gaia/google_service_auth_error.h"
 #include "chrome/common/net/url_request_context_getter.h"
 #include "googleurl/src/gurl.h"
-#include "jingle/notifier/base/notifier_options.h"
 
 class CancelableTask;
 class Profile;
 
-namespace notifier {
-struct NotifierOptions;
-}
-
 namespace browser_sync {
 
 namespace sessions {
@@ -128,8 +123,7 @@ class SyncBackendHost : public browser_sync::ModelSafeWorkerRegistrar {
                   const syncable::ModelTypeSet& types,
                   URLRequestContextGetter* baseline_context_getter,
                   const sync_api::SyncCredentials& credentials,
-                  bool delete_sync_data_folder,
-                  const notifier::NotifierOptions& notifier_options);
+                  bool delete_sync_data_folder);
 
   // Called from |frontend_loop| to update SyncCredentials.
   void UpdateCredentials(const sync_api::SyncCredentials& credentials);
@@ -308,7 +302,6 @@ class SyncBackendHost : public browser_sync::ModelSafeWorkerRegistrar {
           sync_api::HttpPostProviderFactory* http_bridge_factory,
           const sync_api::SyncCredentials& credentials,
           bool delete_sync_data_folder,
-          const notifier::NotifierOptions& notifier_options,
           std::string restored_key_for_bootstrapping,
           bool setup_for_test_mode);
       ~DoInitializeOptions();
@@ -318,7 +311,6 @@ class SyncBackendHost : public browser_sync::ModelSafeWorkerRegistrar {
       sync_api::SyncCredentials credentials;
       std::string lsid;
       bool delete_sync_data_folder;
-      notifier::NotifierOptions notifier_options;
       std::string restored_key_for_bootstrapping;
       bool setup_for_test_mode;
     };
@@ -409,7 +401,7 @@ class SyncBackendHost : public browser_sync::ModelSafeWorkerRegistrar {
       credentials.sync_token = "token";
       DoInitialize(DoInitializeOptions(GURL(), factory, credentials,
                                        delete_sync_data_folder,
-                                       notifier::NotifierOptions(), "", true));
+                                       "", true));
     }
 #endif
 
diff --git a/chrome/browser/sync/notifier/sync_notifier.h b/chrome/browser/sync/notifier/sync_notifier.h
new file mode 100644
index 0000000..da15259
--- /dev/null
+++ b/chrome/browser/sync/notifier/sync_notifier.h
@@ -0,0 +1,47 @@
+// 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.
+//
+// Interface to the sync notifier, which is an object that receives
+// notifications when updates are available for a set of sync types.
+// All the observers are notified when such an event happens.
+
+#ifndef CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_H_
+
+#include <string>
+
+#include "chrome/browser/sync/syncable/model_type.h"
+
+namespace sync_notifier {
+class SyncNotifierObserver;
+
+class SyncNotifier {
+ public:
+  SyncNotifier() {}
+  virtual ~SyncNotifier() {}
+
+  virtual void AddObserver(SyncNotifierObserver* observer) = 0;
+  virtual void RemoveObserver(SyncNotifierObserver* observer) = 0;
+
+  // SetState must be called once, before any call to UpdateCredentials.
+  virtual void SetState(const std::string& state) = 0;
+
+  // The observers won't be notified of any notifications until
+  // UpdateCredentials is called at least once. It can be called more than
+  // once.
+  virtual void UpdateCredentials(
+      const std::string& email, const std::string& token) = 0;
+
+  virtual void UpdateEnabledTypes(const syncable::ModelTypeSet& types) = 0;
+
+  // This is here only to support the old p2p notification implementation,
+  // which is still used by sync integration tests.
+  // TODO(akalin): Remove this once we move the integration tests off p2p
+  // notifications.
+  virtual void SendNotification() = 0;
+};
+}  // namespace sync_notifier
+
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_H_
+
diff --git a/chrome/browser/sync/notifier/sync_notifier_factory.cc b/chrome/browser/sync/notifier/sync_notifier_factory.cc
new file mode 100644
index 0000000..9f2886a
--- /dev/null
+++ b/chrome/browser/sync/notifier/sync_notifier_factory.cc
@@ -0,0 +1,99 @@
+// 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_factory.h"
+
+#include <string>
+
+#include "base/command_line.h"
+#include "base/string_number_conversions.h"
+#include "base/string_util.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"
+#include "net/base/host_port_pair.h"
+
+namespace sync_notifier {
+namespace {
+
+// TODO(akalin): Figure out whether this should be a method of
+// HostPortPair.
+net::HostPortPair StringToHostPortPair(const std::string& host_port_str,
+                                       uint16 default_port) {
+  std::string::size_type colon_index = host_port_str.find(':');
+  if (colon_index == std::string::npos) {
+    return net::HostPortPair(host_port_str, default_port);
+  }
+
+  std::string host = host_port_str.substr(0, colon_index);
+  std::string port_str = host_port_str.substr(colon_index + 1);
+  int port = default_port;
+  if (!base::StringToInt(port_str, &port) ||
+      (port <= 0) || (port > kuint16max)) {
+    LOG(WARNING) << "Could not parse valid port from " << port_str
+                 << "; using port " << default_port;
+    return net::HostPortPair(host, default_port);
+  }
+
+  return net::HostPortPair(host, port);
+}
+
+SyncNotifier* CreateDefaultSyncNotifier(const CommandLine& command_line) {
+  // Contains options specific to how sync clients send and listen to
+  // jingle notifications.
+  notifier::NotifierOptions notifier_options;
+
+  // Override the notification server host from the command-line, if provided.
+  if (command_line.HasSwitch(switches::kSyncNotificationHost)) {
+    std::string value(command_line.GetSwitchValueASCII(
+        switches::kSyncNotificationHost));
+    if (!value.empty()) {
+      notifier_options.xmpp_host_port =
+          StringToHostPortPair(value, notifier::kDefaultXmppPort);
+    }
+    VLOG(1) << "Using " << notifier_options.xmpp_host_port.ToString()
+            << " for test sync notification server.";
+  }
+
+  notifier_options.try_ssltcp_first =
+      command_line.HasSwitch(switches::kSyncTrySsltcpFirstForXmpp);
+  if (notifier_options.try_ssltcp_first)
+    VLOG(1) << "Trying SSL/TCP port before XMPP port for notifications.";
+
+  notifier_options.invalidate_xmpp_login =
+      command_line.HasSwitch(switches::kSyncInvalidateXmppLogin);
+  if (notifier_options.invalidate_xmpp_login) {
+    VLOG(1) << "Invalidating sync XMPP login.";
+  }
+
+  notifier_options.allow_insecure_connection =
+      command_line.HasSwitch(switches::kSyncAllowInsecureXmppConnection);
+  if (notifier_options.allow_insecure_connection) {
+    VLOG(1) << "Allowing insecure XMPP connections.";
+  }
+
+  if (command_line.HasSwitch(switches::kSyncNotificationMethod)) {
+    const std::string notification_method_str(
+        command_line.GetSwitchValueASCII(switches::kSyncNotificationMethod));
+    notifier_options.notification_method =
+        notifier::StringToNotificationMethod(notification_method_str);
+  }
+
+  return new SyncNotifierImpl(notifier_options);
+}
+}  // namespace
+
+SyncNotifierFactory::SyncNotifierFactory() {
+}
+
+SyncNotifierFactory::~SyncNotifierFactory() {
+}
+
+SyncNotifier* SyncNotifierFactory::CreateSyncNotifier(
+    const CommandLine& command_line) {
+  return CreateDefaultSyncNotifier(command_line);
+}
+}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/sync_notifier_factory.h b/chrome/browser/sync/notifier/sync_notifier_factory.h
new file mode 100644
index 0000000..a9da82e
--- /dev/null
+++ b/chrome/browser/sync/notifier/sync_notifier_factory.h
@@ -0,0 +1,25 @@
+// 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.
+
+#ifndef CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_FACTORY_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_FACTORY_H_
+
+class CommandLine;
+
+namespace sync_notifier {
+class SyncNotifier;
+
+// Class to instantiate various implementations of the SyncNotifier interface.
+class SyncNotifierFactory {
+ public:
+  SyncNotifierFactory();
+  ~SyncNotifierFactory();
+
+  // Creates the appropriate sync notifier. The caller should take ownership
+  // of the object returned and delete it when no longer used.
+  SyncNotifier* CreateSyncNotifier(const CommandLine& command_line);
+};
+}  // namespace sync_notifier
+
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_FACTORY_H_
diff --git a/chrome/browser/sync/notifier/sync_notifier_impl.cc b/chrome/browser/sync/notifier/sync_notifier_impl.cc
new file mode 100644
index 0000000..d8d7ca6
--- /dev/null
+++ b/chrome/browser/sync/notifier/sync_notifier_impl.cc
@@ -0,0 +1,195 @@
+// 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)
+    : notifier_options_(notifier_options),
+      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 IncomingNotificationData& notification_data) {
+  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_data.service_url << ": " <<
+        notification_data.service_specific_data;
+    syncable::ModelTypeBitSet model_types;
+    const std::string& model_type_list = notification_data.service_url;
+    const std::string& notification_payload =
+        notification_data.service_specific_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_data.service_url.empty() ||
+             (notification_data.service_url ==
+              browser_sync::kSyncLegacyServiceUrl) ||
+             (notification_data.service_url ==
+              browser_sync::kSyncServiceUrl)) {
+    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_data.service_url;
+  }
+
+  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_, 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));
+    talk_mediator_->AddSubscribedServiceUrl(browser_sync::kSyncServiceUrl);
+    server_notifier_thread_ = NULL;
+  }
+  talk_mediator_->SetDelegate(this);
+  talk_mediator_->SetAuthToken(email, token, SYNC_SERVICE_NAME);
+  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...";
+  OutgoingNotificationData notification_data;
+  notification_data.service_id = browser_sync::kSyncServiceId;
+  notification_data.service_url = browser_sync::kSyncServiceUrl;
+  notification_data.send_content = true;
+  notification_data.priority = browser_sync::kSyncPriority;
+  notification_data.write_to_cache_only = true;
+  notification_data.service_specific_data =
+      browser_sync::kSyncServiceSpecificData;
+  notification_data.require_subscription = true;
+  bool success = talk_mediator_->SendNotification(notification_data);
+  if (success) {
+    VLOG(1) << "Sent XMPP notification";
+  } else {
+    VLOG(1) << "Could not send XMPP notification";
+  }
+}
+}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/sync_notifier_impl.h b/chrome/browser/sync/notifier/sync_notifier_impl.h
new file mode 100644
index 0000000..ccf8d9c
--- /dev/null
+++ b/chrome/browser/sync/notifier/sync_notifier_impl.h
@@ -0,0 +1,70 @@
+// 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.
+//
+// Talk Mediator based implementation of the sync notifier interface.
+// Initializes the talk mediator and registers itself as the delegate.
+
+#ifndef CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_IMPL_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_IMPL_H_
+
+#include <string>
+
+#include "base/observer_list.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"
+#include "jingle/notifier/listener/talk_mediator.h"
+#include "jingle/notifier/listener/talk_mediator_impl.h"
+
+namespace sync_notifier {
+class ServerNotifierThread;
+
+class SyncNotifierImpl
+    : public SyncNotifier,
+      public sync_notifier::StateWriter,
+      public notifier::TalkMediator::Delegate {
+ public:
+  explicit SyncNotifierImpl(const notifier::NotifierOptions& notifier_options);
+
+  virtual ~SyncNotifierImpl();
+
+  // TalkMediator::Delegate implementation.
+  virtual void OnNotificationStateChange(bool notifications_enabled);
+
+  virtual void OnIncomingNotification(
+    const IncomingNotificationData& notification_data);
+
+  virtual void OnOutgoingNotification() {}
+
+  // sync_notifier::StateWriter implementation.
+  virtual void WriteState(const std::string& state);
+
+  // 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 UpdateEnabledTypes(const syncable::ModelTypeSet& types);
+  virtual void SendNotification();
+ private:
+  // Login to the talk mediator with the given credentials.
+  void TalkMediatorLogin(
+      const std::string& email, const std::string& token);
+
+  // Notification (xmpp) handler.
+  scoped_ptr<notifier::TalkMediator> talk_mediator_;
+  syncable::ModelTypeSet enabled_types_;
+  std::string state_;
+
+  notifier::NotifierOptions notifier_options_;
+  ServerNotifierThread* server_notifier_thread_;
+  ObserverList<SyncNotifierObserver> observer_list_;
+};
+}
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_IMPL_H_
diff --git a/chrome/browser/sync/notifier/sync_notifier_observer.h b/chrome/browser/sync/notifier/sync_notifier_observer.h
new file mode 100644
index 0000000..2978d17
--- /dev/null
+++ b/chrome/browser/sync/notifier/sync_notifier_observer.h
@@ -0,0 +1,29 @@
+// 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.
+
+
+#ifndef CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_OBSERVER_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_OBSERVER_H_
+
+#include <string>
+
+#include "chrome/browser/sync/sessions/session_state.h"
+
+namespace sync_notifier {
+
+class SyncNotifierObserver {
+ public:
+  SyncNotifierObserver() {}
+  virtual ~SyncNotifierObserver() {}
+
+  virtual void OnIncomingNotification(
+      const browser_sync::sessions::TypePayloadMap& type_payloads) = 0;
+  virtual void OnNotificationStateChange(bool notifications_enabled) = 0;
+
+  // TODO(nileshagrawal): Find a way to hide state handling inside the
+  // sync notifier implementation.
+  virtual void StoreState(const std::string& state) = 0;
+};
+}  // namespace sync_notifier
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_OBSERVER_H_
diff --git a/chrome/browser/sync/profile_sync_service.cc b/chrome/browser/sync/profile_sync_service.cc
index 007b3ac..19f6836 100644
--- a/chrome/browser/sync/profile_sync_service.cc
+++ b/chrome/browser/sync/profile_sync_service.cc
@@ -14,8 +14,6 @@
 #include "base/metrics/histogram.h"
 #include "base/stl_util-inl.h"
 #include "base/string16.h"
-#include "base/string_number_conversions.h"
-#include "base/string_util.h"
 #include "base/task.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/utf_string_conversions.h"
@@ -44,7 +42,6 @@
 #include "content/common/notification_type.h"
 #include "grit/chromium_strings.h"
 #include "grit/generated_resources.h"
-#include "jingle/notifier/communicator/const_communicator.h"
 #include "net/base/cookie_monster.h"
 #include "ui/base/l10n/l10n_util.h"
 
@@ -262,32 +259,6 @@ void ProfileSyncService::GetDataTypeControllerStates(
       (*state_map)[iter->first] = iter->second.get()->state();
 }
 
-namespace {
-
-// TODO(akalin): Figure out whether this should be a method of
-// HostPortPair.
-net::HostPortPair StringToHostPortPair(const std::string& host_port_str,
-                                       uint16 default_port) {
-  std::string::size_type colon_index = host_port_str.find(':');
-  if (colon_index == std::string::npos) {
-    return net::HostPortPair(host_port_str, default_port);
-  }
-
-  std::string host = host_port_str.substr(0, colon_index);
-  std::string port_str = host_port_str.substr(colon_index + 1);
-  int port = default_port;
-  if (!base::StringToInt(port_str, &port) ||
-      (port <= 0) || (port > kuint16max)) {
-    LOG(WARNING) << "Could not parse valid port from " << port_str
-                 << "; using port " << default_port;
-    return net::HostPortPair(host, default_port);
-  }
-
-  return net::HostPortPair(host, port);
-}
-
-}  // namespace
-
 void ProfileSyncService::InitSettings() {
   const CommandLine& command_line = *CommandLine::ForCurrentProcess();
 
@@ -306,42 +277,6 @@ void ProfileSyncService::InitSettings() {
       }
     }
   }
-
-  // Override the notification server host from the command-line, if provided.
-  if (command_line.HasSwitch(switches::kSyncNotificationHost)) {
-    std::string value(command_line.GetSwitchValueASCII(
-        switches::kSyncNotificationHost));
-    if (!value.empty()) {
-      notifier_options_.xmpp_host_port =
-          StringToHostPortPair(value, notifier::kDefaultXmppPort);
-    }
-    VLOG(1) << "Using " << notifier_options_.xmpp_host_port.ToString()
-            << " for test sync notification server.";
-  }
-
-  notifier_options_.try_ssltcp_first =
-      command_line.HasSwitch(switches::kSyncTrySsltcpFirstForXmpp);
-  if (notifier_options_.try_ssltcp_first)
-    VLOG(1) << "Trying SSL/TCP port before XMPP port for notifications.";
-
-  notifier_options_.invalidate_xmpp_login =
-      command_line.HasSwitch(switches::kSyncInvalidateXmppLogin);
-  if (notifier_options_.invalidate_xmpp_login) {
-    VLOG(1) << "Invalidating sync XMPP login.";
-  }
-
-  notifier_options_.allow_insecure_connection =
-      command_line.HasSwitch(switches::kSyncAllowInsecureXmppConnection);
-  if (notifier_options_.allow_insecure_connection) {
-    VLOG(1) << "Allowing insecure XMPP connections.";
-  }
-
-  if (command_line.HasSwitch(switches::kSyncNotificationMethod)) {
-    const std::string notification_method_str(
-        command_line.GetSwitchValueASCII(switches::kSyncNotificationMethod));
-    notifier_options_.notification_method =
-        notifier::StringToNotificationMethod(notification_method_str);
-  }
 }
 
 void ProfileSyncService::RegisterPreferences() {
@@ -424,8 +359,7 @@ void ProfileSyncService::InitializeBackend(bool delete_sync_data_folder) {
                        types,
                        profile_->GetRequestContext(),
                        credentials,
-                       delete_sync_data_folder,
-                       notifier_options_);
+                       delete_sync_data_folder);
 }
 
 void ProfileSyncService::CreateBackend() {
diff --git a/chrome/browser/sync/profile_sync_service.h b/chrome/browser/sync/profile_sync_service.h
index 97e3fbb..9f10c49 100644
--- a/chrome/browser/sync/profile_sync_service.h
+++ b/chrome/browser/sync/profile_sync_service.h
@@ -31,7 +31,6 @@
 #include "content/common/notification_observer.h"
 #include "content/common/notification_registrar.h"
 #include "googleurl/src/gurl.h"
-#include "jingle/notifier/base/notifier_options.h"
 
 class NotificationDetails;
 class NotificationSource;
@@ -597,10 +596,6 @@ class ProfileSyncService : public browser_sync::SyncFrontend,
   std::string unrecoverable_error_message_;
   scoped_ptr<tracked_objects::Location> unrecoverable_error_location_;
 
-  // Contains options specific to how sync clients send and listen to
-  // notifications.
-  notifier::NotifierOptions notifier_options_;
-
   // Manages the start and stop of the various data types.
   scoped_ptr<browser_sync::DataTypeManager> data_type_manager_;
 
diff --git a/chrome/chrome.gyp b/chrome/chrome.gyp
index 39a6265..5b7fcfd 100644
--- a/chrome/chrome.gyp
+++ b/chrome/chrome.gyp
@@ -950,6 +950,12 @@
         'browser/sync/notifier/server_notifier_thread.cc',
         'browser/sync/notifier/server_notifier_thread.h',
         'browser/sync/notifier/state_writer.h',
+        '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': [
         '..',