Refactor sync notifier out of sync api.

chrome/browser/sync/engine/syncapi.cc

 // 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/engine/syncapi.h"
 
 #include <algorithm>
 #include <bitset>
 #include <iomanip>
 #include <list>
 #include <queue>
 #include <string>
 #include <vector>
 
 #include "base/base64.h"
+#include "base/command_line.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/observer_list.h"
 #include "base/scoped_ptr.h"
 #include "base/sha1.h"
 #include "base/string_number_conversions.h"
 #include "base/string_util.h"
 #include "base/synchronization/lock.h"
 #include "base/task.h"
 #include "base/utf_string_conversions.h"
 #include "base/values.h"
 #include "chrome/browser/browser_thread.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"
 #include "chrome/browser/sync/engine/net/server_connection_manager.h"
 #include "chrome/browser/sync/engine/net/syncapi_server_connection_manager.h"
 #include "chrome/browser/sync/engine/syncer.h"
 #include "chrome/browser/sync/engine/syncer_thread.h"
 #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_callback.h"
+#include "chrome/browser/sync/notifier/sync_notifier.h"
+#include "chrome/browser/sync/notifier/sync_notifier_factory.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"
 #include "chrome/browser/sync/protocol/extension_specifics.pb.h"
 #include "chrome/browser/sync/protocol/nigori_specifics.pb.h"
 #include "chrome/browser/sync/protocol/preference_specifics.pb.h"
 #include "chrome/browser/sync/protocol/proto_value_conversions.h"
 #include "chrome/browser/sync/protocol/service_constants.h"
 #include "chrome/browser/sync/protocol/session_specifics.pb.h"
 #include "chrome/browser/sync/protocol/sync.pb.h"
 #include "chrome/browser/sync/protocol/theme_specifics.pb.h"
 #include "chrome/browser/sync/protocol/typed_url_specifics.pb.h"
 #include "chrome/browser/sync/sessions/sync_session.h"
 #include "chrome/browser/sync/sessions/sync_session_context.h"
 #include "chrome/browser/sync/syncable/autofill_migration.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
 #include "chrome/browser/sync/syncable/nigori_util.h"
 #include "chrome/browser/sync/syncable/syncable.h"
 #include "chrome/browser/sync/util/crypto_helpers.h"
+#include "chrome/common/chrome_switches.h"
 #include "chrome/common/deprecated/event_sys.h"
 #include "chrome/common/net/gaia/gaia_authenticator.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;
 using browser_sync::Cryptographer;
 using browser_sync::KeyParams;
 using browser_sync::ModelSafeRoutingInfo;
 using browser_sync::ModelSafeWorker;
 using browser_sync::ModelSafeWorkerRegistrar;
 using browser_sync::ServerConnectionEvent;
 using browser_sync::SyncEngineEvent;
 using browser_sync::SyncEngineEventListener;
 using browser_sync::Syncer;
 using browser_sync::SyncerThread;
 using browser_sync::kNigoriTag;
 using browser_sync::sessions::SyncSessionContext;
-using notifier::TalkMediator;
-using notifier::TalkMediatorImpl;
 using std::list;
 using std::hex;
 using std::string;
 using std::vector;
 using syncable::Directory;
 using syncable::DirectoryManager;
 using syncable::Entry;
 using syncable::SPECIFICS;
+using sync_notifier::SyncNotifierFactory;
 using sync_pb::AutofillProfileSpecifics;
 
 typedef GoogleServiceAuthError AuthError;
 
 static const int kThreadExitTimeoutMsec = 60000;
 static const int kSSLPort = 443;
 static const int kSyncerThreadDelayMsec = 250;
 
 #if defined(OS_CHROMEOS)
 static const int kChromeOSNetworkChangeReactionDelayHackMsec = 5000;
@@ skipping 1000 matching lines @@
 
 const sync_pb::PasswordSpecificsData&
     SyncManager::ExtraPasswordChangeRecordData::unencrypted() const {
   return unencrypted_;
 }
 
 //////////////////////////////////////////////////////////////////////////
 // SyncManager's implementation: SyncManager::SyncInternal
 class SyncManager::SyncInternal
     : public net::NetworkChangeNotifier::Observer,
