[Sync] Replace uses of ObserverListThreadSafe with WeakHandles

ObserverListThreadSafe was overkill since there was only two threads
involved, and only one observer.

Add MessageLoop member variables to various test classes that need it
(ObserverListThreadSafe had a hack that made it unnecessary before.)

BUG=103732
TEST=Start with a clean profile and set up sync.  about:profiler shouldn't
show ObserverListThreadSafe::Notify calls with sync threads anymore.


Review URL: http://codereview.chromium.org/8586014

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@110836 0039d316-1c4b-4281-b951-d872f2087c98

chrome/browser/sync/engine/syncer_proto_util_unittest.cc

@@ -8,6 +8,7 @@
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
+#include "base/message_loop.h"
 #include "chrome/browser/sync/engine/syncproto.h"
 #include "chrome/browser/sync/syncable/blob.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
@@ -137,6 +138,7 @@ class SyncerProtoUtilTest : public testing::Test {
   }
 
  protected:
+  MessageLoop message_loop_;
   browser_sync::TestDirectorySetterUpper setter_upper_;
 };
 

chrome/browser/sync/engine/syncer_unittest.cc

@@ -16,6 +16,7 @@
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/message_loop.h"
 #include "base/string_number_conversions.h"
 #include "base/stringprintf.h"
 #include "base/time.h"
@@ -468,6 +469,8 @@ class SyncerTest : public testing::Test,
     return GetField(metahandle, field, false);
   }
 
+  MessageLoop message_loop_;
+
   // Some ids to aid tests. Only the root one's value is specific. The rest
   // are named for test clarity.
   // TODO(chron): Get rid of these inbuilt IDs. They only make it

chrome/browser/sync/internal_api/debug_info_event_listener.h

@@ -5,6 +5,7 @@
 #ifndef CHROME_BROWSER_SYNC_INTERNAL_API_DEBUG_INFO_EVENT_LISTENER_H_
 #define CHROME_BROWSER_SYNC_INTERNAL_API_DEBUG_INFO_EVENT_LISTENER_H_
 
+#include <queue>
 #include <string>
 
 #include "chrome/browser/sync/internal_api/sync_manager.h"

chrome/browser/sync/internal_api/sync_manager.cc

@@ -12,7 +12,6 @@
 #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"
@@ -131,8 +130,6 @@ class SyncManager::SyncInternal
   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),
@@ -277,9 +274,6 @@ class SyncManager::SyncInternal
 
   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);
 
@@ -495,11 +489,9 @@ class SyncManager::SyncInternal
   // 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_;
 
@@ -893,7 +885,16 @@ void SyncManager::SyncInternal::StartSyncingNormally() {
 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()));
   if (!share_opened) {
     LOG(ERROR) << "Could not open share for:" << username_for_share();
     return false;
@@ -907,8 +908,6 @@ bool SyncManager::SyncInternal::OpenDirectory() {
   }
 
   connection_manager()->set_client_id(lookup->cache_guid());
-  lookup->AddTransactionObserver(&js_mutation_event_observer_);
-  AddChangeObserver(&js_mutation_event_observer_);
   return true;
 }
 
@@ -1223,16 +1222,6 @@ SyncManager::~SyncManager() {
   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);
@@ -1268,8 +1257,9 @@ 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();
 
@@ -1297,9 +1287,6 @@ void SyncManager::SyncInternal::ShutdownOnSyncThread() {
       // 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());
@@ -1315,8 +1302,9 @@ void SyncManager::SyncInternal::ShutdownOnSyncThread() {
 
   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();
 }
 
@@ -1382,7 +1370,7 @@ void SyncManager::SyncInternal::HandleTransactionCompleteChangeEvent(
     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);
     }
   }
@@ -1418,7 +1406,7 @@ ModelTypeBitSet SyncManager::SyncInternal::HandleTransactionEndingChangeEvent(
     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);
@@ -1962,16 +1950,6 @@ void SyncManager::SyncInternal::StoreState(
   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);

chrome/browser/sync/internal_api/sync_manager.h

@@ -496,10 +496,6 @@ class SyncManager {
   // Request a clearing of all data on the server
   void RequestClearServerData();
 
-  // Add/remove change observers.
-  void AddChangeObserver(ChangeObserver* observer);
-  void RemoveChangeObserver(ChangeObserver* observer);
-
   // Adds a listener to be notified of sync events.
   // NOTE: It is OK (in fact, it's probably a good idea) to call this before
   // having received OnInitializationCompleted.

chrome/browser/sync/internal_api/syncapi_unittest.cc

@@ -192,6 +192,7 @@ class SyncApiTest : public testing::Test {
   }
 
  protected:
+  MessageLoop message_loop_;
   browser_sync::TestUserShare test_user_share_;
 };
 

chrome/browser/sync/js/js_mutation_event_observer.cc

