Refactor sync notifier out of sync api.

chrome/browser/sync/engine/syncapi.cc

Index: chrome/browser/sync/engine/syncapi.cc
diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index 3c31f7703591371156ab00378812a3d02703bfcd..524f64d0937be12810eb4b19a97f5fa1e66c4df9 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -13,6 +13,7 @@
 #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"
@@ -36,8 +37,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_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"
@@ -57,12 +59,9 @@
 #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;
@@ -78,8 +77,6 @@ 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;
@@ -88,6 +85,7 @@ using syncable::Directory;
 using syncable::DirectoryManager;
 using syncable::Entry;
 using syncable::SPECIFICS;
+using sync_notifier::SyncNotifierFactory;
 using sync_pb::AutofillProfileSpecifics;
 
 typedef GoogleServiceAuthError AuthError;
@@ -1108,8 +1106,7 @@ const sync_pb::PasswordSpecificsData&
 // 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 {
@@ -1123,8 +1120,7 @@ class SyncManager::SyncInternal
         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));
   }
 
@@ -1141,7 +1137,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);
 
@@ -1195,11 +1190,11 @@ class SyncManager::SyncInternal
   // 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);
 
@@ -1207,9 +1202,7 @@ class SyncManager::SyncInternal
       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);
 
@@ -1221,7 +1214,6 @@ class SyncManager::SyncInternal
     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
@@ -1329,10 +1321,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();
@@ -1462,8 +1450,8 @@ class SyncManager::SyncInternal
   // 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.
@@ -1510,17 +1498,14 @@ 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_;
+  // 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;
@@ -1539,7 +1524,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);
@@ -1553,7 +1537,6 @@ bool SyncManager::Init(const FilePath& database_location,
                      registrar,
                      user_agent,
                      credentials,
-                     notifier_options,
                      restored_key_for_bootstrapping,
                      setup_for_test_mode);
 }
@@ -1646,7 +1629,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) {
 
@@ -1654,10 +1636,19 @@ 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;
 
+  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(
@@ -1769,35 +1760,15 @@ void SyncManager::SyncInternal::MarkAndNotifyInitializationComplete() {
 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() {
@@ -1850,7 +1821,7 @@ void SyncManager::SyncInternal::UpdateCredentials(
   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();
 }
@@ -1859,51 +1830,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() {
@@ -2121,12 +2048,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...";
@@ -2134,17 +2055,10 @@ void SyncManager::SyncInternal::Shutdown() {
     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
@@ -2472,8 +2386,8 @@ void SyncManager::SyncInternal::OnSyncEngineEvent(
                         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 =
@@ -2485,6 +2399,7 @@ void SyncManager::SyncInternal::OnSyncEngineEvent(
               &SyncManager::SyncInternal::SendPendingXMPPNotification,
               new_pending_notification));
     }
+#endif
   }
 
   if (event.what_happened == SyncEngineEvent::SYNCER_THREAD_PAUSED) {
@@ -2631,8 +2546,10 @@ void SyncManager::SyncInternal::OnNotificationStateChange(
     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
@@ -2650,16 +2567,30 @@ void SyncManager::SyncInternal::OnNotificationStateChange(
             &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(
@@ -2669,8 +2600,7 @@ void SyncManager::SyncInternal::OnIncomingNotification(
   // 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;
@@ -2735,12 +2665,12 @@ void SyncManager::SyncInternal::OnIncomingNotification(
 }
 
 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";