-      public TalkMediator::Delegate,
-      public sync_notifier::StateWriter,
+      public sync_notifier::SyncNotifierCallback,
       public browser_sync::ChannelEventHandler<syncable::DirectoryChangeEvent>,
       public browser_sync::JsBackend,
       public SyncEngineEventListener {
   static const int kDefaultNudgeDelayMilliseconds;
   static const int kPreferencesNudgeDelayMilliseconds;
  public:
   explicit SyncInternal(SyncManager* sync_manager)
       : core_message_loop_(NULL),
         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));
   }
 
   virtual ~SyncInternal() {
     DCHECK(!core_message_loop_);
     DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
   }
 
   bool Init(const FilePath& database_location,
             const std::string& sync_server_and_path,
             int port,
             bool use_ssl,
             HttpPostProviderFactory* post_factory,
             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);
 
   // Sign into sync with given credentials.
   // We do not verify the tokens given. After this call, the tokens are set
   // and the sync DB is open. True if successful, false if something
   // went wrong.
   bool SignIn(const SyncCredentials& credentials);
 
   // Update tokens that we're using in Sync. Email must stay the same.
@@ skipping 33 matching lines @@
       const syncable::DirectoryChangeEvent& event);
   void HandleCalculateChangesChangeEventFromSyncer(
       const syncable::DirectoryChangeEvent& event);
 
   // Listens for notifications from the ServerConnectionManager
   void HandleServerConnectionEvent(const ServerConnectionEvent& event);
 
   // Open the directory named with username_for_share
   bool OpenDirectory();
 