@@ -15,12 +15,21 @@
 
 namespace browser_sync {
 
-JsMutationEventObserver::JsMutationEventObserver() {}
+JsMutationEventObserver::JsMutationEventObserver()
+    : weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {}
 
 JsMutationEventObserver::~JsMutationEventObserver() {
   DCHECK(non_thread_safe_.CalledOnValidThread());
 }
 
+base::WeakPtr<JsMutationEventObserver> JsMutationEventObserver::AsWeakPtr() {
+  return weak_ptr_factory_.GetWeakPtr();
+}
+
+void JsMutationEventObserver::InvalidateWeakPtrs() {
+  weak_ptr_factory_.InvalidateWeakPtrs();
+}
+
 void JsMutationEventObserver::SetJsEventHandler(
     const WeakHandle<JsEventHandler>& event_handler) {
   event_handler_ = event_handler;

chrome/browser/sync/js/js_mutation_event_observer.h

@@ -10,6 +10,7 @@
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
+#include "base/memory/weak_ptr.h"
 #include "base/threading/non_thread_safe.h"
 #include "chrome/browser/sync/internal_api/sync_manager.h"
 #include "chrome/browser/sync/syncable/transaction_observer.h"
@@ -34,6 +35,10 @@ class JsMutationEventObserver
 
   virtual ~JsMutationEventObserver();
 
+  base::WeakPtr<JsMutationEventObserver> AsWeakPtr();
+
+  void InvalidateWeakPtrs();
+
   void SetJsEventHandler(const WeakHandle<JsEventHandler>& event_handler);
 
   // sync_api::SyncManager::ChangeObserver implementation.
@@ -56,6 +61,7 @@ class JsMutationEventObserver
 
  private:
   base::NonThreadSafe non_thread_safe_;
+  base::WeakPtrFactory<JsMutationEventObserver> weak_ptr_factory_;
   WeakHandle<JsEventHandler> event_handler_;
 
   void HandleJsEvent(

chrome/browser/sync/notifier/non_blocking_invalidation_notifier.cc

@@ -7,10 +7,8 @@
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop.h"
-#include "base/observer_list_threadsafe.h"
 #include "base/threading/thread.h"
 #include "chrome/browser/sync/notifier/invalidation_notifier.h"
-#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
 
 namespace sync_notifier {
 
@@ -18,12 +16,11 @@ class NonBlockingInvalidationNotifier::Core
     : public base::RefCountedThreadSafe<NonBlockingInvalidationNotifier::Core>,
       public SyncNotifierObserver {
  public:
-  // Called on parent thread.
-  Core();
-
-  // Called on parent thread.
-  void AddObserver(SyncNotifierObserver* observer);
-  void RemoveObserver(SyncNotifierObserver* observer);
+  // Called on parent thread.  |delegate_observer| should be
+  // initialized.
+  explicit Core(
+      const browser_sync::WeakHandle<SyncNotifierObserver>&
+          delegate_observer);
 
   // Helpers called on I/O thread.
   void Initialize(
@@ -50,14 +47,19 @@ class NonBlockingInvalidationNotifier::Core
   // Called on parent or I/O thread.
   ~Core();
 
+  // The variables below should be used only on the I/O thread.
+  const browser_sync::WeakHandle<SyncNotifierObserver> delegate_observer_;
   scoped_ptr<InvalidationNotifier> invalidation_notifier_;
   scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy_;
-  scoped_refptr<ObserverListThreadSafe<SyncNotifierObserver> > observers_;
+
   DISALLOW_COPY_AND_ASSIGN(Core);
 };
 
-NonBlockingInvalidationNotifier::Core::Core()
-    : observers_(new ObserverListThreadSafe<SyncNotifierObserver>()) {
+NonBlockingInvalidationNotifier::Core::Core(
+    const browser_sync::WeakHandle<SyncNotifierObserver>&
+        delegate_observer)
+    : delegate_observer_(delegate_observer) {
+  DCHECK(delegate_observer_.IsInitialized());
 }
 
 NonBlockingInvalidationNotifier::Core::~Core() {
@@ -92,16 +94,6 @@ void NonBlockingInvalidationNotifier::Core::Teardown() {
   io_message_loop_proxy_ = NULL;
 }
 
-void NonBlockingInvalidationNotifier::Core::AddObserver(
-    SyncNotifierObserver* observer) {
-  observers_->AddObserver(observer);
-}
-
-void NonBlockingInvalidationNotifier::Core::RemoveObserver(
-    SyncNotifierObserver* observer) {
-  observers_->RemoveObserver(observer);
-}
-
 void NonBlockingInvalidationNotifier::Core::SetUniqueId(
     const std::string& unique_id) {
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
@@ -129,21 +121,24 @@ void NonBlockingInvalidationNotifier::Core::UpdateEnabledTypes(
 void NonBlockingInvalidationNotifier::Core::OnIncomingNotification(
         const syncable::ModelTypePayloadMap& type_payloads) {
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
-  observers_->Notify(&SyncNotifierObserver::OnIncomingNotification,
-                     type_payloads);
+  delegate_observer_.Call(FROM_HERE,
+                          &SyncNotifierObserver::OnIncomingNotification,
+                          type_payloads);
 }
 
 void NonBlockingInvalidationNotifier::Core::OnNotificationStateChange(
         bool notifications_enabled) {
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
-  observers_->Notify(&SyncNotifierObserver::OnNotificationStateChange,
-                     notifications_enabled);
+  delegate_observer_.Call(FROM_HERE,
+                          &SyncNotifierObserver::OnNotificationStateChange,
+                          notifications_enabled);
 }
 
 void NonBlockingInvalidationNotifier::Core::StoreState(
     const std::string& state) {
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
-  observers_->Notify(&SyncNotifierObserver::StoreState, state);
+  delegate_observer_.Call(FROM_HERE,
+                          &SyncNotifierObserver::StoreState, state);
 }
 
 NonBlockingInvalidationNotifier::NonBlockingInvalidationNotifier(
@@ -152,7 +147,10 @@ NonBlockingInvalidationNotifier::NonBlockingInvalidationNotifier(
     const browser_sync::WeakHandle<InvalidationVersionTracker>&
         invalidation_version_tracker,
     const std::string& client_info)
-        : core_(new Core),
+        : weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
+          core_(
+              new Core(browser_sync::MakeWeakHandle(
+                  weak_ptr_factory_.GetWeakPtr()))),
           parent_message_loop_proxy_(
               base::MessageLoopProxy::current()),
           io_message_loop_proxy_(notifier_options.request_context_getter->
@@ -183,13 +181,13 @@ NonBlockingInvalidationNotifier::~NonBlockingInvalidationNotifier() {
 void NonBlockingInvalidationNotifier::AddObserver(
     SyncNotifierObserver* observer) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  core_->AddObserver(observer);
+  observers_.AddObserver(observer);
 }
 
 void NonBlockingInvalidationNotifier::RemoveObserver(
     SyncNotifierObserver* observer) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  core_->RemoveObserver(observer);
+  observers_.RemoveObserver(observer);
 }
 
 void NonBlockingInvalidationNotifier::SetUniqueId(
@@ -242,4 +240,25 @@ void NonBlockingInvalidationNotifier::SendNotification(
   // need to forward on the call.
 }
 
+void NonBlockingInvalidationNotifier::OnIncomingNotification(
+        const syncable::ModelTypePayloadMap& type_payloads) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
+                    OnIncomingNotification(type_payloads));
+}
+
+void NonBlockingInvalidationNotifier::OnNotificationStateChange(
+        bool notifications_enabled) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
+                    OnNotificationStateChange(notifications_enabled));
+}
+
+void NonBlockingInvalidationNotifier::StoreState(
+    const std::string& state) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
+                    StoreState(state));
+}
+
 }  // namespace sync_notifier

chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h

@@ -14,8 +14,11 @@
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
 #include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
+#include "base/observer_list.h"
 #include "chrome/browser/sync/notifier/invalidation_version_tracker.h"
 #include "chrome/browser/sync/notifier/sync_notifier.h"
+#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
 #include "chrome/browser/sync/util/weak_handle.h"
 #include "jingle/notifier/base/notifier_options.h"
 
@@ -25,7 +28,9 @@ class MessageLoopProxy;
 
 namespace sync_notifier {
 
-class NonBlockingInvalidationNotifier : public SyncNotifier {
+class NonBlockingInvalidationNotifier
+    : public SyncNotifier,
+      public SyncNotifierObserver {
  public:
   // |invalidation_version_tracker| must be initialized.
   NonBlockingInvalidationNotifier(
@@ -49,13 +54,26 @@ class NonBlockingInvalidationNotifier : public SyncNotifier {
   virtual void SendNotification(
       const syncable::ModelTypeSet& changed_types) OVERRIDE;
 
+  // SyncNotifierObserver implementation.
+  virtual void OnIncomingNotification(
+      const syncable::ModelTypePayloadMap& type_payloads) OVERRIDE;
+  virtual void OnNotificationStateChange(bool notifications_enabled) OVERRIDE;
+  virtual void StoreState(const std::string& state) OVERRIDE;
+
  private:
-  // The real guts of NonBlockingInvalidationNotifier, which allows this class
-  // to not be refcounted.
   class Core;
+
+  base::WeakPtrFactory<NonBlockingInvalidationNotifier> weak_ptr_factory_;
+
+  // Our observers (which must live on the parent thread).
+  ObserverList<SyncNotifierObserver> observers_;
+
+  // The real guts of NonBlockingInvalidationNotifier, which allows
+  // this class to live completely on the parent thread.
   scoped_refptr<Core> core_;
   scoped_refptr<base::MessageLoopProxy> parent_message_loop_proxy_;
   scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy_;
+
   DISALLOW_COPY_AND_ASSIGN(NonBlockingInvalidationNotifier);
 };
 

chrome/browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc

@@ -66,18 +66,29 @@ class NonBlockingInvalidationNotifierTest : public testing::Test {
 };
 
 TEST_F(NonBlockingInvalidationNotifierTest, Basic) {
-  syncable::ModelTypeSet types;
-  types.insert(syncable::BOOKMARKS);
-  types.insert(syncable::AUTOFILL);
+  InSequence dummy;
+
+  syncable::ModelTypePayloadMap type_payloads;
+  type_payloads[syncable::PREFERENCES] = "payload";
+  type_payloads[syncable::BOOKMARKS] = "";
+  type_payloads[syncable::AUTOFILL] = "";
+
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
+  EXPECT_CALL(mock_observer_, StoreState("new_fake_state"));
+  EXPECT_CALL(mock_observer_, OnIncomingNotification(type_payloads));
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(false));
 
-  invalidation_notifier_->SetUniqueId("fake_id");
   invalidation_notifier_->SetState("fake_state");
+  invalidation_notifier_->SetUniqueId("fake_id");
   invalidation_notifier_->UpdateCredentials("foo@bar.com", "fake_token");
-  invalidation_notifier_->UpdateEnabledTypes(types);
-}
 
-// TODO(akalin): Add synchronous operations for testing to
-// NonBlockingInvalidationNotifierTest and use that to test it.
+  invalidation_notifier_->OnNotificationStateChange(true);
+  invalidation_notifier_->StoreState("new_fake_state");
+  invalidation_notifier_->OnIncomingNotification(type_payloads);
+  invalidation_notifier_->OnNotificationStateChange(false);
+
+  ui_loop_.RunAllPending();
+}
 
 }  // namespace
 

chrome/browser/sync/sessions/sync_session_unittest.cc

@@ -7,6 +7,7 @@
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/memory/ref_counted.h"
+#include "base/message_loop.h"
 #include "chrome/browser/sync/engine/conflict_resolver.h"
 #include "chrome/browser/sync/engine/mock_model_safe_workers.h"
 #include "chrome/browser/sync/engine/syncer_types.h"
@@ -101,6 +102,7 @@ class SyncSessionTest : public testing::Test,
     return request_params;
   }
 
+  MessageLoop message_loop_;
   bool controller_invocations_allowed_;
   scoped_ptr<SyncSession> session_;
   scoped_ptr<SyncSessionContext> context_;

chrome/browser/sync/syncable/directory_manager.cc

@@ -42,19 +42,26 @@ DirectoryManager::~DirectoryManager() {
       << "Dir " << managed_directory_->name() << " not closed!";
 }
 
-bool DirectoryManager::Open(const std::string& name,
-                            DirectoryChangeDelegate* delegate) {
+bool DirectoryManager::Open(
+    const std::string& name,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer) {
   bool was_open = false;
   const DirOpenResult result =
-      OpenImpl(name, GetSyncDataDatabasePath(), delegate, &was_open);
+      OpenImpl(name, GetSyncDataDatabasePath(), delegate,
+               transaction_observer, &was_open);
   return syncable::OPENED == result;
 }
 
 // Opens a directory.  Returns false on error.
-DirOpenResult DirectoryManager::OpenImpl(const std::string& name,
-                                         const FilePath& path,
-                                         DirectoryChangeDelegate* delegate,
-                                         bool* was_open) {
+DirOpenResult DirectoryManager::OpenImpl(
+    const std::string& name,
+    const FilePath& path,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer,
+    bool* was_open) {
   bool opened = false;
   {
     base::AutoLock lock(lock_);
@@ -72,7 +79,8 @@ DirOpenResult DirectoryManager::OpenImpl(const std::string& name,
   // Otherwise, open it.
 
   scoped_ptr<Directory> dir(new Directory);
-  const DirOpenResult result = dir->Open(path, name, delegate);
+  const DirOpenResult result =
+      dir->Open(path, name, delegate, transaction_observer);
   if (syncable::OPENED == result) {
     base::AutoLock lock(lock_);
     managed_directory_ = dir.release();

chrome/browser/sync/syncable/directory_manager.h

@@ -22,6 +22,7 @@
 #include "chrome/browser/sync/syncable/dir_open_result.h"
 #include "chrome/browser/sync/syncable/syncable.h"
 #include "chrome/browser/sync/util/cryptographer.h"
+#include "chrome/browser/sync/util/weak_handle.h"
 
 namespace sync_api { class BaseTransaction; }
 namespace syncable { class BaseTransaction; }
@@ -45,8 +46,10 @@ class DirectoryManager {
   // common case.  Does not take ownership of |delegate|, which must
   // be non-NULL.  Starts sending events to |delegate| if the returned
   // result is true.  Note that events to |delegate| may be sent from
-  // *any* thread.
-  bool Open(const std::string& name, DirectoryChangeDelegate* delegate);
+  // *any* thread.  |transaction_observer| must be initialized.
+  bool Open(const std::string& name, DirectoryChangeDelegate* delegate,
+            const browser_sync::WeakHandle<TransactionObserver>&
+                transaction_observer);
 
   // Marks a directory as closed and stops sending events to the
   // delegate.  It might take a while until all the file handles and
@@ -74,8 +77,12 @@ class DirectoryManager {
     return cryptographer_.get();
   }
 
-  DirOpenResult OpenImpl(const std::string& name, const FilePath& path,
-                         DirectoryChangeDelegate* delegate, bool* was_open);
+  DirOpenResult OpenImpl(
+      const std::string& name, const FilePath& path,
+      DirectoryChangeDelegate* delegate,
+      const browser_sync::WeakHandle<TransactionObserver>&
+          transaction_observer,
+      bool* was_open);
 
   // Helpers for friend class ScopedDirLookup:
   friend class ScopedDirLookup;

chrome/browser/sync/syncable/syncable.cc

@@ -349,10 +349,14 @@ DictionaryValue* EntryKernel::ToValue() const {
 ///////////////////////////////////////////////////////////////////////////
 // Directory
 
-void Directory::InitKernel(const std::string& name,
-                           DirectoryChangeDelegate* delegate) {
-  DCHECK(kernel_ == NULL);
-  kernel_ = new Kernel(FilePath(), name, KernelLoadInfo(), delegate);
+void Directory::InitKernelForTest(
+    const std::string& name,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer) {
+  DCHECK(!kernel_);
+  kernel_ =  new Kernel(FilePath(), name, KernelLoadInfo(),
+                        delegate, transaction_observer);
 }
 
 Directory::PersistedKernelInfo::PersistedKernelInfo()
@@ -378,10 +382,11 @@ Directory::SaveChangesSnapshot::SaveChangesSnapshot()
 
 Directory::SaveChangesSnapshot::~SaveChangesSnapshot() {}
 
-Directory::Kernel::Kernel(const FilePath& db_path,
-                          const string& name,
-                          const KernelLoadInfo& info,
-                          DirectoryChangeDelegate* delegate)
+Directory::Kernel::Kernel(
+    const FilePath& db_path, const string& name,
+    const KernelLoadInfo& info, DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer)
     : db_path(db_path),
       refcount(1),
       next_write_transaction_id(0),
@@ -399,8 +404,9 @@ Directory::Kernel::Kernel(const FilePath& db_path,
       cache_guid(info.cache_guid),
       next_metahandle(info.max_metahandle + 1),
       delegate(delegate),
-      observers(new ObserverListThreadSafe<TransactionObserver>()) {
+      transaction_observer(transaction_observer) {
   DCHECK(delegate);
+  DCHECK(transaction_observer.IsInitialized());
 }
 
 void Directory::Kernel::AddRef() {
@@ -432,9 +438,13 @@ Directory::~Directory() {
   Close();
 }
 
-DirOpenResult Directory::Open(const FilePath& file_path, const string& name,
-                              DirectoryChangeDelegate* delegate) {
-  const DirOpenResult result = OpenImpl(file_path, name, delegate);
+DirOpenResult Directory::Open(
+    const FilePath& file_path, const string& name,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer) {
+  const DirOpenResult result =
+      OpenImpl(file_path, name, delegate, transaction_observer);
   if (OPENED != result)
     Close();
   return result;
@@ -461,9 +471,12 @@ DirectoryBackingStore* Directory::CreateBackingStore(
   return new DirectoryBackingStore(dir_name, backing_filepath);
 }
 
-DirOpenResult Directory::OpenImpl(const FilePath& file_path,
-                                  const string& name,
-                                  DirectoryChangeDelegate* delegate) {
+DirOpenResult Directory::OpenImpl(
+    const FilePath& file_path,
+    const string& name,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer) {
   DCHECK_EQ(static_cast<DirectoryBackingStore*>(NULL), store_);
   FilePath db_path(file_path);
   file_util::AbsolutePath(&db_path);
@@ -477,7 +490,7 @@ DirOpenResult Directory::OpenImpl(const FilePath& file_path,
   if (OPENED != result)
     return result;
 
-  kernel_ = new Kernel(db_path, name, info, delegate);
+  kernel_ = new Kernel(db_path, name, info, delegate, transaction_observer);
   kernel_->metahandles_index->swap(metas_bucket);
   InitializeIndices();
   return OPENED;
@@ -1109,14 +1122,6 @@ void Directory::CheckTreeInvariants(syncable::BaseTransaction* trans,
   }
 }
 
-void Directory::AddTransactionObserver(TransactionObserver* observer) {
-  kernel_->observers->AddObserver(observer);
-}
-
-void Directory::RemoveTransactionObserver(TransactionObserver* observer) {
-  kernel_->observers->RemoveObserver(observer);
-}
-
 ///////////////////////////////////////////////////////////////////////////////
 // ScopedKernelLock
 
@@ -1153,13 +1158,13 @@ BaseTransaction::BaseTransaction(const tracked_objects::Location& from_here,
                                  Directory* directory)
     : from_here_(from_here), name_(name), writer_(writer),
       directory_(directory), dirkernel_(directory->kernel_) {
-  dirkernel_->observers->Notify(
+  dirkernel_->transaction_observer.Call(FROM_HERE,
       &TransactionObserver::OnTransactionStart, from_here_, writer_);
 }
 
 BaseTransaction::~BaseTransaction() {
   if (writer_ != INVALID) {
-    dirkernel_->observers->Notify(
+    dirkernel_->transaction_observer.Call(FROM_HERE,
         &TransactionObserver::OnTransactionEnd, from_here_, writer_);
   }
 }
@@ -1268,7 +1273,7 @@ ModelTypeBitSet WriteTransaction::NotifyTransactionChangingAndEnding(
       delegate->HandleTransactionEndingChangeEvent(
           immutable_write_transaction_info, this);
 
-  dirkernel_->observers->Notify(
+  dirkernel_->transaction_observer.Call(FROM_HERE,
       &TransactionObserver::OnTransactionWrite,
       immutable_write_transaction_info, models_with_changes);
 

chrome/browser/sync/syncable/syncable.h

@@ -24,7 +24,6 @@
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/memory/ref_counted.h"
-#include "base/observer_list_threadsafe.h"
 #include "base/synchronization/lock.h"
 #include "base/time.h"
 #include "chrome/browser/sync/protocol/sync.pb.h"
@@ -35,6 +34,7 @@
 #include "chrome/browser/sync/syncable/model_type.h"
 #include "chrome/browser/sync/util/immutable.h"
 #include "chrome/browser/sync/util/time.h"
+#include "chrome/browser/sync/util/weak_handle.h"
 
 namespace base {
 class DictionaryValue;
@@ -821,11 +821,14 @@ class Directory {
   // Does not take ownership of |delegate|, which must not be NULL.
   // Starts sending events to |delegate| if the returned result is
   // OPENED.  Note that events to |delegate| may be sent from *any*
-  // thread.
+  // thread.  |transaction_observer| must be initialized.
   DirOpenResult Open(const FilePath& file_path, const std::string& name,
-                     DirectoryChangeDelegate* delegate);
+                     DirectoryChangeDelegate* delegate,
+                     const browser_sync::WeakHandle<TransactionObserver>&
+                         transaction_observer);
 
-  // Stops sending events to the delegate.
+  // Stops sending events to the delegate and the transaction
+  // observer.
   void Close();
 
   int64 NextMetahandle();
@@ -866,12 +869,6 @@ class Directory {
   // Unique to each account / client pair.
   std::string cache_guid() const;
 
-  // These are backed by a thread-safe observer list, and so can be
-  // called on any thread, and events will be sent to the observer on
-  // the same thread that it was added on.
-  void AddTransactionObserver(TransactionObserver* observer);
-  void RemoveTransactionObserver(TransactionObserver* observer);
-
  protected:  // for friends, mainly used by Entry constructors
   virtual EntryKernel* GetEntryByHandle(int64 handle);
   virtual EntryKernel* GetEntryByHandle(int64 metahandle,
@@ -900,8 +897,11 @@ class Directory {
   // before calling.
   EntryKernel* GetEntryById(const Id& id, ScopedKernelLock* const lock);
 
-  DirOpenResult OpenImpl(const FilePath& file_path, const std::string& name,
-                         DirectoryChangeDelegate* delegate);
+  DirOpenResult OpenImpl(
+      const FilePath& file_path, const std::string& name,
+      DirectoryChangeDelegate* delegate,
+      const browser_sync::WeakHandle<TransactionObserver>&
+          transaction_observer);
 
   template <class T> void TestAndSet(T* kernel_data, const T* data_to_set);
 
@@ -1049,14 +1049,22 @@ class Directory {
   typedef Index<ClientTagIndexer>::Set ClientTagIndex;
 
  protected:
-  // Used by tests.
-  void InitKernel(const std::string& name, DirectoryChangeDelegate* delegate);
+  // Used by tests. |delegate| must not be NULL.
+  // |transaction_observer| must be initialized.
+  void InitKernelForTest(
+      const std::string& name,
+      DirectoryChangeDelegate* delegate,
+      const browser_sync::WeakHandle<TransactionObserver>&
+          transaction_observer);
 
  private:
   struct Kernel {
-    // |delegate| can be NULL.
+    // |delegate| must not be NULL.  |transaction_observer| must be
+    // initialized.
     Kernel(const FilePath& db_path, const std::string& name,
-           const KernelLoadInfo& info, DirectoryChangeDelegate* delegate);
+           const KernelLoadInfo& info, DirectoryChangeDelegate* delegate,
+           const browser_sync::WeakHandle<TransactionObserver>&
+               transaction_observer);
 
     ~Kernel();
 
@@ -1125,11 +1133,11 @@ class Directory {
     // The next metahandle is protected by kernel mutex.
     int64 next_metahandle;
 
-    // The delegate for directory change events.  Can be NULL.
+    // The delegate for directory change events.  Must not be NULL.
     DirectoryChangeDelegate* const delegate;
 
-    // The transaction observers.
-    scoped_refptr<ObserverListThreadSafe<TransactionObserver> > observers;
+    // The transaction observer.
+    const browser_sync::WeakHandle<TransactionObserver> transaction_observer;
   };
 
   // Helper method used to do searches on |parent_id_child_index|.

chrome/browser/sync/syncable/syncable_mock.cc

@@ -5,9 +5,10 @@
 #include "chrome/browser/sync/syncable/syncable_mock.h"
 
 #include "base/location.h"
+#include "chrome/browser/sync/test/null_transaction_observer.h"
 
 MockDirectory::MockDirectory() {
-  InitKernel("myk", &delegate_);
+  InitKernelForTest("myk", &delegate_, syncable::NullTransactionObserver());
 }
 
 MockDirectory::~MockDirectory() {}

chrome/browser/sync/syncable/syncable_unittest.cc

@@ -4,28 +4,15 @@
 
 #include "chrome/browser/sync/syncable/syncable.h"
 
-#include "build/build_config.h"
-
-#include <sys/types.h>
-
-#include <limits>
 #include <string>
 
-#if !defined(OS_WIN)
-#define MAX_PATH PATH_MAX
-#include <ostream>
-#include <stdio.h>
-#include <sys/ipc.h>
-#include <sys/sem.h>
-#include <sys/times.h>
-#endif  // !defined(OS_WIN)
-
 #include "base/compiler_specific.h"
 #include "base/file_path.h"
 #include "base/file_util.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/message_loop.h"
 #include "base/scoped_temp_dir.h"
 #include "base/stringprintf.h"
 #include "base/synchronization/condition_variable.h"
@@ -39,6 +26,7 @@
 #include "chrome/browser/sync/test/engine/test_id_factory.h"
 #include "chrome/browser/sync/test/engine/test_syncable_utils.h"
 #include "chrome/browser/sync/test/null_directory_change_delegate.h"
+#include "chrome/browser/sync/test/null_transaction_observer.h"
 #include "chrome/test/base/values_test_util.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "third_party/sqlite/sqlite3.h"
@@ -103,6 +91,7 @@ class SyncableGeneralTest : public testing::Test {
   virtual void TearDown() {
   }
  protected:
+  MessageLoop message_loop_;
   ScopedTempDir temp_dir_;
   NullDirectoryChangeDelegate delegate_;
   FilePath db_path_;
@@ -110,7 +99,7 @@ class SyncableGeneralTest : public testing::Test {
 
 TEST_F(SyncableGeneralTest, General) {
   Directory dir;
-  dir.Open(db_path_, "SimpleTest", &delegate_);
+  dir.Open(db_path_, "SimpleTest", &delegate_, NullTransactionObserver());
 
   int64 root_metahandle;
   {
@@ -209,7 +198,7 @@ TEST_F(SyncableGeneralTest, General) {
 
 TEST_F(SyncableGeneralTest, ChildrenOps) {
   Directory dir;
-  dir.Open(db_path_, "SimpleTest", &delegate_);
+  dir.Open(db_path_, "SimpleTest", &delegate_, NullTransactionObserver());
 
   int64 root_metahandle;
   {
@@ -289,7 +278,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsProperly) {
   // Test creating a new meta entry.
   {
     Directory dir;
-    dir.Open(db_path_, "IndexTest", &delegate_);
+    dir.Open(db_path_, "IndexTest", &delegate_, NullTransactionObserver());
     {
       WriteTransaction wtrans(FROM_HERE, UNITTEST, &dir);
       MutableEntry me(&wtrans, CREATE, wtrans.root_id(), name);
@@ -305,7 +294,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsProperly) {
   // The DB was closed. Now reopen it. This will cause index regeneration.
   {
     Directory dir;
-    dir.Open(db_path_, "IndexTest", &delegate_);
+    dir.Open(db_path_, "IndexTest", &delegate_, NullTransactionObserver());
 
     ReadTransaction trans(FROM_HERE, &dir);
     Entry me(&trans, GET_BY_CLIENT_TAG, tag);
@@ -325,7 +314,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsDeletedProperly) {
   // Test creating a deleted, unsynced, server meta entry.
   {
     Directory dir;
-    dir.Open(db_path_, "IndexTest", &delegate_);
+    dir.Open(db_path_, "IndexTest", &delegate_, NullTransactionObserver());
     {
       WriteTransaction wtrans(FROM_HERE, UNITTEST, &dir);
       MutableEntry me(&wtrans, CREATE, wtrans.root_id(), "deleted");
@@ -343,7 +332,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsDeletedProperly) {
   // Should still be present and valid in the client tag index.
   {
     Directory dir;
-    dir.Open(db_path_, "IndexTest", &delegate_);
+    dir.Open(db_path_, "IndexTest", &delegate_, NullTransactionObserver());
 
     ReadTransaction trans(FROM_HERE, &dir);
     Entry me(&trans, GET_BY_CLIENT_TAG, tag);
@@ -357,7 +346,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsDeletedProperly) {
 
 TEST_F(SyncableGeneralTest, ToValue) {
   Directory dir;
-  dir.Open(db_path_, "SimpleTest", &delegate_);
+  dir.Open(db_path_, "SimpleTest", &delegate_, NullTransactionObserver());
 
   const Id id = TestIdFactory::FromNumber(99);
   {
@@ -412,6 +401,7 @@ class TestUnsaveableDirectory : public Directory {
 // Test suite for syncable::Directory.
 class SyncableDirectoryTest : public testing::Test {
  protected:
+  MessageLoop message_loop_;
   ScopedTempDir temp_dir_;
   static const char kName[];
   static const Id kId;
@@ -425,7 +415,8 @@ class SyncableDirectoryTest : public testing::Test {
     file_util::Delete(file_path_, true);
     dir_.reset(new Directory());
     ASSERT_TRUE(dir_.get());
-    ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+    ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                                 &delegate_, NullTransactionObserver()));
     ASSERT_TRUE(dir_->good());
   }
 
@@ -439,7 +430,8 @@ class SyncableDirectoryTest : public testing::Test {
   void ReloadDir() {
     dir_.reset(new Directory());
     ASSERT_TRUE(dir_.get());
-    ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+    ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                                 &delegate_, NullTransactionObserver()));
   }
 
   void SaveAndReloadDir() {
@@ -1202,7 +1194,8 @@ TEST_F(SyncableDirectoryTest, TestSimpleFieldsPreservedDuringSaveChanges) {
   dir_->SaveChanges();
   dir_.reset(new Directory());
   ASSERT_TRUE(dir_.get());
-  ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+  ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                               &delegate_, NullTransactionObserver()));
   ASSERT_TRUE(dir_->good());
 
   {
@@ -1301,7 +1294,8 @@ TEST_F(SyncableDirectoryTest, TestSaveChangesFailure) {
   // always fail.
   dir_.reset(new TestUnsaveableDirectory());
   ASSERT_TRUE(dir_.get());
-  ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+  ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                               &delegate_, NullTransactionObserver()));
   ASSERT_TRUE(dir_->good());
   int64 handle2 = 0;
   {
@@ -1374,7 +1368,8 @@ TEST_F(SyncableDirectoryTest, TestSaveChangesFailureWithPurge) {
   // always fail.
   dir_.reset(new TestUnsaveableDirectory());
   ASSERT_TRUE(dir_.get());
-  ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+  ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                               &delegate_, NullTransactionObserver()));
   ASSERT_TRUE(dir_->good());
 
   ModelTypeSet set;
@@ -1496,13 +1491,14 @@ class SyncableDirectoryManager : public testing::Test {
   virtual void TearDown() {
   }
  protected:
+  MessageLoop message_loop_;
   ScopedTempDir temp_dir_;
   NullDirectoryChangeDelegate delegate_;
 };
 
 TEST_F(SyncableDirectoryManager, TestFileRelease) {
   DirectoryManager dm(FilePath(temp_dir_.path()));
-  ASSERT_TRUE(dm.Open("ScopeTest", &delegate_));
+  ASSERT_TRUE(dm.Open("ScopeTest", &delegate_, NullTransactionObserver()));
   {
     ScopedDirLookup(&dm, "ScopeTest");
   }
@@ -1520,7 +1516,8 @@ class ThreadOpenTestDelegate : public base::PlatformThread::Delegate {
  private:
   // PlatformThread::Delegate methods:
   virtual void ThreadMain() {
-    CHECK(directory_manager_->Open("Open", &delegate_));
+    CHECK(directory_manager_->Open(
+        "Open", &delegate_, NullTransactionObserver()));
   }
 
   DISALLOW_COPY_AND_ASSIGN(ThreadOpenTestDelegate);
@@ -1564,6 +1561,7 @@ class ThreadBugDelegate : public base::PlatformThread::Delegate {
 
   // PlatformThread::Delegate methods:
   virtual void ThreadMain() {
+    MessageLoop message_loop;
     const std::string dirname = "ThreadBug1";
     base::AutoLock scoped_lock(step_->mutex);
 
@@ -1573,12 +1571,14 @@ class ThreadBugDelegate : public base::PlatformThread::Delegate {
       }
       switch (step_->number) {
       case 0:
-        directory_manager_->Open(dirname, &delegate_);
+        directory_manager_->Open(
+            dirname, &delegate_, NullTransactionObserver());
         break;
       case 1:
         {
           directory_manager_->Close(dirname);
-          directory_manager_->Open(dirname, &delegate_);
+          directory_manager_->Open(
+              dirname, &delegate_, NullTransactionObserver());
           ScopedDirLookup dir(directory_manager_, dirname);
           CHECK(dir.good());
           WriteTransaction trans(FROM_HERE, UNITTEST, dir);
@@ -1636,6 +1636,7 @@ class DirectoryKernelStalenessBugDelegate : public ThreadBugDelegate {
     : ThreadBugDelegate(role, step, dirman) {}
 
   virtual void ThreadMain() {
+    MessageLoop message_loop;
     const char test_bytes[] = "test data";
     const std::string dirname = "DirectoryKernelStalenessBug";
     base::AutoLock scoped_lock(step_->mutex);
@@ -1652,7 +1653,8 @@ class DirectoryKernelStalenessBugDelegate : public ThreadBugDelegate {
           file_util::Delete(directory_manager_->GetSyncDataDatabasePath(),
                             true);
           // Test.
-          directory_manager_->Open(dirname, &delegate_);
+          directory_manager_->Open(
+              dirname, &delegate_, NullTransactionObserver());
           ScopedDirLookup dir(directory_manager_, dirname);
           CHECK(dir.good());
           WriteTransaction trans(FROM_HERE, UNITTEST, dir);
@@ -1671,7 +1673,8 @@ class DirectoryKernelStalenessBugDelegate : public ThreadBugDelegate {
         break;
       case 1:
         {
-          directory_manager_->Open(dirname, &delegate_);
+          directory_manager_->Open(
+              dirname, &delegate_, NullTransactionObserver());
           ScopedDirLookup dir(directory_manager_, dirname);
           CHECK(dir.good());
         }
@@ -1770,13 +1773,14 @@ class StressTransactionsDelegate : public base::PlatformThread::Delegate {
 };
 
 TEST(SyncableDirectory, StressTransactions) {
+  MessageLoop message_loop;
   ScopedTempDir temp_dir;
   ASSERT_TRUE(temp_dir.CreateUniqueTempDir());
   DirectoryManager dirman(FilePath(temp_dir.path()));
   std::string dirname = "stress";
   file_util::Delete(dirman.GetSyncDataDatabasePath(), true);
   NullDirectoryChangeDelegate delegate;
-  dirman.Open(dirname, &delegate);
+  dirman.Open(dirname, &delegate, NullTransactionObserver());
 
   const int kThreadCount = 7;
   base::PlatformThreadHandle threads[kThreadCount];

chrome/browser/sync/test/engine/syncer_command_test.h

@@ -11,6 +11,7 @@
 #include <vector>
 
 #include "base/compiler_specific.h"
+#include "base/message_loop.h"
 #include "chrome/browser/sync/engine/model_safe_worker.h"
 #include "chrome/browser/sync/sessions/debug_info_getter.h"
 #include "chrome/browser/sync/sessions/sync_session.h"
@@ -157,6 +158,7 @@ class SyncerCommandTestWithParam : public testing::TestWithParam<T>,
   }
 
  private:
+  MessageLoop message_loop_;
   scoped_ptr<TestDirectorySetterUpper> syncdb_;
   scoped_ptr<sessions::SyncSessionContext> context_;
   scoped_ptr<MockConnectionManager> mock_server_;

chrome/browser/sync/test/engine/test_directory_setter_upper.cc

@@ -10,9 +10,11 @@
 #include "base/string_util.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
 #include "chrome/browser/sync/syncable/syncable.h"
+#include "chrome/browser/sync/test/null_transaction_observer.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 using syncable::DirectoryManager;
+using syncable::NullTransactionObserver;
 using syncable::ReadTransaction;
 using syncable::ScopedDirLookup;
 
@@ -38,7 +40,7 @@ void TestDirectorySetterUpper::reset_directory_manager(DirectoryManager* d) {
 
 void TestDirectorySetterUpper::SetUp() {
   Init();
-  ASSERT_TRUE(manager()->Open(name(), &delegate_));
+  ASSERT_TRUE(manager()->Open(name(), &delegate_, NullTransactionObserver()));
 }
 
 void TestDirectorySetterUpper::TearDown() {
@@ -80,7 +82,8 @@ void ManuallyOpenedTestDirectorySetterUpper::SetUp() {
 }
 
 void ManuallyOpenedTestDirectorySetterUpper::Open() {
-  ASSERT_TRUE(manager()->Open(name(), &delegate_));
+  ASSERT_TRUE(
+      manager()->Open(name(), &delegate_, NullTransactionObserver()));
   was_opened_ = true;
 }
 
@@ -111,7 +114,7 @@ void TriggeredOpenTestDirectorySetterUpper::TearDown() {
 
 MockDirectorySetterUpper::MockDirectory::MockDirectory(
     const std::string& name) {
-  InitKernel(name, &delegate_);
+  InitKernelForTest(name, &delegate_, NullTransactionObserver());
 }
 
 MockDirectorySetterUpper::MockDirectory::~MockDirectory() {}

chrome/browser/sync/test/null_transaction_observer.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/test/null_transaction_observer.h"
+
+#include "base/memory/weak_ptr.h"
+
+namespace syncable {
+
+browser_sync::WeakHandle<TransactionObserver> NullTransactionObserver() {
+  return browser_sync::MakeWeakHandle(base::WeakPtr<TransactionObserver>());
+}
+
+}  // namespace syncable

chrome/browser/sync/test/null_transaction_observer.h

+// 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.
+
+#ifndef CHROME_BROWSER_SYNC_TEST_NULL_TRANSACTION_OBSERVER_H_
+#define CHROME_BROWSER_SYNC_TEST_NULL_TRANSACTION_OBSERVER_H_
+#pragma once
+
+#include "chrome/browser/sync/util/weak_handle.h"
+
+namespace syncable {
+
+class TransactionObserver;
+
+// Returns an initialized weak handle to a transaction observer that
+// does nothing.
+browser_sync::WeakHandle<TransactionObserver> NullTransactionObserver();
+
+}  // namespace syncable
+
+#endif  // CHROME_BROWSER_SYNC_TEST_NULL_TRANSACTION_OBSERVER_H_

chrome/chrome_tests.gypi

@@ -356,6 +356,8 @@
         'browser/sync/js/js_test_util.h',
         'browser/sync/test/null_directory_change_delegate.cc',
         'browser/sync/test/null_directory_change_delegate.h',
+        'browser/sync/test/null_transaction_observer.cc',
+        'browser/sync/test/null_transaction_observer.h',
         'browser/sync/test/engine/test_directory_setter_upper.cc',
         'browser/sync/test/engine/test_directory_setter_upper.h',
       ],