[Sync] Replace uses of ObserverListThreadSafe with WeakHandles

chrome/browser/sync/internal_api/sync_manager.cc

 // Copyright (c) 2011 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/internal_api/sync_manager.h"
 
 #include <string>
 
 #include "base/base64.h"
 #include "base/command_line.h"
 #include "base/compiler_specific.h"
 #include "base/json/json_writer.h"
 #include "base/memory/ref_counted.h"
 #include "base/observer_list.h"
-#include "base/observer_list_threadsafe.h"
 #include "base/string_number_conversions.h"
 #include "base/values.h"
 #include "chrome/browser/sync/engine/all_status.h"
 #include "chrome/browser/sync/engine/net/server_connection_manager.h"
 #include "chrome/browser/sync/engine/nigori_util.h"
 #include "chrome/browser/sync/engine/syncapi_internal.h"
 #include "chrome/browser/sync/engine/syncer_types.h"
 #include "chrome/browser/sync/engine/sync_scheduler.h"
 #include "chrome/browser/sync/internal_api/base_node.h"
 #include "chrome/browser/sync/internal_api/change_reorder_buffer.h"
@@ skipping 98 matching lines @@
       public JsBackend,
       public SyncEngineEventListener,
       public ServerConnectionEventListener,
       public syncable::DirectoryChangeDelegate {
   static const int kDefaultNudgeDelayMilliseconds;
   static const int kPreferencesNudgeDelayMilliseconds;
  public:
   explicit SyncInternal(const std::string& name)
       : name_(name),
         weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
-        change_observers_(
-            new ObserverListThreadSafe<SyncManager::ChangeObserver>()),
         registrar_(NULL),
         change_delegate_(NULL),
         initialized_(false),
         setup_for_test_mode_(false),
         observing_ip_address_changes_(false),
         created_on_loop_(MessageLoop::current()) {
     // Pre-fill |notification_info_map_|.
     for (int i = syncable::FIRST_REAL_MODEL_TYPE;
          i < syncable::MODEL_TYPE_COUNT; ++i) {
       notification_info_map_.insert(
@@ skipping 124 matching lines @@
 
   // SyncNotifierObserver implementation.
   virtual void OnNotificationStateChange(
       bool notifications_enabled) OVERRIDE;
 
   virtual void OnIncomingNotification(
       const syncable::ModelTypePayloadMap& type_payloads) OVERRIDE;
 
   virtual void StoreState(const std::string& cookie) OVERRIDE;
 
-  void AddChangeObserver(SyncManager::ChangeObserver* observer);
-  void RemoveChangeObserver(SyncManager::ChangeObserver* observer);
-
   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();
   }
   SyncScheduler* scheduler() { return scheduler_.get(); }
   UserShare* GetUserShare() {
@@ skipping 195 matching lines @@
   // we'd have another worker class which implements
   // HandleCalculateChangesChangeEventFromSyncApi() and we'd pass it a
   // WeakHandle when we construct it.
   WeakHandle<SyncInternal> weak_handle_this_;
 
   // We couple the DirectoryManager and username together in a UserShare member
   // so we can return a handle to share_ to clients of the API for use when
   // constructing any transaction type.
   UserShare share_;
 
-  // Even though observers are always added/removed from the sync
-  // thread, we still need to use a thread-safe observer list as we
-  // can notify from any thread.
-  scoped_refptr<ObserverListThreadSafe<SyncManager::ChangeObserver> >
-      change_observers_;
+  // This can be called from any thread, but only between calls to
+  // OpenDirectory() and ShutdownOnSyncThread().
+  browser_sync::WeakHandle<SyncManager::ChangeObserver> change_observer_;
 
   ObserverList<SyncManager::Observer> observers_;
 
   // The ServerConnectionManager used to abstract communication between the
   // client (the Syncer) and the sync server.
   scoped_ptr<SyncAPIServerConnectionManager> connection_manager_;
 
   // The scheduler that runs the Syncer. Needs to be explicitly
   // Start()ed.
   scoped_ptr<SyncScheduler> scheduler_;
@@ skipping 373 matching lines @@
   // Start the sync scheduler. This won't actually result in any
   // syncing until at least the DirectoryManager broadcasts the OPENED
   // event, and a valid server connection is detected.
   if (scheduler())  // NULL during certain unittests.
     scheduler()->Start(SyncScheduler::NORMAL_MODE, base::Closure());
 }
 
 bool SyncManager::SyncInternal::OpenDirectory() {
   DCHECK(!initialized_) << "Should only happen once";
 
-  bool share_opened = dir_manager()->Open(username_for_share(), this);
+  // Set before Open().
+  change_observer_ =
+      browser_sync::MakeWeakHandle(js_mutation_event_observer_.AsWeakPtr());
+
+  bool share_opened =
+      dir_manager()->Open(
+          username_for_share(),
+          this,
+          browser_sync::MakeWeakHandle(
+              js_mutation_event_observer_.AsWeakPtr()));
   DCHECK(share_opened);
   if (!share_opened) {
     LOG(ERROR) << "Could not open share for:" << username_for_share();
     return false;
   }
 
   // Database has to be initialized for the guid to be available.
   syncable::ScopedDirLookup lookup(dir_manager(), username_for_share());
   if (!lookup.good()) {
     NOTREACHED();
     return false;
   }
 
   connection_manager()->set_client_id(lookup->cache_guid());
-  lookup->AddTransactionObserver(&js_mutation_event_observer_);
-  AddChangeObserver(&js_mutation_event_observer_);
   return true;
 }
 
 bool SyncManager::SyncInternal::SignIn(const SyncCredentials& credentials) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(share_.name.empty());
   share_.name = credentials.email;
 
   VLOG(1) << "Signing in user: " << username_for_share();
   if (!OpenDirectory())
@@ skipping 293 matching lines @@
   // NOTE: We notify from within a transaction.
   FOR_EACH_OBSERVER(
       SyncManager::Observer, observers_, OnEncryptionComplete());
 }
 
 SyncManager::~SyncManager() {
   DCHECK(thread_checker_.CalledOnValidThread());
   delete data_;
 }
 
-void SyncManager::AddChangeObserver(ChangeObserver* observer) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  data_->AddChangeObserver(observer);
-}
-
-void SyncManager::RemoveChangeObserver(ChangeObserver* observer) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  data_->RemoveChangeObserver(observer);
-}
-
 void SyncManager::AddObserver(Observer* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   data_->AddObserver(observer);
 }
 
 void SyncManager::RemoveObserver(Observer* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   data_->RemoveObserver(observer);
 }
 