-  // Login to the talk mediator with the given credentials.
-  void TalkMediatorLogin(
+  // Login to the sync notifier with the given credentials.
+  void SyncNotifierLogin(
       const std::string& email, const std::string& token);
 
-  // TalkMediator::Delegate implementation.
+  // SyncNotifierCallback 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);
+  virtual void StoreCookie(const std::string& cookie);
 
   void AddObserver(SyncManager::Observer* observer);
 
   void RemoveObserver(SyncManager::Observer* observer);
 
   // Accessors for the private members.
   DirectoryManager* dir_manager() { return share_.dir_manager.get(); }
   SyncAPIServerConnectionManager* connection_manager() {
     return connection_manager_.get();
   }
   SyncerThread* 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
   // types.
   const std::string& username_for_share() const {
     return share_.name;
   }
 
   Status GetStatus();
 
@@ skipping 87 matching lines @@
 
   // browser_sync::JsBackend implementation.
   virtual void SetParentJsEventRouter(browser_sync::JsEventRouter* router);
   virtual void RemoveParentJsEventRouter();
   virtual const browser_sync::JsEventRouter* GetParentJsEventRouter() const;
   virtual void ProcessMessage(const std::string& name,
                               const browser_sync::JsArgList& args,
                               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();
 
   // Helper to set initialized_ to true and raise an event to clients to notify
   // that initialization is complete and it is safe to send us changes. If
   // already initialized, this is a no-op.
   void MarkAndNotifyInitializationComplete();
 
   // If there's a pending notification to be sent, either from the
@@ skipping 109 matching lines @@
 
   browser_sync::JsEventRouter* parent_router_;
 
   // The ServerConnectionManager used to abstract communication between the
   // client (the Syncer) and the sync server.
   scoped_ptr<SyncAPIServerConnectionManager> connection_manager_;
 
   // The thread that runs the Syncer. Needs to be explicitly Start()ed.
   scoped_refptr<SyncerThread> syncer_thread_;
 
-  // Notification (xmpp) handler.
-  scoped_ptr<TalkMediator> talk_mediator_;
+  // SyncNotifier to setup notifications.
+  scoped_ptr<sync_notifier::SyncNotifier> sync_notifier_;
 
   // A multi-purpose status watch object that aggregates stats from various
   // sync components.
   AllStatus allstatus_;
 
   // Each element of this array is a store of change records produced by
   // HandleChangeEvent during the CALCULATE_CHANGES step.  The changes are
   // segregated by model type, and are stored here to be processed and
   // forwarded to the observer slightly later, at the TRANSACTION_ENDING
   // step by HandleTransactionEndingChangeEvent. The list is cleared in the
@@ skipping 26 matching lines @@
 
   // 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
   // (as in the steady-state), and the syncable::Directory has been opened,
   // meaning we are ready to accept changes.  Protected by initialized_mutex_
   // as it can get read/set by both the SyncerThread and the AuthWatcherThread.
   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_;
+  // We still use p2p notifications for tests and it set from the command line.
+  bool using_p2p_notifications_;
 
   ScopedRunnableMethodFactory<SyncManager::SyncInternal> method_factory_;
-
-  sync_notifier::ServerNotifierThread* server_notifier_thread_;
 };
 const int SyncManager::SyncInternal::kDefaultNudgeDelayMilliseconds = 200;
 const int SyncManager::SyncInternal::kPreferencesNudgeDelayMilliseconds = 2000;
 
 SyncManager::Observer::~Observer() {}
 
 SyncManager::SyncManager() {
   data_ = new SyncInternal(this);
 }
 
 bool SyncManager::Init(const FilePath& database_location,
                        const char* sync_server_and_path,
                        int sync_server_port,
                        bool use_ssl,
                        HttpPostProviderFactory* post_factory,
                        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);
   VLOG(1) << "SyncManager starting Init...";
   string server_string(sync_server_and_path);
   return data_->Init(database_location,
                      server_string,
                      sync_server_port,
                      use_ssl,
                      post_factory,
                      registrar,
                      user_agent,
                      credentials,
-                     notifier_options,
                      restored_key_for_bootstrapping,
                      setup_for_test_mode);
 }
 
 void SyncManager::UpdateCredentials(const SyncCredentials& credentials) {
   data_->UpdateCredentials(credentials);
 }
 
 void SyncManager::UpdateEnabledTypes(const syncable::ModelTypeSet& types) {
   data_->UpdateEnabledTypes(types);
@@ skipping 72 matching lines @@
 
 bool SyncManager::SyncInternal::Init(
     const FilePath& database_location,
     const std::string& sync_server_and_path,
     int port,
     bool use_ssl,
     HttpPostProviderFactory* post_factory,
     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) {
 
   VLOG(1) << "Starting SyncInternal initialization.";
 
   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;
 
+  const CommandLine& command_line = *CommandLine::ForCurrentProcess();
+  if (command_line.HasSwitch(switches::kSyncNotificationMethod)) {
+    const std::string notification_method_str(
+        command_line.GetSwitchValueASCII(switches::kSyncNotificationMethod));
+    // Directly using string "p2p" as we do not want to depend on jingle.
+    if (notification_method_str == "p2p") {
+       using_p2p_notifications_ = true;
+    }
+  }
+
   share_.dir_manager.reset(new DirectoryManager(database_location));
 
   connection_manager_.reset(new SyncAPIServerConnectionManager(
       sync_server_and_path, port, use_ssl, user_agent, post_factory));
 
   connection_manager_hookup_.reset(
       NewEventListenerHookup(connection_manager()->channel(), this,
           &SyncManager::SyncInternal::HandleServerConnectionEvent));
 
   net::NetworkChangeNotifier::AddObserver(this);
@@ skipping 91 matching lines @@
   }
 
   // Notify that initialization is complete.
   FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                     OnInitializationComplete());
 }
 
 void SyncManager::SyncInternal::SendPendingXMPPNotification(
     bool new_pending_notification) {
   DCHECK_EQ(MessageLoop::current(), core_message_loop_);
-  DCHECK_NE(notifier_options_.notification_method,
-            notifier::NOTIFICATION_SERVER);
+  DCHECK(using_p2p_notifications_);
   notification_pending_ = notification_pending_ || new_pending_notification;
   if (!notification_pending_) {
     VLOG(1) << "Not sending notification: no pending notification";
     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";
-  }
+#if defined(UNIT_TEST)
+  ((SyncNotifierImpl*) sync_notifier_)->SendNotification();
+#endif
 }
 
 bool SyncManager::SyncInternal::OpenDirectory() {
   DCHECK(!initialized()) << "Should only happen once";
 
   bool share_opened = dir_manager()->Open(username_for_share());
   DCHECK(share_opened);
   if (!share_opened) {
     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                       OnStopSyncingPermanently());
@@ skipping 32 matching lines @@
     UpdateCredentials(credentials);
   }
   return true;
 }
 
 void SyncManager::SyncInternal::UpdateCredentials(
     const SyncCredentials& credentials) {
   DCHECK_EQ(MessageLoop::current(), core_message_loop_);
   DCHECK_EQ(credentials.email, share_.name);
   connection_manager()->set_auth_token(credentials.sync_token);
-  TalkMediatorLogin(credentials.email, credentials.sync_token);
+  SyncNotifierLogin(credentials.email, credentials.sync_token);
   CheckServerReachable();
   sync_manager_->RequestNudge();
 }
 
 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() {
   FOR_EACH_OBSERVER(
       SyncManager::Observer, observers_,
       OnAuthError(AuthError(AuthError::INVALID_GAIA_CREDENTIALS)));
 }
 
 void SyncManager::SyncInternal::SetUsingExplicitPassphrasePrefForMigration() {
   WriteTransaction trans(&share_);
@@ skipping 197 matching lines @@
   return data_;
 }
 
 void SyncManager::Shutdown() {
   data_->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...";
     }
     syncer_thread_ = NULL;
   }
 
