[SyncFS] Construct and destruct sync_worker in worker_task_runner

chrome/browser/sync_file_system/drive_backend/sync_engine.cc

 // Copyright 2013 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_file_system/drive_backend/sync_engine.h"
 
 #include <vector>
 
 #include "base/bind.h"
 #include "base/metrics/histogram.h"
+#include "base/run_loop.h"
 #include "base/threading/sequenced_worker_pool.h"
 #include "base/time/time.h"
 #include "base/values.h"
 #include "chrome/browser/drive/drive_api_service.h"
 #include "chrome/browser/drive/drive_notification_manager.h"
 #include "chrome/browser/drive/drive_notification_manager_factory.h"
 #include "chrome/browser/drive/drive_service_interface.h"
 #include "chrome/browser/drive/drive_uploader.h"
 #include "chrome/browser/extensions/extension_service.h"
 #include "chrome/browser/profiles/profile.h"
@@ skipping 38 matching lines @@
 class RemoteChangeProcessor;
 
 namespace drive_backend {
 
 class SyncEngine::WorkerObserver : public SyncWorker::Observer {
  public:
   WorkerObserver(base::SequencedTaskRunner* ui_task_runner,
                  base::WeakPtr<SyncEngine> sync_engine)
       : ui_task_runner_(ui_task_runner),
         sync_engine_(sync_engine) {}
-
   virtual ~WorkerObserver() {}
 
   virtual void OnPendingFileListUpdated(int item_count) OVERRIDE {
     if (ui_task_runner_->RunsTasksOnCurrentThread()) {
       sync_engine_->OnPendingFileListUpdated(item_count);
       return;
     }
 
     ui_task_runner_->PostTask(
         FROM_HERE,
@@ skipping 96 matching lines @@
   scoped_refptr<base::SingleThreadTaskRunner>
       worker_task_runner = base::MessageLoopProxy::current();
 
   scoped_ptr<drive_backend::SyncEngine> sync_engine(
       new SyncEngine(drive_service.Pass(),
                      drive_uploader.Pass(),
                      worker_task_runner,
                      notification_manager,
                      extension_service,
                      signin_manager));
+  base::RunLoop run_loop;
   sync_engine->Initialize(GetSyncFileSystemDir(context->GetPath()),
                           file_task_runner.get(),
+                          run_loop.QuitClosure(),
                           NULL);
+  run_loop.Run();

[nhiroki, 2014/05/26 05:31:32]

Hmm... this can block the UI thread?

 
   return sync_engine.Pass();
 }
 
 void SyncEngine::AppendDependsOnFactories(
     std::set<BrowserContextKeyedServiceFactory*>* factories) {
   DCHECK(factories);
   factories->insert(drive::DriveNotificationManagerFactory::GetInstance());
   factories->insert(SigninManagerFactory::GetInstance());
   factories->insert(
       extensions::ExtensionsBrowserClient::Get()->GetExtensionSystemFactory());
 }
 
 SyncEngine::~SyncEngine() {
   net::NetworkChangeNotifier::RemoveNetworkChangeObserver(this);
   GetDriveService()->RemoveObserver(this);
   if (notification_manager_)
     notification_manager_->RemoveObserver(this);
 
-  // TODO(tzik): Destroy |sync_worker_| and |worker_observer_| on the worker.
+  worker_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(
+          &SyncWorker::DestructOnWorker,
+          base::Passed(&sync_worker_),
+          base::Passed(worker_observer_.PassAs<SyncWorker::Observer>())));
 }
 
 void SyncEngine::Initialize(const base::FilePath& base_dir,
                             base::SequencedTaskRunner* file_task_runner,
+                            const base::Closure& closure,
                             leveldb::Env* env_override) {
   // DriveServiceWrapper and DriveServiceOnWorker relay communications
   // between DriveService and syncers in SyncWorker.
   scoped_ptr<drive::DriveServiceInterface>
       drive_service_on_worker(
           new DriveServiceOnWorker(drive_service_wrapper_->AsWeakPtr(),
                                    base::MessageLoopProxy::current(),
                                    worker_task_runner_));
   scoped_ptr<drive::DriveUploaderInterface>
       drive_uploader_on_worker(
           new DriveUploaderOnWorker(drive_uploader_wrapper_->AsWeakPtr(),
                                     base::MessageLoopProxy::current(),
                                     worker_task_runner_));
   scoped_ptr<SyncEngineContext>
       sync_engine_context(
           new SyncEngineContext(drive_service_on_worker.Pass(),
                                 drive_uploader_on_worker.Pass(),
                                 base::MessageLoopProxy::current(),
                                 worker_task_runner_,
                                 file_task_runner));
 
   worker_observer_.reset(
       new WorkerObserver(base::MessageLoopProxy::current(),
                          weak_ptr_factory_.GetWeakPtr()));
 
   base::WeakPtr<ExtensionServiceInterface> extension_service_weak_ptr;
   if (extension_service_)
     extension_service_weak_ptr = extension_service_->AsWeakPtr();
 
-  // TODO(peria): Use PostTask on |worker_task_runner_| to call this function.
-  sync_worker_ = SyncWorker::CreateOnWorker(
-      base_dir,
-      worker_observer_.get(),
-      extension_service_weak_ptr,
-      sync_engine_context.Pass(),
-      env_override);
-
   if (notification_manager_)
     notification_manager_->AddObserver(this);
   GetDriveService()->AddObserver(this);
   net::NetworkChangeNotifier::AddNetworkChangeObserver(this);

[nhiroki, 2014/05/26 05:22:44]

I haven't read sourcecode closely, but I guess these observers should be added
after a worker is created.

[peria, 2014/05/26 05:32:35]

Done.
Yes, those observers communicate with SyncWorker through SyncEngine!

+
+  PostTaskAndReplyWithResult(
+      worker_task_runner_,
+      FROM_HERE,
+      base::Bind(&SyncWorker::CreateOnWorker,
+                 base_dir,
+                 worker_observer_.get(),
+                 extension_service_weak_ptr,
+                 base::Passed(sync_engine_context.Pass()),
+                 env_override),
+      base::Bind(&SyncEngine::SetSyncWorker,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 closure));
 }
 
 void SyncEngine::AddServiceObserver(SyncServiceObserver* observer) {
   service_observers_.AddObserver(observer);
 }
 
 void SyncEngine::AddFileStatusObserver(FileStatusObserver* observer) {
   file_status_observers_.AddObserver(observer);
 }
 
@@ skipping 274 matching lines @@
     bool is_app_enabled = extension_service_->IsExtensionEnabled(app_id);
     bool is_app_root_tracker_enabled =
         tracker.tracker_kind() == TRACKER_KIND_APP_ROOT;
     if (is_app_enabled && !is_app_root_tracker_enabled)
       EnableOrigin(origin, base::Bind(&EmptyStatusCallback));
     else if (!is_app_enabled && is_app_root_tracker_enabled)
       DisableOrigin(origin, base::Bind(&EmptyStatusCallback));
   }
 }
 
+void SyncEngine::SetSyncWorker(const base::Closure& closure,
+                               scoped_ptr<SyncWorker> sync_worker) {
+  sync_worker_ = sync_worker.Pass();
+  closure.Run();
+}
+
 }  // namespace drive_backend
 }  // namespace sync_file_system