chrome/browser/sync_file_system/drive_backend/sync_engine.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "chrome/browser/sync_file_system/drive_backend/sync_engine.h"
   6
   7 #include <vector>
   8
   9 #include "base/bind.h"
  10 #include "base/metrics/histogram.h"
  11 #include "base/thread_task_runner_handle.h"
  12 #include "base/threading/sequenced_worker_pool.h"
  13 #include "base/time/time.h"
  14 #include "base/values.h"
  15 #include "chrome/browser/drive/drive_notification_manager_factory.h"
  16 #include "chrome/browser/extensions/extension_service.h"
  17 #include "chrome/browser/profiles/profile.h"
  18 #include "chrome/browser/signin/profile_oauth2_token_service_factory.h"
  19 #include "chrome/browser/signin/signin_manager_factory.h"
  20 #include "chrome/browser/sync_file_system/drive_backend/callback_helper.h"
  21 #include "chrome/browser/sync_file_system/drive_backend/conflict_resolver.h"
  22 #include "chrome/browser/sync_file_system/drive_backend/drive_backend_constants.h"
  23 #include "chrome/browser/sync_file_system/drive_backend/drive_service_on_worker.h"
  24 #include "chrome/browser/sync_file_system/drive_backend/drive_service_wrapper.h"
  25 #include "chrome/browser/sync_file_system/drive_backend/drive_uploader_on_worker.h"
  26 #include "chrome/browser/sync_file_system/drive_backend/drive_uploader_wrapper.h"
  27 #include "chrome/browser/sync_file_system/drive_backend/list_changes_task.h"
  28 #include "chrome/browser/sync_file_system/drive_backend/local_to_remote_syncer.h"
  29 #include "chrome/browser/sync_file_system/drive_backend/metadata_database.h"
  30 #include "chrome/browser/sync_file_system/drive_backend/register_app_task.h"
  31 #include "chrome/browser/sync_file_system/drive_backend/remote_change_processor_on_worker.h"
  32 #include "chrome/browser/sync_file_system/drive_backend/remote_change_processor_wrapper.h"
  33 #include "chrome/browser/sync_file_system/drive_backend/remote_to_local_syncer.h"
  34 #include "chrome/browser/sync_file_system/drive_backend/sync_engine_context.h"
  35 #include "chrome/browser/sync_file_system/drive_backend/sync_engine_initializer.h"
  36 #include "chrome/browser/sync_file_system/drive_backend/sync_task.h"
  37 #include "chrome/browser/sync_file_system/drive_backend/sync_worker.h"
  38 #include "chrome/browser/sync_file_system/drive_backend/sync_worker_interface.h"
  39 #include "chrome/browser/sync_file_system/drive_backend/uninstall_app_task.h"
  40 #include "chrome/browser/sync_file_system/file_status_observer.h"
  41 #include "chrome/browser/sync_file_system/logger.h"
  42 #include "chrome/browser/sync_file_system/syncable_file_system_util.h"
  43 #include "components/drive/drive_notification_manager.h"
  44 #include "components/drive/drive_uploader.h"
  45 #include "components/drive/service/drive_api_service.h"
  46 #include "components/drive/service/drive_service_interface.h"
  47 #include "components/signin/core/browser/profile_oauth2_token_service.h"
  48 #include "components/signin/core/browser/signin_manager.h"
  49 #include "content/public/browser/browser_thread.h"
  50 #include "extensions/browser/extension_system.h"
  51 #include "extensions/browser/extension_system_provider.h"
  52 #include "extensions/browser/extensions_browser_client.h"
  53 #include "extensions/common/extension.h"
  54 #include "google_apis/drive/drive_api_url_generator.h"
  55 #include "net/url_request/url_request_context_getter.h"
  56 #include "storage/browser/blob/scoped_file.h"
  57 #include "storage/common/fileapi/file_system_util.h"
  58
  59 namespace sync_file_system {
  60
  61 class RemoteChangeProcessor;
  62
  63 namespace drive_backend {
  64
  65 scoped_ptr<drive::DriveServiceInterface>
  66 SyncEngine::DriveServiceFactory::CreateDriveService(
  67     OAuth2TokenService* oauth2_token_service,
  68     net::URLRequestContextGetter* url_request_context_getter,
  69     base::SequencedTaskRunner* blocking_task_runner) {
  70   return scoped_ptr<drive::DriveServiceInterface>(
  71       new drive::DriveAPIService(
  72           oauth2_token_service,
  73           url_request_context_getter,
  74           blocking_task_runner,
  75           GURL(google_apis::DriveApiUrlGenerator::kBaseUrlForProduction),
  76           GURL(google_apis::DriveApiUrlGenerator::
  77                kBaseDownloadUrlForProduction),
  78           std::string() /* custom_user_agent */));
  79 }
  80
  81 class SyncEngine::WorkerObserver : public SyncWorkerInterface::Observer {
  82  public:
  83   WorkerObserver(base::SequencedTaskRunner* ui_task_runner,
  84                  base::WeakPtr<SyncEngine> sync_engine)
  85       : ui_task_runner_(ui_task_runner),
  86         sync_engine_(sync_engine) {
  87     sequence_checker_.DetachFromSequence();
  88   }
  89
  90   ~WorkerObserver() override {
  91     DCHECK(sequence_checker_.CalledOnValidSequencedThread());
  92   }
  93
  94   void OnPendingFileListUpdated(int item_count) override {
  95     if (ui_task_runner_->RunsTasksOnCurrentThread()) {
  96       if (sync_engine_)
  97         sync_engine_->OnPendingFileListUpdated(item_count);
  98       return;
  99     }
 100
 101     DCHECK(sequence_checker_.CalledOnValidSequencedThread());
 102     ui_task_runner_->PostTask(
 103         FROM_HERE,
 104         base::Bind(&SyncEngine::OnPendingFileListUpdated,
 105                    sync_engine_,
 106                    item_count));
 107   }
 108
 109   void OnFileStatusChanged(const storage::FileSystemURL& url,
 110                            SyncFileType file_type,
 111                            SyncFileStatus file_status,
 112                            SyncAction sync_action,
 113                            SyncDirection direction) override {
 114     if (ui_task_runner_->RunsTasksOnCurrentThread()) {
 115       if (sync_engine_)
 116         sync_engine_->OnFileStatusChanged(
 117             url, file_type, file_status, sync_action, direction);
 118       return;
 119     }
 120
 121     DCHECK(sequence_checker_.CalledOnValidSequencedThread());
 122     ui_task_runner_->PostTask(
 123         FROM_HERE,
 124         base::Bind(&SyncEngine::OnFileStatusChanged,
 125                    sync_engine_,
 126                    url, file_type, file_status, sync_action, direction));
 127   }
 128
 129   void UpdateServiceState(RemoteServiceState state,
 130                           const std::string& description) override {
 131     if (ui_task_runner_->RunsTasksOnCurrentThread()) {
 132       if (sync_engine_)
 133         sync_engine_->UpdateServiceState(state, description);
 134       return;
 135     }
 136
 137     DCHECK(sequence_checker_.CalledOnValidSequencedThread());
 138     ui_task_runner_->PostTask(
 139         FROM_HERE,
 140         base::Bind(&SyncEngine::UpdateServiceState,
 141                    sync_engine_, state, description));
 142   }
 143
 144   void DetachFromSequence() {
 145     sequence_checker_.DetachFromSequence();
 146   }
 147
 148  private:
 149   scoped_refptr<base::SequencedTaskRunner> ui_task_runner_;
 150   base::WeakPtr<SyncEngine> sync_engine_;
 151
 152   base::SequenceChecker sequence_checker_;
 153
 154   DISALLOW_COPY_AND_ASSIGN(WorkerObserver);
 155 };
 156
 157 namespace {
 158
 159 void DidRegisterOrigin(const base::TimeTicks& start_time,
 160                        const SyncStatusCallback& callback,
 161                        SyncStatusCode status) {
 162   base::TimeDelta delta(base::TimeTicks::Now() - start_time);
 163   LOCAL_HISTOGRAM_TIMES("SyncFileSystem.RegisterOriginTime", delta);
 164   callback.Run(status);
 165 }
 166
 167 }  // namespace
 168
 169 scoped_ptr<SyncEngine> SyncEngine::CreateForBrowserContext(
 170     content::BrowserContext* context,
 171     TaskLogger* task_logger) {
 172   scoped_refptr<base::SequencedWorkerPool> worker_pool =
 173       content::BrowserThread::GetBlockingPool();
 174
 175   scoped_refptr<base::SingleThreadTaskRunner> ui_task_runner =
 176       base::ThreadTaskRunnerHandle::Get();
 177   scoped_refptr<base::SequencedTaskRunner> worker_task_runner =
 178       worker_pool->GetSequencedTaskRunnerWithShutdownBehavior(
 179           worker_pool->GetSequenceToken(),
 180           base::SequencedWorkerPool::SKIP_ON_SHUTDOWN);
 181   scoped_refptr<base::SequencedTaskRunner> drive_task_runner =
 182       worker_pool->GetSequencedTaskRunnerWithShutdownBehavior(
 183           worker_pool->GetSequenceToken(),
 184           base::SequencedWorkerPool::SKIP_ON_SHUTDOWN);
 185
 186   Profile* profile = Profile::FromBrowserContext(context);
 187   drive::DriveNotificationManager* notification_manager =
 188       drive::DriveNotificationManagerFactory::GetForBrowserContext(context);
 189   ExtensionService* extension_service =
 190       extensions::ExtensionSystem::Get(context)->extension_service();
 191   SigninManagerBase* signin_manager =
 192       SigninManagerFactory::GetForProfile(profile);
 193   OAuth2TokenService* token_service =
 194       ProfileOAuth2TokenServiceFactory::GetForProfile(profile);
 195   scoped_refptr<net::URLRequestContextGetter> request_context =
 196       context->GetRequestContext();
 197
 198   scoped_ptr<drive_backend::SyncEngine> sync_engine(
 199       new SyncEngine(ui_task_runner.get(),
 200                      worker_task_runner.get(),
 201                      drive_task_runner.get(),
 202                      worker_pool.get(),
 203                      GetSyncFileSystemDir(context->GetPath()),
 204                      task_logger,
 205                      notification_manager,
 206                      extension_service,
 207                      signin_manager,
 208                      token_service,
 209                      request_context.get(),
 210                      make_scoped_ptr(new DriveServiceFactory()),
 211                      nullptr /* env_override */));
 212
 213   sync_engine->Initialize();
 214   return sync_engine.Pass();
 215 }
 216
 217 void SyncEngine::AppendDependsOnFactories(
 218     std::set<BrowserContextKeyedServiceFactory*>* factories) {
 219   DCHECK(factories);
 220   factories->insert(drive::DriveNotificationManagerFactory::GetInstance());
 221   factories->insert(SigninManagerFactory::GetInstance());
 222   factories->insert(
 223       extensions::ExtensionsBrowserClient::Get()->GetExtensionSystemFactory());
 224   factories->insert(ProfileOAuth2TokenServiceFactory::GetInstance());
 225 }
 226
 227 SyncEngine::~SyncEngine() {
 228   Reset();
 229
 230   net::NetworkChangeNotifier::RemoveNetworkChangeObserver(this);
 231   if (signin_manager_)
 232     signin_manager_->RemoveObserver(this);
 233   if (notification_manager_)
 234     notification_manager_->RemoveObserver(this);
 235 }
 236
 237 void SyncEngine::Reset() {
 238   if (drive_service_)
 239     drive_service_->RemoveObserver(this);
 240
 241   worker_task_runner_->DeleteSoon(FROM_HERE, sync_worker_.release());
 242   worker_task_runner_->DeleteSoon(FROM_HERE, worker_observer_.release());
 243   worker_task_runner_->DeleteSoon(FROM_HERE,
 244                                   remote_change_processor_on_worker_.release());
 245
 246   drive_service_wrapper_.reset();
 247   drive_service_.reset();
 248   drive_uploader_wrapper_.reset();
 249   drive_uploader_.reset();
 250   remote_change_processor_wrapper_.reset();
 251   callback_tracker_.AbortAll();
 252 }
 253
 254 void SyncEngine::Initialize() {
 255   Reset();
 256
 257   if (!signin_manager_ || !signin_manager_->IsAuthenticated())
 258     return;
 259
 260   DCHECK(drive_service_factory_);
 261   scoped_ptr<drive::DriveServiceInterface> drive_service =
 262       drive_service_factory_->CreateDriveService(
 263           token_service_, request_context_.get(), drive_task_runner_.get());
 264   scoped_ptr<drive::DriveUploaderInterface> drive_uploader(
 265       new drive::DriveUploader(drive_service.get(), drive_task_runner_.get()));
 266
 267   InitializeInternal(drive_service.Pass(), drive_uploader.Pass(), nullptr);
 268 }
 269
 270 void SyncEngine::InitializeForTesting(
 271     scoped_ptr<drive::DriveServiceInterface> drive_service,
 272     scoped_ptr<drive::DriveUploaderInterface> drive_uploader,
 273     scoped_ptr<SyncWorkerInterface> sync_worker) {
 274   Reset();
 275   InitializeInternal(drive_service.Pass(), drive_uploader.Pass(),
 276                      sync_worker.Pass());
 277 }
 278
 279 void SyncEngine::InitializeInternal(
 280     scoped_ptr<drive::DriveServiceInterface> drive_service,
 281     scoped_ptr<drive::DriveUploaderInterface> drive_uploader,
 282     scoped_ptr<SyncWorkerInterface> sync_worker) {
 283   drive_service_ = drive_service.Pass();
 284   drive_service_wrapper_.reset(new DriveServiceWrapper(drive_service_.get()));
 285
 286   std::string account_id;
 287   if (signin_manager_)
 288     account_id = signin_manager_->GetAuthenticatedAccountId();
 289   drive_service_->Initialize(account_id);
 290
 291   drive_uploader_ = drive_uploader.Pass();
 292   drive_uploader_wrapper_.reset(
 293       new DriveUploaderWrapper(drive_uploader_.get()));
 294
 295   // DriveServiceWrapper and DriveServiceOnWorker relay communications
 296   // between DriveService and syncers in SyncWorker.
 297   scoped_ptr<drive::DriveServiceInterface> drive_service_on_worker(
 298       new DriveServiceOnWorker(drive_service_wrapper_->AsWeakPtr(),
 299                                ui_task_runner_.get(),
 300                                worker_task_runner_.get()));
 301   scoped_ptr<drive::DriveUploaderInterface> drive_uploader_on_worker(
 302       new DriveUploaderOnWorker(drive_uploader_wrapper_->AsWeakPtr(),
 303                                 ui_task_runner_.get(),
 304                                 worker_task_runner_.get()));
 305   scoped_ptr<SyncEngineContext> sync_engine_context(
 306       new SyncEngineContext(drive_service_on_worker.Pass(),
 307                             drive_uploader_on_worker.Pass(),
 308                             task_logger_,
 309                             ui_task_runner_.get(),
 310                             worker_task_runner_.get(),
 311                             worker_pool_.get()));
 312
 313   worker_observer_.reset(new WorkerObserver(ui_task_runner_.get(),
 314                                             weak_ptr_factory_.GetWeakPtr()));
 315
 316   base::WeakPtr<ExtensionServiceInterface> extension_service_weak_ptr;
 317   if (extension_service_)
 318     extension_service_weak_ptr = extension_service_->AsWeakPtr();
 319
 320   if (!sync_worker) {
 321     sync_worker.reset(new SyncWorker(
 322         sync_file_system_dir_,
 323         extension_service_weak_ptr,
 324         env_override_));
 325   }
 326
 327   sync_worker_ = sync_worker.Pass();
 328   sync_worker_->AddObserver(worker_observer_.get());
 329
 330   worker_task_runner_->PostTask(
 331       FROM_HERE,
 332       base::Bind(&SyncWorkerInterface::Initialize,
 333                  base::Unretained(sync_worker_.get()),
 334                  base::Passed(&sync_engine_context)));
 335   if (remote_change_processor_)
 336     SetRemoteChangeProcessor(remote_change_processor_);
 337
 338   drive_service_->AddObserver(this);
 339
 340   service_state_ = REMOTE_SERVICE_TEMPORARY_UNAVAILABLE;
 341   OnNetworkChanged(net::NetworkChangeNotifier::GetConnectionType());
 342   if (drive_service_->HasRefreshToken())
 343     OnReadyToSendRequests();
 344   else
 345     OnRefreshTokenInvalid();
 346 }
 347
 348 void SyncEngine::AddServiceObserver(SyncServiceObserver* observer) {
 349   service_observers_.AddObserver(observer);
 350 }
 351
 352 void SyncEngine::AddFileStatusObserver(FileStatusObserver* observer) {
 353   file_status_observers_.AddObserver(observer);
 354 }
 355
 356 void SyncEngine::RegisterOrigin(const GURL& origin,
 357                                 const SyncStatusCallback& callback) {
 358   if (!sync_worker_) {
 359     // TODO(tzik): Record |origin| and retry the registration after late
 360     // sign-in.  Then, return SYNC_STATUS_OK.
 361     if (!signin_manager_ || !signin_manager_->IsAuthenticated())
 362       callback.Run(SYNC_STATUS_AUTHENTICATION_FAILED);
 363     else
 364       callback.Run(SYNC_STATUS_ABORT);
 365     return;
 366   }
 367
 368   SyncStatusCallback relayed_callback = RelayCallbackToCurrentThread(
 369       FROM_HERE, base::Bind(&DidRegisterOrigin, base::TimeTicks::Now(),
 370                             TrackCallback(callback)));
 371
 372   worker_task_runner_->PostTask(
 373       FROM_HERE,
 374       base::Bind(&SyncWorkerInterface::RegisterOrigin,
 375                  base::Unretained(sync_worker_.get()),
 376                  origin, relayed_callback));
 377 }
 378
 379 void SyncEngine::EnableOrigin(
 380     const GURL& origin, const SyncStatusCallback& callback) {
 381   if (!sync_worker_) {
 382     // It's safe to return OK immediately since this is also checked in
 383     // SyncWorker initialization.
 384     callback.Run(SYNC_STATUS_OK);
 385     return;
 386   }
 387
 388   SyncStatusCallback relayed_callback = RelayCallbackToCurrentThread(
 389       FROM_HERE, TrackCallback(callback));
 390
 391   worker_task_runner_->PostTask(
 392       FROM_HERE,
 393       base::Bind(&SyncWorkerInterface::EnableOrigin,
 394                  base::Unretained(sync_worker_.get()),
 395                  origin, relayed_callback));
 396 }
 397
 398 void SyncEngine::DisableOrigin(
 399     const GURL& origin, const SyncStatusCallback& callback) {
 400   if (!sync_worker_) {
 401     // It's safe to return OK immediately since this is also checked in
 402     // SyncWorker initialization.
 403     callback.Run(SYNC_STATUS_OK);
 404     return;
 405   }
 406
 407   SyncStatusCallback relayed_callback = RelayCallbackToCurrentThread(
 408       FROM_HERE, TrackCallback(callback));
 409
 410   worker_task_runner_->PostTask(
 411       FROM_HERE,
 412       base::Bind(&SyncWorkerInterface::DisableOrigin,
 413                  base::Unretained(sync_worker_.get()),
 414                  origin,
 415                  relayed_callback));
 416 }
 417
 418 void SyncEngine::UninstallOrigin(
 419     const GURL& origin,
 420     UninstallFlag flag,
 421     const SyncStatusCallback& callback) {
 422   if (!sync_worker_) {
 423     // It's safe to return OK immediately since this is also checked in
 424     // SyncWorker initialization.
 425     callback.Run(SYNC_STATUS_OK);
 426     return;
 427   }
 428
 429   SyncStatusCallback relayed_callback = RelayCallbackToCurrentThread(
 430       FROM_HERE, TrackCallback(callback));
 431   worker_task_runner_->PostTask(
 432       FROM_HERE,
 433       base::Bind(&SyncWorkerInterface::UninstallOrigin,
 434                  base::Unretained(sync_worker_.get()),
 435                  origin, flag, relayed_callback));
 436 }
 437
 438 void SyncEngine::ProcessRemoteChange(const SyncFileCallback& callback) {
 439   if (GetCurrentState() == REMOTE_SERVICE_DISABLED) {
 440     callback.Run(SYNC_STATUS_SYNC_DISABLED, storage::FileSystemURL());
 441     return;
 442   }
 443
 444   base::Closure abort_closure =
 445       base::Bind(callback, SYNC_STATUS_ABORT, storage::FileSystemURL());
 446
 447   if (!sync_worker_) {
 448     abort_closure.Run();
 449     return;
 450   }
 451
 452   SyncFileCallback tracked_callback = callback_tracker_.Register(
 453       abort_closure, callback);
 454   SyncFileCallback relayed_callback = RelayCallbackToCurrentThread(
 455       FROM_HERE, tracked_callback);
 456   worker_task_runner_->PostTask(
 457       FROM_HERE,
 458       base::Bind(&SyncWorkerInterface::ProcessRemoteChange,
 459                  base::Unretained(sync_worker_.get()),
 460                  relayed_callback));
 461 }
 462
 463 void SyncEngine::SetRemoteChangeProcessor(RemoteChangeProcessor* processor) {
 464   remote_change_processor_ = processor;
 465
 466   if (!sync_worker_)
 467     return;
 468
 469   remote_change_processor_wrapper_.reset(
 470       new RemoteChangeProcessorWrapper(processor));
 471
 472   remote_change_processor_on_worker_.reset(new RemoteChangeProcessorOnWorker(
 473       remote_change_processor_wrapper_->AsWeakPtr(),
 474       ui_task_runner_.get(),
 475       worker_task_runner_.get()));
 476
 477   worker_task_runner_->PostTask(
 478       FROM_HERE,
 479       base::Bind(&SyncWorkerInterface::SetRemoteChangeProcessor,
 480                  base::Unretained(sync_worker_.get()),
 481                  remote_change_processor_on_worker_.get()));
 482 }
 483
 484 LocalChangeProcessor* SyncEngine::GetLocalChangeProcessor() {
 485   return this;
 486 }
 487
 488 RemoteServiceState SyncEngine::GetCurrentState() const {
 489   if (!sync_enabled_)
 490     return REMOTE_SERVICE_DISABLED;
 491   if (!has_refresh_token_)
 492     return REMOTE_SERVICE_AUTHENTICATION_REQUIRED;
 493   return service_state_;
 494 }
 495
 496 void SyncEngine::GetOriginStatusMap(const StatusMapCallback& callback) {
 497   base::Closure abort_closure =
 498       base::Bind(callback, base::Passed(scoped_ptr<OriginStatusMap>()));
 499
 500   if (!sync_worker_) {
 501     abort_closure.Run();
 502     return;
 503   }
 504
 505   StatusMapCallback tracked_callback =
 506       callback_tracker_.Register(abort_closure, callback);
 507   StatusMapCallback relayed_callback =
 508       RelayCallbackToCurrentThread(FROM_HERE, tracked_callback);
 509
 510   worker_task_runner_->PostTask(
 511       FROM_HERE,
 512       base::Bind(&SyncWorkerInterface::GetOriginStatusMap,
 513                  base::Unretained(sync_worker_.get()),
 514                  relayed_callback));
 515 }
 516
 517 void SyncEngine::DumpFiles(const GURL& origin,
 518                            const ListCallback& callback) {
 519   base::Closure abort_closure =
 520       base::Bind(callback, base::Passed(scoped_ptr<base::ListValue>()));
 521
 522   if (!sync_worker_) {
 523     abort_closure.Run();
 524     return;
 525   }
 526
 527   ListCallback tracked_callback =
 528       callback_tracker_.Register(abort_closure, callback);
 529
 530   PostTaskAndReplyWithResult(worker_task_runner_.get(),
 531                              FROM_HERE,
 532                              base::Bind(&SyncWorkerInterface::DumpFiles,
 533                                         base::Unretained(sync_worker_.get()),
 534                                         origin),
 535                              tracked_callback);
 536 }
 537
 538 void SyncEngine::DumpDatabase(const ListCallback& callback) {
 539   base::Closure abort_closure =
 540       base::Bind(callback, base::Passed(scoped_ptr<base::ListValue>()));
 541
 542   if (!sync_worker_) {
 543     abort_closure.Run();
 544     return;
 545   }
 546
 547   ListCallback tracked_callback =
 548       callback_tracker_.Register(abort_closure, callback);
 549
 550   PostTaskAndReplyWithResult(worker_task_runner_.get(),
 551                              FROM_HERE,
 552                              base::Bind(&SyncWorkerInterface::DumpDatabase,
 553                                         base::Unretained(sync_worker_.get())),
 554                              tracked_callback);
 555 }
 556
 557 void SyncEngine::SetSyncEnabled(bool sync_enabled) {
 558   if (sync_enabled_ == sync_enabled)
 559     return;
 560   sync_enabled_ = sync_enabled;
 561
 562   if (sync_enabled_) {
 563     if (!sync_worker_)
 564       Initialize();
 565
 566     // Have no login credential.
 567     if (!sync_worker_)
 568       return;
 569
 570     worker_task_runner_->PostTask(
 571         FROM_HERE,
 572         base::Bind(&SyncWorkerInterface::SetSyncEnabled,
 573                    base::Unretained(sync_worker_.get()),
 574                    sync_enabled_));
 575     return;
 576   }
 577
 578   if (!sync_worker_)
 579     return;
 580
 581   // TODO(tzik): Consider removing SyncWorkerInterface::SetSyncEnabled and
 582   // let SyncEngine handle the flag.
 583   worker_task_runner_->PostTask(
 584       FROM_HERE,
 585       base::Bind(&SyncWorkerInterface::SetSyncEnabled,
 586                  base::Unretained(sync_worker_.get()),
 587                  sync_enabled_));
 588   Reset();
 589 }
 590
 591 void SyncEngine::PromoteDemotedChanges(const base::Closure& callback) {
 592   if (!sync_worker_) {
 593     callback.Run();
 594     return;
 595   }
 596
 597   base::Closure relayed_callback = RelayCallbackToCurrentThread(
 598       FROM_HERE, callback_tracker_.Register(callback, callback));
 599
 600   worker_task_runner_->PostTask(
 601       FROM_HERE,
 602       base::Bind(&SyncWorkerInterface::PromoteDemotedChanges,
 603                  base::Unretained(sync_worker_.get()),
 604                  relayed_callback));
 605 }
 606
 607 void SyncEngine::ApplyLocalChange(const FileChange& local_change,
 608                                   const base::FilePath& local_path,
 609                                   const SyncFileMetadata& local_metadata,
 610                                   const storage::FileSystemURL& url,
 611                                   const SyncStatusCallback& callback) {
 612   if (GetCurrentState() == REMOTE_SERVICE_DISABLED) {
 613     callback.Run(SYNC_STATUS_SYNC_DISABLED);
 614     return;
 615   }
 616
 617   if (!sync_worker_) {
 618     callback.Run(SYNC_STATUS_ABORT);
 619     return;
 620   }
 621
 622   SyncStatusCallback relayed_callback = RelayCallbackToCurrentThread(
 623       FROM_HERE, TrackCallback(callback));
 624   worker_task_runner_->PostTask(
 625       FROM_HERE,
 626       base::Bind(&SyncWorkerInterface::ApplyLocalChange,
 627                  base::Unretained(sync_worker_.get()),
 628                  local_change,
 629                  local_path,
 630                  local_metadata,
 631                  url,
 632                  relayed_callback));
 633 }
 634
 635 void SyncEngine::OnNotificationReceived() {
 636   if (!sync_worker_)
 637     return;
 638
 639   worker_task_runner_->PostTask(
 640       FROM_HERE,
 641       base::Bind(&SyncWorkerInterface::ActivateService,
 642                  base::Unretained(sync_worker_.get()),
 643                  REMOTE_SERVICE_OK,
 644                  "Got push notification for Drive"));
 645 }
 646
 647 void SyncEngine::OnPushNotificationEnabled(bool /* enabled */) {}
 648
 649 void SyncEngine::OnReadyToSendRequests() {
 650   has_refresh_token_ = true;
 651   if (!sync_worker_)
 652     return;
 653
 654   worker_task_runner_->PostTask(
 655       FROM_HERE,
 656       base::Bind(&SyncWorkerInterface::ActivateService,
 657                  base::Unretained(sync_worker_.get()),
 658                  REMOTE_SERVICE_OK,
 659                  "Authenticated"));
 660 }
 661
 662 void SyncEngine::OnRefreshTokenInvalid() {
 663   has_refresh_token_ = false;
 664   if (!sync_worker_)
 665     return;
 666
 667   worker_task_runner_->PostTask(
 668       FROM_HERE,
 669       base::Bind(&SyncWorkerInterface::DeactivateService,
 670                  base::Unretained(sync_worker_.get()),
 671                  "Found invalid refresh token."));
 672 }
 673
 674 void SyncEngine::OnNetworkChanged(
 675     net::NetworkChangeNotifier::ConnectionType type) {
 676   if (!sync_worker_)
 677     return;
 678
 679   bool network_available_old = network_available_;
 680   network_available_ = (type != net::NetworkChangeNotifier::CONNECTION_NONE);
 681
 682   if (!network_available_old && network_available_) {
 683     worker_task_runner_->PostTask(
 684         FROM_HERE,
 685         base::Bind(&SyncWorkerInterface::ActivateService,
 686                    base::Unretained(sync_worker_.get()),
 687                    REMOTE_SERVICE_OK,
 688                    "Connected"));
 689   } else if (network_available_old && !network_available_) {
 690     worker_task_runner_->PostTask(
 691         FROM_HERE,
 692         base::Bind(&SyncWorkerInterface::DeactivateService,
 693                    base::Unretained(sync_worker_.get()),
 694                    "Disconnected"));
 695   }
 696 }
 697
 698 void SyncEngine::GoogleSigninFailed(const GoogleServiceAuthError& error) {
 699   Reset();
 700   UpdateServiceState(REMOTE_SERVICE_AUTHENTICATION_REQUIRED,
 701                      "Failed to sign in.");
 702 }
 703
 704 void SyncEngine::GoogleSigninSucceeded(const std::string& account_id,
 705                                        const std::string& username,
 706                                        const std::string& password) {
 707   Initialize();
 708 }
 709
 710 void SyncEngine::GoogleSignedOut(const std::string& account_id,
 711                                  const std::string& username) {
 712   Reset();
 713   UpdateServiceState(REMOTE_SERVICE_AUTHENTICATION_REQUIRED,
 714                      "User signed out.");
 715 }
 716
 717 SyncEngine::SyncEngine(
 718     const scoped_refptr<base::SingleThreadTaskRunner>& ui_task_runner,
 719     const scoped_refptr<base::SequencedTaskRunner>& worker_task_runner,
 720     const scoped_refptr<base::SequencedTaskRunner>& drive_task_runner,
 721     const scoped_refptr<base::SequencedWorkerPool>& worker_pool,
 722     const base::FilePath& sync_file_system_dir,
 723     TaskLogger* task_logger,
 724     drive::DriveNotificationManager* notification_manager,
 725     ExtensionServiceInterface* extension_service,
 726     SigninManagerBase* signin_manager,
 727     OAuth2TokenService* token_service,
 728     net::URLRequestContextGetter* request_context,
 729     scoped_ptr<DriveServiceFactory> drive_service_factory,
 730     leveldb::Env* env_override)
 731     : ui_task_runner_(ui_task_runner),
 732       worker_task_runner_(worker_task_runner),
 733       drive_task_runner_(drive_task_runner),
 734       worker_pool_(worker_pool),
 735       sync_file_system_dir_(sync_file_system_dir),
 736       task_logger_(task_logger),
 737       notification_manager_(notification_manager),
 738       extension_service_(extension_service),
 739       signin_manager_(signin_manager),
 740       token_service_(token_service),
 741       request_context_(request_context),
 742       drive_service_factory_(drive_service_factory.Pass()),
 743       remote_change_processor_(nullptr),
 744       service_state_(REMOTE_SERVICE_TEMPORARY_UNAVAILABLE),
 745       has_refresh_token_(false),
 746       network_available_(false),
 747       sync_enabled_(false),
 748       env_override_(env_override),
 749       weak_ptr_factory_(this) {
 750   DCHECK(sync_file_system_dir_.IsAbsolute());
 751   if (notification_manager_)
 752     notification_manager_->AddObserver(this);
 753   if (signin_manager_)
 754     signin_manager_->AddObserver(this);
 755   net::NetworkChangeNotifier::AddNetworkChangeObserver(this);
 756 }
 757
 758 void SyncEngine::OnPendingFileListUpdated(int item_count) {
 759   FOR_EACH_OBSERVER(
 760       SyncServiceObserver,
 761       service_observers_,
 762       OnRemoteChangeQueueUpdated(item_count));
 763 }
 764
 765 void SyncEngine::OnFileStatusChanged(const storage::FileSystemURL& url,
 766                                      SyncFileType file_type,
 767                                      SyncFileStatus file_status,
 768                                      SyncAction sync_action,
 769                                      SyncDirection direction) {
 770   FOR_EACH_OBSERVER(FileStatusObserver,
 771                     file_status_observers_,
 772                     OnFileStatusChanged(
 773                         url, file_type, file_status, sync_action, direction));
 774 }
 775
 776 void SyncEngine::UpdateServiceState(RemoteServiceState state,
 777                                     const std::string& description) {
 778   service_state_ = state;
 779
 780   FOR_EACH_OBSERVER(
 781       SyncServiceObserver, service_observers_,
 782       OnRemoteServiceStateUpdated(GetCurrentState(), description));
 783 }
 784
 785 SyncStatusCallback SyncEngine::TrackCallback(
 786     const SyncStatusCallback& callback) {
 787   return callback_tracker_.Register(
 788       base::Bind(callback, SYNC_STATUS_ABORT),
 789       callback);
 790 }
 791
 792 }  // namespace drive_backend
 793 }  // namespace sync_file_system