Refactor sync-specific parts out of SyncNotifier/SyncNotifierObserver

sync/internal_api/sync_manager.cc

 // Copyright (c) 2012 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 "sync/internal_api/public/sync_manager.h"
 
 #include <string>
 
 #include "base/base64.h"
 #include "base/bind.h"
@@ skipping 27 matching lines @@
 #include "sync/internal_api/public/util/experiments.h"
 #include "sync/internal_api/public/write_node.h"
 #include "sync/internal_api/public/write_transaction.h"
 #include "sync/internal_api/syncapi_internal.h"
 #include "sync/internal_api/syncapi_server_connection_manager.h"
 #include "sync/js/js_arg_list.h"
 #include "sync/js/js_backend.h"
 #include "sync/js/js_event_details.h"
 #include "sync/js/js_event_handler.h"
 #include "sync/js/js_reply_handler.h"
+#include "sync/notifier/invalidation_util.h"
 #include "sync/notifier/notifications_disabled_reason.h"
 #include "sync/notifier/sync_notifier.h"
 #include "sync/notifier/sync_notifier_observer.h"
 #include "sync/protocol/encryption.pb.h"
 #include "sync/protocol/proto_value_conversions.h"
 #include "sync/protocol/sync.pb.h"
 #include "sync/syncable/directory.h"
 #include "sync/syncable/directory_change_delegate.h"
 #include "sync/syncable/entry.h"
 #include "sync/syncable/in_memory_directory_backing_store.h"
@@ skipping 234 matching lines @@
 
   // Open the directory named with username_for_share
   bool OpenDirectory();
 
   // Cryptographer::Observer implementation.
   virtual void OnEncryptedTypesChanged(
       syncer::ModelTypeSet encrypted_types,
       bool encrypt_everything) OVERRIDE;
 
   // SyncNotifierObserver implementation.
+  virtual ObjectIdSet GetHandledIds() OVERRIDE;
   virtual void OnNotificationsEnabled() OVERRIDE;
   virtual void OnNotificationsDisabled(
       syncer::NotificationsDisabledReason reason) OVERRIDE;
   virtual void OnIncomingNotification(
-      const syncer::ModelTypePayloadMap& type_payloads,
+      const ObjectIdPayloadMap& id_payloads,
       syncer::IncomingNotificationSource source) OVERRIDE;
 
   void AddObserver(SyncManager::Observer* observer);
   void RemoveObserver(SyncManager::Observer* observer);
 
   // Accessors for the private members.
   syncable::Directory* directory() { return share_.directory.get(); }
   SyncAPIServerConnectionManager* connection_manager() {
     return connection_manager_.get();
   }
@@ skipping 224 matching lines @@
   // create SyncSessions.  Must outlive the SyncScheduler.
   scoped_ptr<SyncSessionContext> session_context_;
 
   // The scheduler that runs the Syncer. Needs to be explicitly
   // Start()ed.
   scoped_ptr<SyncScheduler> scheduler_;
 
   // The SyncNotifier which notifies us when updates need to be downloaded.
   scoped_ptr<syncer::SyncNotifier> sync_notifier_;
 
+  // The model types we want to register for notifications on represented as
+  // invalidation::ObjectIds.
+  ObjectIdSet enabled_sync_types_;
+
   // 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
   // TRANSACTION_COMPLETE step by HandleTransactionCompleteChangeEvent.
@@ skipping 387 matching lines @@
   // on the UI thread and if shutdown wins then that tasks would execute on
   // a freed pointer. This is because UI thread is not shut down.
   FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                     OnInitializationComplete(
                         MakeWeakHandle(weak_ptr_factory_.GetWeakPtr()),
                         signed_in));
 
   if (!signed_in && testing_mode_ == NON_TEST)
     return false;
 
-  sync_notifier_->AddObserver(this);
+  sync_notifier_->AddHandler(this);
 
   return signed_in;
 }
 
 void SyncManager::SyncInternal::UpdateCryptographerAndNigori(
     const std::string& chrome_version,
     const base::Closure& done_callback) {
   DCHECK(initialized_);
   syncer::GetSessionName(
       blocking_task_runner_,
@@ skipping 202 matching lines @@
         credentials.email, credentials.sync_token);
     if (initialized_ && scheduler()) {
       scheduler()->OnCredentialsUpdated();
     }
   }
 }
 
 void SyncManager::SyncInternal::UpdateEnabledTypes(
     const ModelTypeSet& enabled_types) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  sync_notifier_->UpdateEnabledTypes(enabled_types);
