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 894bd1e48385dfec3f75b2e452773ab5870b306e..bdfaa995589a1939a2de5a2eee6ceedfff327685 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_observer.h"

[akalin, 2011/03/10 04:26:45]

fix include order to be alphabetical

[Agrawal, 2011/03/11 01:13:07]

Done.

+#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"
@@ -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 {
@@ -1128,8 +1123,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));
   }
 
@@ -1146,7 +1140,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 +1193,18 @@ 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.
+  // SyncNotifierObserver implementation.
   virtual void OnNotificationStateChange(
       bool notifications_enabled);
 
   virtual void OnIncomingNotification(
-      const IncomingNotificationData& notification_data);
+      const browser_sync::sessions::TypePayloadMap& type_payloads);
 
-  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);
 
@@ -1226,7 +1216,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 +1323,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();
@@ -1467,8 +1452,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_;
+  // SyncNotifier to setup notifications.
+  scoped_ptr<sync_notifier::SyncNotifier> sync_notifier_;
 
   // A multi-purpose status watch object that aggregates stats from various
   // sync components.
@@ -1515,17 +1500,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;
@@ -1544,7 +1526,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 +1539,6 @@ bool SyncManager::Init(const FilePath& database_location,
                      registrar,
                      user_agent,
                      credentials,
-                     notifier_options,
                      restored_key_for_bootstrapping,
                      setup_for_test_mode);
 }
@@ -1660,7 +1640,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,10 +1647,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();

[akalin, 2011/03/10 04:26:45]

this logic should be either in the factory or the notifier implementation;
ideally, the syncapi shouldn't have to worry about the notification method.

[Agrawal, 2011/03/11 01:13:07]

I was using this to prevent calling SendNotification for the normal code path. I
am now calling SendNotification but doing noting in the implementation if we are
using p2p notifications (the implementation knows about the notification
method).

+  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(
@@ -1785,35 +1773,14 @@ 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_);

[akalin, 2011/03/10 04:26:45]

this can be omitted

[Agrawal, 2011/03/11 01:13:07]

Done.

   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";
-  }
+  allstatus_.IncrementNotificationsSent();
+  sync_notifier_->SendNotification(&notification_pending_);

[akalin, 2011/03/10 04:26:45]

since this is only used in tests, just assume the notification was successful
and set notification_pending_ to false.

[Agrawal, 2011/03/11 01:13:07]

Done.

 }
 
 bool SyncManager::SyncInternal::OpenDirectory() {
@@ -1866,7 +1833,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();
 }
 
@@ -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,10 @@ 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.";
+  // Notify the notifier of shutdown.
+  if (sync_notifier_.get()) {
+    sync_notifier_->RemoveObserver(this);
+    sync_notifier_.reset();
   }
 
   // Pump any messages the auth watcher, syncer thread, or talk
@@ -2487,8 +2397,7 @@ void SyncManager::SyncInternal::OnSyncEngineEvent(
                         OnSyncCycleCompleted(event.snapshot));
     }
 
-    if (notifier_options_.notification_method !=
-        notifier::NOTIFICATION_SERVER) {
+    if (using_p2p_notifications_) {

[akalin, 2011/03/10 04:26:45]

You can remove this check; just add a comment that this only takes effect during
sync integration tests.

[Agrawal, 2011/03/11 01:13:07]

Done.

       // 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 =
@@ -2646,8 +2555,9 @@ void SyncManager::SyncInternal::OnNotificationStateChange(
     parent_router_->RouteJsEvent("onSyncNotificationStateChange",
                                  browser_sync::JsArgList(args), NULL);
   }
-  if ((notifier_options_.notification_method !=
-       notifier::NOTIFICATION_SERVER) && notifications_enabled) {
+
+  // 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
@@ -2667,66 +2577,38 @@ void SyncManager::SyncInternal::OnNotificationStateChange(
   }
 }
 
-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;
+  }
+
+  SyncNotifierFactory sync_notifier_factory;

[akalin, 2011/03/10 04:26:45]

It's been a defect that the talk mediator gets reinitialized on new credentials.
 I think we can fix this now, and provide a cleaner interface.  Basically,
replace the Login() method with an UpdateCredentials() method that takes
email/token and can be called multiple times, and a SetState() function that
should only be called once (before any call to UpdateCredentials()). (For now
it's okay to reinitialize TalkMediator in the implementation of
UpdateCredentials(), but we can change that later).  Then we can simply call
sync_notifier_->UpdateCredentials() here.

So basically sync_notifier_ will be constructed from the factory at Init()
instead of here, and the state will be set there.

[Agrawal, 2011/03/11 01:13:07]

Done. As fas as I understand, we need to call setstate after OpenDirectory is
called in SignIn. I have moved sync notifier construction there too.

+  sync_notifier_.reset(sync_notifier_factory.CreateSyncNotifier(
+      *CommandLine::ForCurrentProcess()));
+  sync_notifier_->AddObserver(this);
+  sync_notifier_->Login(email, token, state);
 }
 
 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 +2620,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 +2631,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::StoreCookie(
+    const std::string& state) {
   syncable::ScopedDirLookup lookup(dir_manager(), username_for_share());
   if (!lookup.good()) {
     LOG(ERROR) << "Could not write notification state";
@@ -2837,10 +2714,14 @@ void SyncManager::TriggerOnNotificationStateChangeForTest(
 
 void SyncManager::TriggerOnIncomingNotificationForTest(
     const syncable::ModelTypeBitSet& model_types) {
-  IncomingNotificationData notification_data;
-  notification_data.service_url = model_types.to_string();
+  browser_sync::sessions::TypePayloadMap model_types_with_payloads;
+
+  model_types_with_payloads =
+      browser_sync::sessions::MakeTypePayloadMapFromBitSet(model_types,
+          std::string());
+
   // Here we rely on the default notification method being SERVER.
-  data_->OnIncomingNotification(notification_data);
+  data_->OnIncomingNotification(model_types_with_payloads);
 }
 
 }  // namespace sync_api