-  // 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.";
+  // Notify the notifier of shutdown.
+  if (sync_notifier_.get()) {
+    sync_notifier_->Logout();
+    sync_notifier_.reset();
   }
 
   // Pump any messages the auth watcher, syncer thread, or talk
   // mediator posted before they shut down. (See OnSyncEngineEvent(),
   // and HandleTalkMediatorEvent() for the
   // events that may be posted.)
   {
     CHECK(core_message_loop_);
     bool old_state = core_message_loop_->NestableTasksAllowed();
     core_message_loop_->SetNestableTasksAllowed(true);
@@ skipping 307 matching lines @@
     }
 
     if (!initialized())
       return;
 
     if (!event.snapshot->has_more_to_sync) {
       FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                         OnSyncCycleCompleted(event.snapshot));
     }
 
-    if (notifier_options_.notification_method !=
-        notifier::NOTIFICATION_SERVER) {
+#if defined(UNIT_TEST)
+    if (using_p2p_notifications_) {
       // 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);
       core_message_loop_->PostTask(
           FROM_HERE,
           NewRunnableMethod(
               this,
               &SyncManager::SyncInternal::SendPendingXMPPNotification,
               new_pending_notification));
     }
+#endif
   }
 
   if (event.what_happened == SyncEngineEvent::SYNCER_THREAD_PAUSED) {
     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                       OnPaused());
     return;
   }
 
   if (event.what_happened == SyncEngineEvent::SYNCER_THREAD_RESUMED) {
     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
@@ skipping 126 matching lines @@
   if (syncer_thread()) {
     syncer_thread()->SetNotificationsEnabled(notifications_enabled);
   }
   if (parent_router_) {
     ListValue args;
     args.Append(Value::CreateBooleanValue(notifications_enabled));
     // TODO(akalin): Tidy up grammar in event names.
     parent_router_->RouteJsEvent("onSyncNotificationStateChange",
                                  browser_sync::JsArgList(args), NULL);
   }
-  if ((notifier_options_.notification_method !=
-       notifier::NOTIFICATION_SERVER) && notifications_enabled) {
+
+#if defined(UNIT_TEST)
+  // P2P notifications used for integration tests.
+  if (using_p2p_notifications_ && 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));
   }
+#endif
 }
 
-void SyncManager::SyncInternal::TalkMediatorLogin(
+void SyncManager::SyncInternal::SyncNotifierLogin(
     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();
+  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;
+  }
+
+  scoped_ptr<SyncNotifierFactory> sync_notifier_factory(

[akalin, 2011/03/08 02:48:30]

no need for this to be a scoped_ptr, just create it on the stack

[Agrawal, 2011/03/08 23:07:28]

Done.

+      new SyncNotifierFactory());
+  sync_notifier_.reset(sync_notifier_factory->CreateSyncNotifier());
+  sync_notifier_->Login(email, token, state, this);
 }
 
 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) {
+  if (!using_p2p_notifications_) {
     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.";
@@ skipping 44 matching lines @@
       const std::string& model_type_str =
           syncable::ModelTypeToString(it->first);
       changed_types->Append(Value::CreateStringValue(model_type_str));
     }
     parent_router_->RouteJsEvent("onSyncIncomingNotification",
                                  browser_sync::JsArgList(args), NULL);
   }
 }
 
 void SyncManager::SyncInternal::OnOutgoingNotification() {
-  DCHECK_NE(notifier_options_.notification_method,
-            notifier::NOTIFICATION_SERVER);
+  DCHECK(using_p2p_notifications_);
   allstatus_.IncrementNotificationsSent();
 }
 
-void SyncManager::SyncInternal::WriteState(const std::string& state) {
+void SyncManager::SyncInternal::StoreCookie(
+    const std::string& state) {
   syncable::ScopedDirLookup lookup(dir_manager(), username_for_share());
   if (!lookup.good()) {
     LOG(ERROR) << "Could not write notification state";
     // TODO(akalin): Propagate result callback all the way to this
     // function and call it with "false" to signal failure.
     return;
   }
   if (VLOG_IS_ON(1)) {
     std::string encoded_state;
     base::Base64Encode(state, &encoded_state);
@@ skipping 68 matching lines @@
 
 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);
 }
 
 }  // namespace sync_api