@@ skipping 15 matching lines @@
 
 void SyncManager::ShutdownOnSyncThread() {
   DCHECK(thread_checker_.CalledOnValidThread());
   data_->ShutdownOnSyncThread();
 }
 
 void SyncManager::SyncInternal::ShutdownOnSyncThread() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   // Prevent any in-flight method calls from running.  Also
-  // invalidates |weak_handle_this_|.
+  // invalidates |weak_handle_this_| and |change_observer_|.
   weak_ptr_factory_.InvalidateWeakPtrs();
+  js_mutation_event_observer_.InvalidateWeakPtrs();
 
   scheduler_.reset();
 
   SetJsEventHandler(WeakHandle<JsEventHandler>());
   RemoveObserver(&js_sync_manager_observer_);
 
   RemoveObserver(&debug_info_event_listener_);
 
   if (sync_notifier_.get()) {
     sync_notifier_->RemoveObserver(this);
   }
   sync_notifier_.reset();
 
   if (connection_manager_.get()) {
     connection_manager_->RemoveListener(this);
   }
   connection_manager_.reset();
 
   net::NetworkChangeNotifier::RemoveIPAddressObserver(this);
   observing_ip_address_changes_ = false;
 
   if (dir_manager()) {
     {
       // Cryptographer should only be accessed while holding a
       // transaction.
       ReadTransaction trans(FROM_HERE, GetUserShare());
       trans.GetCryptographer()->RemoveObserver(this);
-      trans.GetLookup()->
-          RemoveTransactionObserver(&js_mutation_event_observer_);
-      RemoveChangeObserver(&js_mutation_event_observer_);
     }
     dir_manager()->FinalSaveChangesForAll();
     dir_manager()->Close(username_for_share());
   }
 
   // Reset the DirectoryManager and UserSettings so they relinquish sqlite
   // handles to backing files.
   share_.dir_manager.reset();
 
   setup_for_test_mode_ = false;
   change_delegate_ = NULL;
   registrar_ = NULL;
 
   initialized_ = false;
 