+  enabled_sync_types_ = ModelTypeSetToObjectIdSet(enabled_types);
+  sync_notifier_->ReloadHandlers();
 }
 
 void SyncManager::SyncInternal::SetEncryptionPassphrase(
     const std::string& passphrase,
     bool is_explicit) {
   // We do not accept empty passphrases.
   if (passphrase.empty()) {
     NOTREACHED() << "Cannot encrypt with an empty passphrase.";
     return;
   }
@@ skipping 465 matching lines @@
 
   scheduler_.reset();
   session_context_.reset();
 
   SetJsEventHandler(WeakHandle<JsEventHandler>());
   RemoveObserver(&js_sync_manager_observer_);
 
   RemoveObserver(&debug_info_event_listener_);
 
   if (sync_notifier_.get()) {
-    sync_notifier_->RemoveObserver(this);
+    sync_notifier_->RemoveHandler(this);
   }
   sync_notifier_.reset();
 
   if (connection_manager_.get()) {
     connection_manager_->RemoveListener(this);
   }
   connection_manager_.reset();
 
   net::NetworkChangeNotifier::RemoveIPAddressObserver(this);
   observing_ip_address_changes_ = false;
@@ skipping 562 matching lines @@
 void SyncManager::SyncInternal::UpdateNotificationInfo(
     const syncer::ModelTypePayloadMap& type_payloads) {
   for (syncer::ModelTypePayloadMap::const_iterator it = type_payloads.begin();
        it != type_payloads.end(); ++it) {
     NotificationInfo* info = &notification_info_map_[it->first];
     info->total_count++;
     info->payload = it->second;
   }
 }
 
+ObjectIdSet SyncManager::SyncInternal::GetHandledIds() {
+  return enabled_sync_types_;
+}
+
 void SyncManager::SyncInternal::OnNotificationsEnabled() {
   DVLOG(1) << "Notifications enabled";
   allstatus_.SetNotificationsEnabled(true);
   if (scheduler()) {
     scheduler()->set_notifications_enabled(true);
   }
   // TODO(akalin): Separate onNotificationStateChange into
   // enabled/disabled events.
   if (js_event_handler_.IsInitialized()) {
     DictionaryValue details;
@@ skipping 19 matching lines @@
     js_event_handler_.Call(FROM_HERE,
                            &JsEventHandler::HandleJsEvent,
                            "onNotificationStateChange",
                            JsEventDetails(&details));
   }
   // TODO(akalin): Treat a CREDENTIALS_REJECTED state as an auth
   // error.
 }
 
 void SyncManager::SyncInternal::OnIncomingNotification(
-    const syncer::ModelTypePayloadMap& type_payloads,
+    const ObjectIdPayloadMap& id_payloads,
     syncer::IncomingNotificationSource source) {
   DCHECK(thread_checker_.CalledOnValidThread());
+  ModelTypePayloadMap type_payloads =
+      ObjectIdPayloadMapToModelTypePayloadMap(id_payloads);
   if (source == syncer::LOCAL_NOTIFICATION) {
     if (scheduler()) {
       scheduler()->ScheduleNudgeWithPayloadsAsync(
           TimeDelta::FromMilliseconds(kSyncRefreshDelayMsec),
           syncer::NUDGE_SOURCE_LOCAL_REFRESH,
           type_payloads, FROM_HERE);
     }
   } else if (!type_payloads.empty()) {
     if (scheduler()) {
       scheduler()->ScheduleNudgeWithPayloadsAsync(
@@ skipping 101 matching lines @@
 
 void SyncManager::SimulateDisableNotificationsForTest(int reason) {
   DCHECK(thread_checker_.CalledOnValidThread());
   data_->OnNotificationsDisabled(
       static_cast<syncer::NotificationsDisabledReason>(reason));
 }
 
 void SyncManager::TriggerOnIncomingNotificationForTest(
     ModelTypeSet model_types) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  syncer::ModelTypePayloadMap model_types_with_payloads =
-      syncer::ModelTypePayloadMapFromEnumSet(model_types,
-          std::string());
-
-  data_->OnIncomingNotification(model_types_with_payloads,
-                                syncer::REMOTE_NOTIFICATION);
+  ModelTypePayloadMap type_payloads =
+      ModelTypePayloadMapFromEnumSet(model_types, std::string());
+  data_->OnIncomingNotification(
+      ModelTypePayloadMapToObjectIdPayloadMap(type_payloads),
+      REMOTE_NOTIFICATION);
 }
 
 const char* ConnectionStatusToString(ConnectionStatus status) {
   switch (status) {
     case CONNECTION_OK:
       return "CONNECTION_OK";
     case CONNECTION_AUTH_ERROR:
       return "CONNECTION_AUTH_ERROR";
     case CONNECTION_SERVER_ERROR:
       return "CONNECTION_SERVER_ERROR";
@@ skipping 40 matching lines @@
     share->directory->GetDownloadProgress(i.Get(), &marker);
 
     if (marker.token().empty())
       result.Put(i.Get());
 
   }
   return result;
 }
 
 }  // namespace syncer