Merge 208347 "Revert 208315 "Make use of InvalidationService""

sync/internal_api/sync_manager_impl.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/sync_manager_impl.h"
 
 #include <string>
 
 #include "base/base64.h"
 #include "base/bind.h"
@@ skipping 331 matching lines @@
     const base::FilePath& database_location,
     const WeakHandle<JsEventHandler>& event_handler,
     const std::string& sync_server_and_path,
     int port,
     bool use_ssl,
     scoped_ptr<HttpPostProviderFactory> post_factory,
     const std::vector<ModelSafeWorker*>& workers,
     ExtensionsActivityMonitor* extensions_activity_monitor,
     SyncManager::ChangeDelegate* change_delegate,
     const SyncCredentials& credentials,
+    scoped_ptr<Invalidator> invalidator,
     const std::string& invalidator_client_id,
     const std::string& restored_key_for_bootstrapping,
     const std::string& restored_keystore_key_for_bootstrapping,
     scoped_ptr<InternalComponentsFactory> internal_components_factory,
     Encryptor* encryptor,
     UnrecoverableErrorHandler* unrecoverable_error_handler,
     ReportUnrecoverableErrorFunction report_unrecoverable_error_function,
     bool use_oauth2_token) {
   CHECK(!initialized_);
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(post_factory.get());
   DCHECK(!credentials.email.empty());
   DCHECK(!credentials.sync_token.empty());
   DVLOG(1) << "SyncManager starting Init...";
 
   weak_handle_this_ = MakeWeakHandle(weak_ptr_factory_.GetWeakPtr());
 
   change_delegate_ = change_delegate;
 
+  invalidator_ = invalidator.Pass();
+  invalidator_->RegisterHandler(this);
+
   AddObserver(&js_sync_manager_observer_);
   SetJsEventHandler(event_handler);
 
   AddObserver(&debug_info_event_listener_);
 
   database_path_ = database_location.Append(
       syncable::Directory::kSyncDatabaseFilename);
   encryptor_ = encryptor;
   unrecoverable_error_handler_ = unrecoverable_error_handler;
   report_unrecoverable_error_function_ = report_unrecoverable_error_function;
@@ skipping 217 matching lines @@
 void SyncManagerImpl::UpdateCredentials(const SyncCredentials& credentials) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(initialized_);
   DCHECK(!credentials.email.empty());
   DCHECK(!credentials.sync_token.empty());
 
   observing_network_connectivity_changes_ = true;
   if (!connection_manager_->SetAuthToken(credentials.sync_token))
     return;  // Auth token is known to be invalid, so exit early.
 
+  invalidator_->UpdateCredentials(credentials.email, credentials.sync_token);
   scheduler_->OnCredentialsUpdated();
 
   // TODO(zea): pass the credential age to the debug info event listener.
 }
 
+void SyncManagerImpl::UpdateEnabledTypes(ModelTypeSet enabled_types) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(initialized_);
+  invalidator_->UpdateRegisteredIds(
+      this,
+      ModelTypeSetToObjectIdSet(enabled_types));
+}
+
+void SyncManagerImpl::RegisterInvalidationHandler(
+    InvalidationHandler* handler) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(initialized_);
+  invalidator_->RegisterHandler(handler);
+}
+
+void SyncManagerImpl::UpdateRegisteredInvalidationIds(
+    InvalidationHandler* handler,
+    const ObjectIdSet& ids) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(initialized_);
+  invalidator_->UpdateRegisteredIds(handler, ids);
+}
+
+void SyncManagerImpl::UnregisterInvalidationHandler(
+    InvalidationHandler* handler) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(initialized_);
+  invalidator_->UnregisterHandler(handler);
+}
+
+void SyncManagerImpl::AcknowledgeInvalidation(
+    const invalidation::ObjectId& id, const syncer::AckHandle& ack_handle) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(initialized_);
+  invalidator_->Acknowledge(id, ack_handle);
+}
+
 void SyncManagerImpl::AddObserver(SyncManager::Observer* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   observers_.AddObserver(observer);
 }
 
 void SyncManagerImpl::RemoveObserver(SyncManager::Observer* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   observers_.RemoveObserver(observer);
 }
 
@@ skipping 18 matching lines @@
   if (sync_encryption_handler_) {
     sync_encryption_handler_->RemoveObserver(&debug_info_event_listener_);
     sync_encryption_handler_->RemoveObserver(this);
   }
 
   SetJsEventHandler(WeakHandle<JsEventHandler>());
   RemoveObserver(&js_sync_manager_observer_);
 
   RemoveObserver(&debug_info_event_listener_);
 
-  // |connection_manager_| may end up being NULL here in tests (in synchronous
-  // initialization mode).
+  // |invalidator_| and |connection_manager_| may end up being NULL here in
+  // tests (in synchronous initialization mode).
   //
   // TODO(akalin): Fix this behavior.