-  // We reset this here, since only now we know it will not be
+  // We reset these here, since only now we know they will not be
   // accessed from other threads (since we shut down everything).
+  change_observer_.Reset();
   weak_handle_this_.Reset();
 }
 
 void SyncManager::SyncInternal::OnIPAddressChanged() {
   VLOG(1) << "IP address change detected";
   if (!observing_ip_address_changes_) {
     VLOG(1) << "IP address change dropped.";
     return;
   }
 
@@ skipping 45 matching lines @@
   // released. This allows work to be performed without blocking other threads
   // from acquiring a transaction.
   if (!change_delegate_)
     return;
 
   // Call commit.
   for (int i = 0; i < syncable::MODEL_TYPE_COUNT; ++i) {
     const syncable::ModelType type = syncable::ModelTypeFromInt(i);
     if (models_with_changes.test(type)) {
       change_delegate_->OnChangesComplete(type);
-      change_observers_->Notify(
+      change_observer_.Call(FROM_HERE,
           &SyncManager::ChangeObserver::OnChangesComplete, type);
     }
   }
 }
 
 ModelTypeBitSet SyncManager::SyncInternal::HandleTransactionEndingChangeEvent(
     const ImmutableWriteTransactionInfo& write_transaction_info,
     syncable::BaseTransaction* trans) {
   // This notification happens immediately before a syncable WriteTransaction
   // falls out of scope. It happens while the channel mutex is still held,
@@ skipping 15 matching lines @@
       continue;
 
     ImmutableChangeRecordList ordered_changes;
     // TODO(akalin): Propagate up the error further (see
     // http://crbug.com/100907).
     CHECK(change_buffers_[type].GetAllChangesInTreeOrder(&read_trans,
                                                          &ordered_changes));
     if (!ordered_changes.Get().empty()) {
       change_delegate_->
           OnChangesApplied(type, &read_trans, ordered_changes);
-      change_observers_->Notify(
+      change_observer_.Call(FROM_HERE,
           &SyncManager::ChangeObserver::OnChangesApplied,
           type, write_transaction_info.Get().id, ordered_changes);
       models_with_changes.set(i, true);
     }
     change_buffers_[i].Clear();
   }
   return models_with_changes;
 }
 
 void SyncManager::SyncInternal::HandleCalculateChangesChangeEventFromSyncApi(
@@ skipping 523 matching lines @@
   }
   if (VLOG_IS_ON(1)) {
     std::string encoded_state;
     base::Base64Encode(state, &encoded_state);
     VLOG(1) << "Writing notification state: " << encoded_state;
   }
   lookup->SetNotificationState(state);
   lookup->SaveChanges();
 }
 
-void SyncManager::SyncInternal::AddChangeObserver(
-    SyncManager::ChangeObserver* observer) {
-  change_observers_->AddObserver(observer);
-}
-
-void SyncManager::SyncInternal::RemoveChangeObserver(
-    SyncManager::ChangeObserver* observer) {
-  change_observers_->RemoveObserver(observer);
-}
-
 void SyncManager::SyncInternal::AddObserver(
     SyncManager::Observer* observer) {
   observers_.AddObserver(observer);
 }
 
 void SyncManager::SyncInternal::RemoveObserver(
     SyncManager::Observer* observer) {
   observers_.RemoveObserver(observer);
 }
 
@@ skipping 124 matching lines @@
     lookup->GetDownloadProgress(*i, &marker);
 
     if (marker.token().empty())
       result.insert(*i);
 
   }
   return result;
 }
 
 }  // namespace sync_api