+
+  if (invalidator_)
+    invalidator_->UnregisterHandler(this);
+  invalidator_.reset();
+
   if (connection_manager_)
     connection_manager_->RemoveListener(this);
   connection_manager_.reset();
 
   net::NetworkChangeNotifier::RemoveIPAddressObserver(this);
   net::NetworkChangeNotifier::RemoveConnectionTypeObserver(this);
   observing_network_connectivity_changes_ = false;
 
   if (initialized_ && directory()) {
     directory()->SaveChanges();
@@ skipping 269 matching lines @@
   if (event.what_happened == SyncEngineEvent::SYNC_CYCLE_ENDED) {
     if (!initialized_) {
       LOG(INFO) << "OnSyncCycleCompleted not sent because sync api is not "
                 << "initialized";
       return;
     }
 
     DVLOG(1) << "Sending OnSyncCycleCompleted";
     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                       OnSyncCycleCompleted(event.snapshot));
+
+    // This is here for tests, which are still using p2p notifications.
+    bool is_notifiable_commit =
+        (event.snapshot.model_neutral_state().num_successful_commits > 0);
+    if (is_notifiable_commit) {
+      if (invalidator_) {
+        const ObjectIdInvalidationMap& invalidation_map =
+            ModelTypeInvalidationMapToObjectIdInvalidationMap(
+                event.snapshot.source().types);
+        invalidator_->SendInvalidation(invalidation_map);
+      } else {
+        DVLOG(1) << "Not sending invalidation: invalidator_ is NULL";
+      }
+    }
   }
 
   if (event.what_happened == SyncEngineEvent::STOP_SYNCING_PERMANENTLY) {
     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                       OnStopSyncingPermanently());
     return;
   }
 
   if (event.what_happened == SyncEngineEvent::UPDATED_TOKEN) {
     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
@@ skipping 202 matching lines @@
     const ModelTypeInvalidationMap& invalidation_map) {
   for (ModelTypeInvalidationMap::const_iterator it = invalidation_map.begin();
        it != invalidation_map.end(); ++it) {
     NotificationInfo* info = &notification_info_map_[it->first];
     info->total_count++;
     info->payload = it->second.payload;
   }
 }
 
 void SyncManagerImpl::OnInvalidatorStateChange(InvalidatorState state) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-
   const std::string& state_str = InvalidatorStateToString(state);
   invalidator_state_ = state;
   DVLOG(1) << "Invalidator state changed to: " << state_str;
   const bool notifications_enabled =
       (invalidator_state_ == INVALIDATIONS_ENABLED);
   allstatus_.SetNotificationsEnabled(notifications_enabled);
   scheduler_->SetNotificationsEnabled(notifications_enabled);
 
+  if (invalidator_state_ == syncer::INVALIDATION_CREDENTIALS_REJECTED) {
+    // If the invalidator's credentials were rejected, that means that
+    // our sync credentials are also bad, so invalidate those.
+    connection_manager_->OnInvalidationCredentialsRejected();
+  }
+
   if (js_event_handler_.IsInitialized()) {
     base::DictionaryValue details;
     details.SetString("state", state_str);
     js_event_handler_.Call(FROM_HERE,
                            &JsEventHandler::HandleJsEvent,
                            "onNotificationStateChange",
                            JsEventDetails(&details));
   }
 }
 
 void SyncManagerImpl::OnIncomingInvalidation(
     const ObjectIdInvalidationMap& invalidation_map) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
+  // TODO(dcheng): Acknowledge immediately for now. Fix this once the
+  // invalidator doesn't repeatedly ping for unacknowledged invaliations, since
+  // it conflicts with the sync scheduler's internal backoff algorithm.
+  // See http://crbug.com/124149 for more information.
+  for (ObjectIdInvalidationMap::const_iterator it = invalidation_map.begin();
+       it != invalidation_map.end(); ++it) {
+    invalidator_->Acknowledge(it->first, it->second.ack_handle);
+  }
+
   const ModelTypeInvalidationMap& type_invalidation_map =
       ObjectIdInvalidationMapToModelTypeInvalidationMap(invalidation_map);
   if (type_invalidation_map.empty()) {
     LOG(WARNING) << "Sync received invalidation without any type information.";
   } else {
     allstatus_.IncrementNudgeCounter(NUDGE_SOURCE_NOTIFICATION);
     scheduler_->ScheduleInvalidationNudge(
         TimeDelta::FromMilliseconds(kSyncSchedulerDelayMsec),
         type_invalidation_map, FROM_HERE);
     allstatus_.IncrementNotificationsReceived();
@@ skipping 122 matching lines @@
 int SyncManagerImpl::GetDefaultNudgeDelay() {
   return kDefaultNudgeDelayMilliseconds;
 }
 
 // static.
 int SyncManagerImpl::GetPreferencesNudgeDelay() {
   return kPreferencesNudgeDelayMilliseconds;
 }
 
 }  // namespace syncer