chrome/browser/chromeos/drive/job_scheduler.cc

   1 // Copyright (c) 2012 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/chromeos/drive/job_scheduler.h"
   6
   7 #include "base/message_loop/message_loop.h"
   8 #include "base/prefs/pref_service.h"
   9 #include "base/rand_util.h"
  10 #include "base/strings/string_number_conversions.h"
  11 #include "base/strings/stringprintf.h"
  12 #include "chrome/browser/chromeos/drive/file_system_util.h"
  13 #include "chrome/browser/chromeos/drive/logging.h"
  14 #include "chrome/common/pref_names.h"
  15 #include "content/public/browser/browser_thread.h"
  16 #include "google_apis/drive/drive_api_parser.h"
  17
  18 using content::BrowserThread;
  19
  20 namespace drive {
  21
  22 namespace {
  23
  24 // All jobs are retried at maximum of kMaxRetryCount when they fail due to
  25 // throttling or server error.  The delay before retrying a job is shared among
  26 // jobs. It doubles in length on each failure, upto 2^kMaxThrottleCount seconds.
  27 //
  28 // According to the API documentation, kMaxRetryCount should be the same as
  29 // kMaxThrottleCount (https://developers.google.com/drive/handle-errors).
  30 // But currently multiplied by 2 to ensure upload related jobs retried for a
  31 // sufficient number of times. crbug.com/269918
  32 const int kMaxThrottleCount = 4;
  33 const int kMaxRetryCount = 2 * kMaxThrottleCount;
  34
  35 // GetDefaultValue returns a value constructed by the default constructor.
  36 template<typename T> struct DefaultValueCreator {
  37   static T GetDefaultValue() { return T(); }
  38 };
  39 template<typename T> struct DefaultValueCreator<const T&> {
  40   static T GetDefaultValue() { return T(); }
  41 };
  42
  43 // Helper of CreateErrorRunCallback implementation.
  44 // Provides:
  45 // - ResultType; the type of the Callback which should be returned by
  46 //     CreateErrorRunCallback.
  47 // - Run(): a static function which takes the original |callback| and |error|,
  48 //     and runs the |callback|.Run() with the error code and default values
  49 //     for remaining arguments.
  50 template<typename CallbackType> struct CreateErrorRunCallbackHelper;
  51
  52 // CreateErrorRunCallback with two arguments.
  53 template<typename P1>
  54 struct CreateErrorRunCallbackHelper<void(google_apis::GDataErrorCode, P1)> {
  55   static void Run(
  56       const base::Callback<void(google_apis::GDataErrorCode, P1)>& callback,
  57       google_apis::GDataErrorCode error) {
  58     callback.Run(error, DefaultValueCreator<P1>::GetDefaultValue());
  59   }
  60 };
  61
  62 // Returns a callback with the tail parameter bound to its default value.
  63 // In other words, returned_callback.Run(error) runs callback.Run(error, T()).
  64 template<typename CallbackType>
  65 base::Callback<void(google_apis::GDataErrorCode)>
  66 CreateErrorRunCallback(const base::Callback<CallbackType>& callback) {
  67   return base::Bind(&CreateErrorRunCallbackHelper<CallbackType>::Run, callback);
  68 }
  69
  70 // Parameter struct for RunUploadNewFile.
  71 struct UploadNewFileParams {
  72   std::string parent_resource_id;
  73   base::FilePath local_file_path;
  74   std::string title;
  75   std::string content_type;
  76   UploadCompletionCallback callback;
  77   google_apis::ProgressCallback progress_callback;
  78 };
  79
  80 // Helper function to work around the arity limitation of base::Bind.
  81 google_apis::CancelCallback RunUploadNewFile(
  82     DriveUploaderInterface* uploader,
  83     const UploadNewFileParams& params) {
  84   return uploader->UploadNewFile(params.parent_resource_id,
  85                                  params.local_file_path,
  86                                  params.title,
  87                                  params.content_type,
  88                                  params.callback,
  89                                  params.progress_callback);
  90 }
  91
  92 // Parameter struct for RunUploadExistingFile.
  93 struct UploadExistingFileParams {
  94   std::string resource_id;
  95   base::FilePath local_file_path;
  96   std::string content_type;
  97   std::string etag;
  98   UploadCompletionCallback callback;
  99   google_apis::ProgressCallback progress_callback;
 100 };
 101
 102 // Helper function to work around the arity limitation of base::Bind.
 103 google_apis::CancelCallback RunUploadExistingFile(
 104     DriveUploaderInterface* uploader,
 105     const UploadExistingFileParams& params) {
 106   return uploader->UploadExistingFile(params.resource_id,
 107                                       params.local_file_path,
 108                                       params.content_type,
 109                                       params.etag,
 110                                       params.callback,
 111                                       params.progress_callback);
 112 }
 113
 114 // Parameter struct for RunResumeUploadFile.
 115 struct ResumeUploadFileParams {
 116   GURL upload_location;
 117   base::FilePath local_file_path;
 118   std::string content_type;
 119   UploadCompletionCallback callback;
 120   google_apis::ProgressCallback progress_callback;
 121 };
 122
 123 // Helper function to adjust the return type.
 124 google_apis::CancelCallback RunResumeUploadFile(
 125     DriveUploaderInterface* uploader,
 126     const ResumeUploadFileParams& params) {
 127   return uploader->ResumeUploadFile(params.upload_location,
 128                                     params.local_file_path,
 129                                     params.content_type,
 130                                     params.callback,
 131                                     params.progress_callback);
 132 }
 133
 134 }  // namespace
 135
 136 // Metadata jobs are cheap, so we run them concurrently. File jobs run serially.
 137 const int JobScheduler::kMaxJobCount[] = {
 138   5,  // METADATA_QUEUE
 139   1,  // FILE_QUEUE
 140 };
 141
 142 JobScheduler::JobEntry::JobEntry(JobType type)
 143     : job_info(type),
 144       context(ClientContext(USER_INITIATED)),
 145       retry_count(0) {
 146   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 147 }
 148
 149 JobScheduler::JobEntry::~JobEntry() {
 150   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 151 }
 152
 153 struct JobScheduler::ResumeUploadParams {
 154   base::FilePath drive_file_path;
 155   base::FilePath local_file_path;
 156   std::string content_type;
 157 };
 158
 159 JobScheduler::JobScheduler(
 160     PrefService* pref_service,
 161     DriveServiceInterface* drive_service,
 162     base::SequencedTaskRunner* blocking_task_runner)
 163     : throttle_count_(0),
 164       wait_until_(base::Time::Now()),
 165       disable_throttling_(false),
 166       drive_service_(drive_service),
 167       uploader_(new DriveUploader(drive_service, blocking_task_runner)),
 168       pref_service_(pref_service),
 169       weak_ptr_factory_(this) {
 170   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 171
 172   for (int i = 0; i < NUM_QUEUES; ++i)
 173     queue_[i].reset(new JobQueue(kMaxJobCount[i], NUM_CONTEXT_TYPES));
 174
 175   net::NetworkChangeNotifier::AddConnectionTypeObserver(this);
 176 }
 177
 178 JobScheduler::~JobScheduler() {
 179   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 180
 181   size_t num_queued_jobs = 0;
 182   for (int i = 0; i < NUM_QUEUES; ++i)
 183     num_queued_jobs += queue_[i]->GetNumberOfJobs();
 184   DCHECK_EQ(num_queued_jobs, job_map_.size());
 185
 186   net::NetworkChangeNotifier::RemoveConnectionTypeObserver(this);
 187 }
 188
 189 std::vector<JobInfo> JobScheduler::GetJobInfoList() {
 190   std::vector<JobInfo> job_info_list;
 191   for (JobIDMap::iterator iter(&job_map_); !iter.IsAtEnd(); iter.Advance())
 192     job_info_list.push_back(iter.GetCurrentValue()->job_info);
 193   return job_info_list;
 194 }
 195
 196 void JobScheduler::AddObserver(JobListObserver* observer) {
 197   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 198   observer_list_.AddObserver(observer);
 199 }
 200
 201 void JobScheduler::RemoveObserver(JobListObserver* observer) {
 202   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 203   observer_list_.RemoveObserver(observer);
 204 }
 205
 206 void JobScheduler::CancelJob(JobID job_id) {
 207   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 208
 209   JobEntry* job = job_map_.Lookup(job_id);
 210   if (job) {
 211     if (job->job_info.state == STATE_RUNNING) {
 212       // If the job is running an HTTP request, cancel it via |cancel_callback|
 213       // returned from the request, and wait for termination in the normal
 214       // callback handler, OnJobDone.
 215       if (!job->cancel_callback.is_null())
 216         job->cancel_callback.Run();
 217     } else {
 218       AbortNotRunningJob(job, google_apis::GDATA_CANCELLED);
 219     }
 220   }
 221 }
 222
 223 void JobScheduler::CancelAllJobs() {
 224   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 225
 226   // CancelJob may remove the entry from |job_map_|. That's OK. IDMap supports
 227   // removable during iteration.
 228   for (JobIDMap::iterator iter(&job_map_); !iter.IsAtEnd(); iter.Advance())
 229     CancelJob(iter.GetCurrentKey());
 230 }
 231
 232 void JobScheduler::GetAboutResource(
 233     const google_apis::AboutResourceCallback& callback) {
 234   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 235   DCHECK(!callback.is_null());
 236
 237   JobEntry* new_job = CreateNewJob(TYPE_GET_ABOUT_RESOURCE);
 238   new_job->task = base::Bind(
 239       &DriveServiceInterface::GetAboutResource,
 240       base::Unretained(drive_service_),
 241       base::Bind(&JobScheduler::OnGetAboutResourceJobDone,
 242                  weak_ptr_factory_.GetWeakPtr(),
 243                  new_job->job_info.job_id,
 244                  callback));
 245   new_job->abort_callback = CreateErrorRunCallback(callback);
 246   StartJob(new_job);
 247 }
 248
 249 void JobScheduler::GetAppList(const google_apis::AppListCallback& callback) {
 250   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 251   DCHECK(!callback.is_null());
 252
 253   JobEntry* new_job = CreateNewJob(TYPE_GET_APP_LIST);
 254   new_job->task = base::Bind(
 255       &DriveServiceInterface::GetAppList,
 256       base::Unretained(drive_service_),
 257       base::Bind(&JobScheduler::OnGetAppListJobDone,
 258                  weak_ptr_factory_.GetWeakPtr(),
 259                  new_job->job_info.job_id,
 260                  callback));
 261   new_job->abort_callback = CreateErrorRunCallback(callback);
 262   StartJob(new_job);
 263 }
 264
 265 void JobScheduler::GetAllResourceList(
 266     const google_apis::GetResourceListCallback& callback) {
 267   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 268   DCHECK(!callback.is_null());
 269
 270   JobEntry* new_job = CreateNewJob(TYPE_GET_ALL_RESOURCE_LIST);
 271   new_job->task = base::Bind(
 272       &DriveServiceInterface::GetAllResourceList,
 273       base::Unretained(drive_service_),
 274       base::Bind(&JobScheduler::OnGetResourceListJobDone,
 275                  weak_ptr_factory_.GetWeakPtr(),
 276                  new_job->job_info.job_id,
 277                  callback));
 278   new_job->abort_callback = CreateErrorRunCallback(callback);
 279   StartJob(new_job);
 280 }
 281
 282 void JobScheduler::GetResourceListInDirectory(
 283     const std::string& directory_resource_id,
 284     const google_apis::GetResourceListCallback& callback) {
 285   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 286   DCHECK(!callback.is_null());
 287
 288   JobEntry* new_job = CreateNewJob(
 289       TYPE_GET_RESOURCE_LIST_IN_DIRECTORY);
 290   new_job->task = base::Bind(
 291       &DriveServiceInterface::GetResourceListInDirectory,
 292       base::Unretained(drive_service_),
 293       directory_resource_id,
 294       base::Bind(&JobScheduler::OnGetResourceListJobDone,
 295                  weak_ptr_factory_.GetWeakPtr(),
 296                  new_job->job_info.job_id,
 297                  callback));
 298   new_job->abort_callback = CreateErrorRunCallback(callback);
 299   StartJob(new_job);
 300 }
 301
 302 void JobScheduler::Search(
 303     const std::string& search_query,
 304     const google_apis::GetResourceListCallback& callback) {
 305   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 306   DCHECK(!callback.is_null());
 307
 308   JobEntry* new_job = CreateNewJob(TYPE_SEARCH);
 309   new_job->task = base::Bind(
 310       &DriveServiceInterface::Search,
 311       base::Unretained(drive_service_),
 312       search_query,
 313       base::Bind(&JobScheduler::OnGetResourceListJobDone,
 314                  weak_ptr_factory_.GetWeakPtr(),
 315                  new_job->job_info.job_id,
 316                  callback));
 317   new_job->abort_callback = CreateErrorRunCallback(callback);
 318   StartJob(new_job);
 319 }
 320
 321 void JobScheduler::GetChangeList(
 322     int64 start_changestamp,
 323     const google_apis::GetResourceListCallback& callback) {
 324   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 325   DCHECK(!callback.is_null());
 326
 327   JobEntry* new_job = CreateNewJob(TYPE_GET_CHANGE_LIST);
 328   new_job->task = base::Bind(
 329       &DriveServiceInterface::GetChangeList,
 330       base::Unretained(drive_service_),
 331       start_changestamp,
 332       base::Bind(&JobScheduler::OnGetResourceListJobDone,
 333                  weak_ptr_factory_.GetWeakPtr(),
 334                  new_job->job_info.job_id,
 335                  callback));
 336   new_job->abort_callback = CreateErrorRunCallback(callback);
 337   StartJob(new_job);
 338 }
 339
 340 void JobScheduler::GetRemainingChangeList(
 341     const GURL& next_link,
 342     const google_apis::GetResourceListCallback& callback) {
 343   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 344   DCHECK(!callback.is_null());
 345
 346   JobEntry* new_job = CreateNewJob(TYPE_GET_REMAINING_CHANGE_LIST);
 347   new_job->task = base::Bind(
 348       &DriveServiceInterface::GetRemainingChangeList,
 349       base::Unretained(drive_service_),
 350       next_link,
 351       base::Bind(&JobScheduler::OnGetResourceListJobDone,
 352                  weak_ptr_factory_.GetWeakPtr(),
 353                  new_job->job_info.job_id,
 354                  callback));
 355   new_job->abort_callback = CreateErrorRunCallback(callback);
 356   StartJob(new_job);
 357 }
 358
 359 void JobScheduler::GetRemainingFileList(
 360     const GURL& next_link,
 361     const google_apis::GetResourceListCallback& callback) {
 362   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 363   DCHECK(!callback.is_null());
 364
 365   JobEntry* new_job = CreateNewJob(TYPE_GET_REMAINING_FILE_LIST);
 366   new_job->task = base::Bind(
 367       &DriveServiceInterface::GetRemainingFileList,
 368       base::Unretained(drive_service_),
 369       next_link,
 370       base::Bind(&JobScheduler::OnGetResourceListJobDone,
 371                  weak_ptr_factory_.GetWeakPtr(),
 372                  new_job->job_info.job_id,
 373                  callback));
 374   new_job->abort_callback = CreateErrorRunCallback(callback);
 375   StartJob(new_job);
 376 }
 377
 378 void JobScheduler::GetResourceEntry(
 379     const std::string& resource_id,
 380     const ClientContext& context,
 381     const google_apis::GetResourceEntryCallback& callback) {
 382   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 383   DCHECK(!callback.is_null());
 384
 385   JobEntry* new_job = CreateNewJob(TYPE_GET_RESOURCE_ENTRY);
 386   new_job->context = context;
 387   new_job->task = base::Bind(
 388       &DriveServiceInterface::GetResourceEntry,
 389       base::Unretained(drive_service_),
 390       resource_id,
 391       base::Bind(&JobScheduler::OnGetResourceEntryJobDone,
 392                  weak_ptr_factory_.GetWeakPtr(),
 393                  new_job->job_info.job_id,
 394                  callback));
 395   new_job->abort_callback = CreateErrorRunCallback(callback);
 396   StartJob(new_job);
 397 }
 398
 399 void JobScheduler::GetShareUrl(
 400     const std::string& resource_id,
 401     const GURL& embed_origin,
 402     const ClientContext& context,
 403     const google_apis::GetShareUrlCallback& callback) {
 404   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 405   DCHECK(!callback.is_null());
 406
 407   JobEntry* new_job = CreateNewJob(TYPE_GET_SHARE_URL);
 408   new_job->context = context;
 409   new_job->task = base::Bind(
 410       &DriveServiceInterface::GetShareUrl,
 411       base::Unretained(drive_service_),
 412       resource_id,
 413       embed_origin,
 414       base::Bind(&JobScheduler::OnGetShareUrlJobDone,
 415                  weak_ptr_factory_.GetWeakPtr(),
 416                  new_job->job_info.job_id,
 417                  callback));
 418   new_job->abort_callback = CreateErrorRunCallback(callback);
 419   StartJob(new_job);
 420 }
 421
 422 void JobScheduler::TrashResource(
 423     const std::string& resource_id,
 424     const ClientContext& context,
 425     const google_apis::EntryActionCallback& callback) {
 426   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 427   DCHECK(!callback.is_null());
 428
 429   JobEntry* new_job = CreateNewJob(TYPE_TRASH_RESOURCE);
 430   new_job->context = context;
 431   new_job->task = base::Bind(
 432       &DriveServiceInterface::TrashResource,
 433       base::Unretained(drive_service_),
 434       resource_id,
 435       base::Bind(&JobScheduler::OnEntryActionJobDone,
 436                  weak_ptr_factory_.GetWeakPtr(),
 437                  new_job->job_info.job_id,
 438                  callback));
 439   new_job->abort_callback = callback;
 440   StartJob(new_job);
 441 }
 442
 443 void JobScheduler::CopyResource(
 444     const std::string& resource_id,
 445     const std::string& parent_resource_id,
 446     const std::string& new_title,
 447     const base::Time& last_modified,
 448     const google_apis::GetResourceEntryCallback& callback) {
 449   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 450   DCHECK(!callback.is_null());
 451
 452   JobEntry* new_job = CreateNewJob(TYPE_COPY_RESOURCE);
 453   new_job->task = base::Bind(
 454       &DriveServiceInterface::CopyResource,
 455       base::Unretained(drive_service_),
 456       resource_id,
 457       parent_resource_id,
 458       new_title,
 459       last_modified,
 460       base::Bind(&JobScheduler::OnGetResourceEntryJobDone,
 461                  weak_ptr_factory_.GetWeakPtr(),
 462                  new_job->job_info.job_id,
 463                  callback));
 464   new_job->abort_callback = CreateErrorRunCallback(callback);
 465   StartJob(new_job);
 466 }
 467
 468 void JobScheduler::UpdateResource(
 469     const std::string& resource_id,
 470     const std::string& parent_resource_id,
 471     const std::string& new_title,
 472     const base::Time& last_modified,
 473     const base::Time& last_viewed_by_me,
 474     const ClientContext& context,
 475     const google_apis::GetResourceEntryCallback& callback) {
 476   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 477   DCHECK(!callback.is_null());
 478
 479   JobEntry* new_job = CreateNewJob(TYPE_UPDATE_RESOURCE);
 480   new_job->context = context;
 481   new_job->task = base::Bind(
 482       &DriveServiceInterface::UpdateResource,
 483       base::Unretained(drive_service_),
 484       resource_id,
 485       parent_resource_id,
 486       new_title,
 487       last_modified,
 488       last_viewed_by_me,
 489       base::Bind(&JobScheduler::OnGetResourceEntryJobDone,
 490                  weak_ptr_factory_.GetWeakPtr(),
 491                  new_job->job_info.job_id,
 492                  callback));
 493   new_job->abort_callback = CreateErrorRunCallback(callback);
 494   StartJob(new_job);
 495 }
 496
 497 void JobScheduler::RenameResource(
 498     const std::string& resource_id,
 499     const std::string& new_title,
 500     const google_apis::EntryActionCallback& callback) {
 501   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 502   DCHECK(!callback.is_null());
 503
 504   JobEntry* new_job = CreateNewJob(TYPE_RENAME_RESOURCE);
 505   new_job->task = base::Bind(
 506       &DriveServiceInterface::RenameResource,
 507       base::Unretained(drive_service_),
 508       resource_id,
 509       new_title,
 510       base::Bind(&JobScheduler::OnEntryActionJobDone,
 511                  weak_ptr_factory_.GetWeakPtr(),
 512                  new_job->job_info.job_id,
 513                  callback));
 514   new_job->abort_callback = callback;
 515   StartJob(new_job);
 516 }
 517
 518 void JobScheduler::AddResourceToDirectory(
 519     const std::string& parent_resource_id,
 520     const std::string& resource_id,
 521     const google_apis::EntryActionCallback& callback) {
 522   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 523   DCHECK(!callback.is_null());
 524
 525   JobEntry* new_job = CreateNewJob(TYPE_ADD_RESOURCE_TO_DIRECTORY);
 526   new_job->task = base::Bind(
 527       &DriveServiceInterface::AddResourceToDirectory,
 528       base::Unretained(drive_service_),
 529       parent_resource_id,
 530       resource_id,
 531       base::Bind(&JobScheduler::OnEntryActionJobDone,
 532                  weak_ptr_factory_.GetWeakPtr(),
 533                  new_job->job_info.job_id,
 534                  callback));
 535   new_job->abort_callback = callback;
 536   StartJob(new_job);
 537 }
 538
 539 void JobScheduler::RemoveResourceFromDirectory(
 540     const std::string& parent_resource_id,
 541     const std::string& resource_id,
 542     const ClientContext& context,
 543     const google_apis::EntryActionCallback& callback) {
 544   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 545
 546   JobEntry* new_job = CreateNewJob(TYPE_REMOVE_RESOURCE_FROM_DIRECTORY);
 547   new_job->context = context;
 548   new_job->task = base::Bind(
 549       &DriveServiceInterface::RemoveResourceFromDirectory,
 550       base::Unretained(drive_service_),
 551       parent_resource_id,
 552       resource_id,
 553       base::Bind(&JobScheduler::OnEntryActionJobDone,
 554                  weak_ptr_factory_.GetWeakPtr(),
 555                  new_job->job_info.job_id,
 556                  callback));
 557   new_job->abort_callback = callback;
 558   StartJob(new_job);
 559 }
 560
 561 void JobScheduler::AddNewDirectory(
 562     const std::string& parent_resource_id,
 563     const std::string& directory_title,
 564     const google_apis::GetResourceEntryCallback& callback) {
 565   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 566
 567   JobEntry* new_job = CreateNewJob(TYPE_ADD_NEW_DIRECTORY);
 568   new_job->task = base::Bind(
 569       &DriveServiceInterface::AddNewDirectory,
 570       base::Unretained(drive_service_),
 571       parent_resource_id,
 572       directory_title,
 573       base::Bind(&JobScheduler::OnGetResourceEntryJobDone,
 574                  weak_ptr_factory_.GetWeakPtr(),
 575                  new_job->job_info.job_id,
 576                  callback));
 577   new_job->abort_callback = CreateErrorRunCallback(callback);
 578   StartJob(new_job);
 579 }
 580
 581 JobID JobScheduler::DownloadFile(
 582     const base::FilePath& virtual_path,
 583     int64 expected_file_size,
 584     const base::FilePath& local_cache_path,
 585     const std::string& resource_id,
 586     const ClientContext& context,
 587     const google_apis::DownloadActionCallback& download_action_callback,
 588     const google_apis::GetContentCallback& get_content_callback) {
 589   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 590
 591   JobEntry* new_job = CreateNewJob(TYPE_DOWNLOAD_FILE);
 592   new_job->job_info.file_path = virtual_path;
 593   new_job->job_info.num_total_bytes = expected_file_size;
 594   new_job->context = context;
 595   new_job->task = base::Bind(
 596       &DriveServiceInterface::DownloadFile,
 597       base::Unretained(drive_service_),
 598       local_cache_path,
 599       resource_id,
 600       base::Bind(&JobScheduler::OnDownloadActionJobDone,
 601                  weak_ptr_factory_.GetWeakPtr(),
 602                  new_job->job_info.job_id,
 603                  download_action_callback),
 604       get_content_callback,
 605       base::Bind(&JobScheduler::UpdateProgress,
 606                  weak_ptr_factory_.GetWeakPtr(),
 607                  new_job->job_info.job_id));
 608   new_job->abort_callback = CreateErrorRunCallback(download_action_callback);
 609   StartJob(new_job);
 610   return new_job->job_info.job_id;
 611 }
 612
 613 void JobScheduler::UploadNewFile(
 614     const std::string& parent_resource_id,
 615     const base::FilePath& drive_file_path,
 616     const base::FilePath& local_file_path,
 617     const std::string& title,
 618     const std::string& content_type,
 619     const ClientContext& context,
 620     const google_apis::GetResourceEntryCallback& callback) {
 621   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 622
 623   JobEntry* new_job = CreateNewJob(TYPE_UPLOAD_NEW_FILE);
 624   new_job->job_info.file_path = drive_file_path;
 625   new_job->context = context;
 626
 627   UploadNewFileParams params;
 628   params.parent_resource_id = parent_resource_id;
 629   params.local_file_path = local_file_path;
 630   params.title = title;
 631   params.content_type = content_type;
 632
 633   ResumeUploadParams resume_params;
 634   resume_params.local_file_path = params.local_file_path;
 635   resume_params.content_type = params.content_type;
 636
 637   params.callback = base::Bind(&JobScheduler::OnUploadCompletionJobDone,
 638                                weak_ptr_factory_.GetWeakPtr(),
 639                                new_job->job_info.job_id,
 640                                resume_params,
 641                                callback);
 642   params.progress_callback = base::Bind(&JobScheduler::UpdateProgress,
 643                                         weak_ptr_factory_.GetWeakPtr(),
 644                                         new_job->job_info.job_id);
 645   new_job->task = base::Bind(&RunUploadNewFile, uploader_.get(), params);
 646   new_job->abort_callback = CreateErrorRunCallback(callback);
 647   StartJob(new_job);
 648 }
 649
 650 void JobScheduler::UploadExistingFile(
 651     const std::string& resource_id,
 652     const base::FilePath& drive_file_path,
 653     const base::FilePath& local_file_path,
 654     const std::string& content_type,
 655     const std::string& etag,
 656     const ClientContext& context,
 657     const google_apis::GetResourceEntryCallback& callback) {
 658   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 659
 660   JobEntry* new_job = CreateNewJob(TYPE_UPLOAD_EXISTING_FILE);
 661   new_job->job_info.file_path = drive_file_path;
 662   new_job->context = context;
 663
 664   UploadExistingFileParams params;
 665   params.resource_id = resource_id;
 666   params.local_file_path = local_file_path;
 667   params.content_type = content_type;
 668   params.etag = etag;
 669
 670   ResumeUploadParams resume_params;
 671   resume_params.local_file_path = params.local_file_path;
 672   resume_params.content_type = params.content_type;
 673
 674   params.callback = base::Bind(&JobScheduler::OnUploadCompletionJobDone,
 675                                weak_ptr_factory_.GetWeakPtr(),
 676                                new_job->job_info.job_id,
 677                                resume_params,
 678                                callback);
 679   params.progress_callback = base::Bind(&JobScheduler::UpdateProgress,
 680                                         weak_ptr_factory_.GetWeakPtr(),
 681                                         new_job->job_info.job_id);
 682   new_job->task = base::Bind(&RunUploadExistingFile, uploader_.get(), params);
 683   new_job->abort_callback = CreateErrorRunCallback(callback);
 684   StartJob(new_job);
 685 }
 686
 687 void JobScheduler::CreateFile(
 688     const std::string& parent_resource_id,
 689     const base::FilePath& drive_file_path,
 690     const std::string& title,
 691     const std::string& content_type,
 692     const ClientContext& context,
 693     const google_apis::GetResourceEntryCallback& callback) {
 694   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 695
 696   const base::FilePath kDevNull(FILE_PATH_LITERAL("/dev/null"));
 697
 698   JobEntry* new_job = CreateNewJob(TYPE_CREATE_FILE);
 699   new_job->job_info.file_path = drive_file_path;
 700   new_job->context = context;
 701
 702   UploadNewFileParams params;
 703   params.parent_resource_id = parent_resource_id;
 704   params.local_file_path = kDevNull;  // Upload an empty file.
 705   params.title = title;
 706   params.content_type = content_type;
 707
 708   ResumeUploadParams resume_params;
 709   resume_params.local_file_path = params.local_file_path;
 710   resume_params.content_type = params.content_type;
 711
 712   params.callback = base::Bind(&JobScheduler::OnUploadCompletionJobDone,
 713                                weak_ptr_factory_.GetWeakPtr(),
 714                                new_job->job_info.job_id,
 715                                resume_params,
 716                                callback);
 717   params.progress_callback = google_apis::ProgressCallback();
 718
 719   new_job->task = base::Bind(&RunUploadNewFile, uploader_.get(), params);
 720   new_job->abort_callback = CreateErrorRunCallback(callback);
 721   StartJob(new_job);
 722 }
 723
 724 void JobScheduler::GetResourceListInDirectoryByWapi(
 725     const std::string& directory_resource_id,
 726     const google_apis::GetResourceListCallback& callback) {
 727   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 728   DCHECK(!callback.is_null());
 729
 730   JobEntry* new_job = CreateNewJob(
 731       TYPE_GET_RESOURCE_LIST_IN_DIRECTORY_BY_WAPI);
 732   new_job->task = base::Bind(
 733       &DriveServiceInterface::GetResourceListInDirectoryByWapi,
 734       base::Unretained(drive_service_),
 735       directory_resource_id,
 736       base::Bind(&JobScheduler::OnGetResourceListJobDone,
 737                  weak_ptr_factory_.GetWeakPtr(),
 738                  new_job->job_info.job_id,
 739                  callback));
 740   new_job->abort_callback = CreateErrorRunCallback(callback);
 741   StartJob(new_job);
 742 }
 743
 744 void JobScheduler::GetRemainingResourceList(
 745     const GURL& next_link,
 746     const google_apis::GetResourceListCallback& callback) {
 747   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 748   DCHECK(!callback.is_null());
 749
 750   JobEntry* new_job = CreateNewJob(TYPE_GET_REMAINING_RESOURCE_LIST);
 751   new_job->task = base::Bind(
 752       &DriveServiceInterface::GetRemainingResourceList,
 753       base::Unretained(drive_service_),
 754       next_link,
 755       base::Bind(&JobScheduler::OnGetResourceListJobDone,
 756                  weak_ptr_factory_.GetWeakPtr(),
 757                  new_job->job_info.job_id,
 758                  callback));
 759   new_job->abort_callback = CreateErrorRunCallback(callback);
 760   StartJob(new_job);
 761 }
 762
 763 JobScheduler::JobEntry* JobScheduler::CreateNewJob(JobType type) {
 764   JobEntry* job = new JobEntry(type);
 765   job->job_info.job_id = job_map_.Add(job);  // Takes the ownership of |job|.
 766   return job;
 767 }
 768
 769 void JobScheduler::StartJob(JobEntry* job) {
 770   DCHECK(!job->task.is_null());
 771
 772   QueueJob(job->job_info.job_id);
 773   NotifyJobAdded(job->job_info);
 774   DoJobLoop(GetJobQueueType(job->job_info.job_type));
 775 }
 776
 777 void JobScheduler::QueueJob(JobID job_id) {
 778   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 779
 780   JobEntry* job_entry = job_map_.Lookup(job_id);
 781   DCHECK(job_entry);
 782   const JobInfo& job_info = job_entry->job_info;
 783
 784   QueueType queue_type = GetJobQueueType(job_info.job_type);
 785   queue_[queue_type]->Push(job_id, job_entry->context.type);
 786
 787   const std::string retry_prefix = job_entry->retry_count > 0 ?
 788       base::StringPrintf(" (retry %d)", job_entry->retry_count) : "";
 789   util::Log(logging::LOG_INFO,
 790             "Job queued%s: %s - %s",
 791             retry_prefix.c_str(),
 792             job_info.ToString().c_str(),
 793             GetQueueInfo(queue_type).c_str());
 794 }
 795
 796 void JobScheduler::DoJobLoop(QueueType queue_type) {
 797   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 798
 799   const int accepted_priority = GetCurrentAcceptedPriority(queue_type);
 800
 801   // Abort all USER_INITAITED jobs when not accepted.
 802   if (accepted_priority < USER_INITIATED) {
 803     std::vector<JobID> jobs;
 804     queue_[queue_type]->GetQueuedJobs(USER_INITIATED, &jobs);
 805     for (size_t i = 0; i < jobs.size(); ++i) {
 806       JobEntry* job = job_map_.Lookup(jobs[i]);
 807       DCHECK(job);
 808       AbortNotRunningJob(job, google_apis::GDATA_NO_CONNECTION);
 809     }
 810   }
 811
 812   // Wait when throttled.
 813   const base::Time now = base::Time::Now();
 814   if (now < wait_until_) {
 815     base::MessageLoopProxy::current()->PostDelayedTask(
 816         FROM_HERE,
 817         base::Bind(&JobScheduler::DoJobLoop,
 818                    weak_ptr_factory_.GetWeakPtr(),
 819                    queue_type),
 820         wait_until_ - now);
 821     return;
 822   }
 823
 824   // Run the job with the highest priority in the queue.
 825   JobID job_id = -1;
 826   if (!queue_[queue_type]->PopForRun(accepted_priority, &job_id))
 827     return;
 828
 829   JobEntry* entry = job_map_.Lookup(job_id);
 830   DCHECK(entry);
 831
 832   JobInfo* job_info = &entry->job_info;
 833   job_info->state = STATE_RUNNING;
 834   job_info->start_time = now;
 835   NotifyJobUpdated(*job_info);
 836
 837   entry->cancel_callback = entry->task.Run();
 838
 839   UpdateWait();
 840
 841   util::Log(logging::LOG_INFO,
 842             "Job started: %s - %s",
 843             job_info->ToString().c_str(),
 844             GetQueueInfo(queue_type).c_str());
 845 }
 846
 847 int JobScheduler::GetCurrentAcceptedPriority(QueueType queue_type) {
 848   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 849
 850   const int kNoJobShouldRun = -1;
 851
 852   // Should stop if Drive was disabled while running the fetch loop.
 853   if (pref_service_->GetBoolean(prefs::kDisableDrive))
 854     return kNoJobShouldRun;
 855
 856   // Should stop if the network is not online.
 857   if (net::NetworkChangeNotifier::IsOffline())
 858     return kNoJobShouldRun;
 859
 860   // For the file queue, if it is on cellular network, only user initiated
 861   // operations are allowed to start.
 862   if (queue_type == FILE_QUEUE &&
 863       pref_service_->GetBoolean(prefs::kDisableDriveOverCellular) &&
 864       net::NetworkChangeNotifier::IsConnectionCellular(
 865           net::NetworkChangeNotifier::GetConnectionType()))
 866     return USER_INITIATED;
 867
 868   // Otherwise, every operations including background tasks are allowed.
 869   return BACKGROUND;
 870 }
 871
 872 void JobScheduler::UpdateWait() {
 873   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 874
 875   if (disable_throttling_ || throttle_count_ == 0)
 876     return;
 877
 878   // Exponential backoff: https://developers.google.com/drive/handle-errors.
 879   base::TimeDelta delay =
 880       base::TimeDelta::FromSeconds(1 << (throttle_count_ - 1)) +
 881       base::TimeDelta::FromMilliseconds(base::RandInt(0, 1000));
 882   VLOG(1) << "Throttling for " << delay.InMillisecondsF();
 883
 884   wait_until_ = std::max(wait_until_, base::Time::Now() + delay);
 885 }
 886
 887 bool JobScheduler::OnJobDone(JobID job_id, google_apis::GDataErrorCode error) {
 888   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 889
 890   JobEntry* job_entry = job_map_.Lookup(job_id);
 891   DCHECK(job_entry);
 892   JobInfo* job_info = &job_entry->job_info;
 893   QueueType queue_type = GetJobQueueType(job_info->job_type);
 894   queue_[queue_type]->MarkFinished(job_id);
 895
 896   const base::TimeDelta elapsed = base::Time::Now() - job_info->start_time;
 897   bool success = (GDataToFileError(error) == FILE_ERROR_OK);
 898   util::Log(success ? logging::LOG_INFO : logging::LOG_WARNING,
 899             "Job done: %s => %s (elapsed time: %sms) - %s",
 900             job_info->ToString().c_str(),
 901             GDataErrorCodeToString(error).c_str(),
 902             base::Int64ToString(elapsed.InMilliseconds()).c_str(),
 903             GetQueueInfo(queue_type).c_str());
 904
 905   // Retry, depending on the error.
 906   const bool is_server_error =
 907       error == google_apis::HTTP_SERVICE_UNAVAILABLE ||
 908       error == google_apis::HTTP_INTERNAL_SERVER_ERROR;
 909   if (is_server_error) {
 910     if (throttle_count_ < kMaxThrottleCount)
 911       ++throttle_count_;
 912     UpdateWait();
 913   } else {
 914     throttle_count_ = 0;
 915   }
 916
 917   const bool should_retry =
 918       is_server_error && job_entry->retry_count < kMaxRetryCount;
 919   if (should_retry) {
 920     job_entry->cancel_callback.Reset();
 921     job_info->state = STATE_RETRY;
 922     NotifyJobUpdated(*job_info);
 923
 924     ++job_entry->retry_count;
 925
 926     // Requeue the job.
 927     QueueJob(job_id);
 928   } else {
 929     NotifyJobDone(*job_info, error);
 930     // The job has finished, no retry will happen in the scheduler. Now we can
 931     // remove the job info from the map.
 932     job_map_.Remove(job_id);
 933   }
 934
 935   // Post a task to continue the job loop.  This allows us to finish handling
 936   // the current job before starting the next one.
 937   base::MessageLoopProxy::current()->PostTask(FROM_HERE,
 938       base::Bind(&JobScheduler::DoJobLoop,
 939                  weak_ptr_factory_.GetWeakPtr(),
 940                  queue_type));
 941   return !should_retry;
 942 }
 943
 944 void JobScheduler::OnGetResourceListJobDone(
 945     JobID job_id,
 946     const google_apis::GetResourceListCallback& callback,
 947     google_apis::GDataErrorCode error,
 948     scoped_ptr<google_apis::ResourceList> resource_list) {
 949   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 950   DCHECK(!callback.is_null());
 951
 952   if (OnJobDone(job_id, error))
 953     callback.Run(error, resource_list.Pass());
 954 }
 955
 956 void JobScheduler::OnGetResourceEntryJobDone(
 957     JobID job_id,
 958     const google_apis::GetResourceEntryCallback& callback,
 959     google_apis::GDataErrorCode error,
 960     scoped_ptr<google_apis::ResourceEntry> entry) {
 961   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 962   DCHECK(!callback.is_null());
 963
 964   if (OnJobDone(job_id, error))
 965     callback.Run(error, entry.Pass());
 966 }
 967
 968 void JobScheduler::OnGetAboutResourceJobDone(
 969     JobID job_id,
 970     const google_apis::AboutResourceCallback& callback,
 971     google_apis::GDataErrorCode error,
 972     scoped_ptr<google_apis::AboutResource> about_resource) {
 973   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 974   DCHECK(!callback.is_null());
 975
 976   if (OnJobDone(job_id, error))
 977     callback.Run(error, about_resource.Pass());
 978 }
 979
 980 void JobScheduler::OnGetShareUrlJobDone(
 981     JobID job_id,
 982     const google_apis::GetShareUrlCallback& callback,
 983     google_apis::GDataErrorCode error,
 984     const GURL& share_url) {
 985   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 986   DCHECK(!callback.is_null());
 987
 988   if (OnJobDone(job_id, error))
 989     callback.Run(error, share_url);
 990 }
 991
 992 void JobScheduler::OnGetAppListJobDone(
 993     JobID job_id,
 994     const google_apis::AppListCallback& callback,
 995     google_apis::GDataErrorCode error,
 996     scoped_ptr<google_apis::AppList> app_list) {
 997   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 998   DCHECK(!callback.is_null());
 999
1000   if (OnJobDone(job_id, error))
1001     callback.Run(error, app_list.Pass());
1002 }
1003
1004 void JobScheduler::OnEntryActionJobDone(
1005     JobID job_id,
1006     const google_apis::EntryActionCallback& callback,
1007     google_apis::GDataErrorCode error) {
1008   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1009   DCHECK(!callback.is_null());
1010
1011   if (OnJobDone(job_id, error))
1012     callback.Run(error);
1013 }
1014
1015 void JobScheduler::OnDownloadActionJobDone(
1016     JobID job_id,
1017     const google_apis::DownloadActionCallback& callback,
1018     google_apis::GDataErrorCode error,
1019     const base::FilePath& temp_file) {
1020   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1021   DCHECK(!callback.is_null());
1022
1023   if (OnJobDone(job_id, error))
1024     callback.Run(error, temp_file);
1025 }
1026
1027 void JobScheduler::OnUploadCompletionJobDone(
1028     JobID job_id,
1029     const ResumeUploadParams& resume_params,
1030     const google_apis::GetResourceEntryCallback& callback,
1031     google_apis::GDataErrorCode error,
1032     const GURL& upload_location,
1033     scoped_ptr<google_apis::ResourceEntry> resource_entry) {
1034   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1035   DCHECK(!callback.is_null());
1036
1037   if (!upload_location.is_empty()) {
1038     // If upload_location is available, update the task to resume the
1039     // upload process from the terminated point.
1040     // When we need to retry, the error code should be HTTP_SERVICE_UNAVAILABLE
1041     // so OnJobDone called below will be in charge to re-queue the job.
1042     JobEntry* job_entry = job_map_.Lookup(job_id);
1043     DCHECK(job_entry);
1044
1045     ResumeUploadFileParams params;
1046     params.upload_location = upload_location;
1047     params.local_file_path = resume_params.local_file_path;
1048     params.content_type = resume_params.content_type;
1049     params.callback = base::Bind(&JobScheduler::OnResumeUploadFileDone,
1050                                  weak_ptr_factory_.GetWeakPtr(),
1051                                  job_id,
1052                                  job_entry->task,
1053                                  callback);
1054     params.progress_callback = base::Bind(&JobScheduler::UpdateProgress,
1055                                           weak_ptr_factory_.GetWeakPtr(),
1056                                           job_id);
1057     job_entry->task = base::Bind(&RunResumeUploadFile, uploader_.get(), params);
1058   }
1059
1060   if (OnJobDone(job_id, error))
1061     callback.Run(error, resource_entry.Pass());
1062 }
1063
1064 void JobScheduler::OnResumeUploadFileDone(
1065     JobID job_id,
1066     const base::Callback<google_apis::CancelCallback()>& original_task,
1067     const google_apis::GetResourceEntryCallback& callback,
1068     google_apis::GDataErrorCode error,
1069     const GURL& upload_location,
1070     scoped_ptr<google_apis::ResourceEntry> resource_entry) {
1071   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1072   DCHECK(!original_task.is_null());
1073   DCHECK(!callback.is_null());
1074
1075   if (upload_location.is_empty()) {
1076     // If upload_location is not available, we should discard it and stop trying
1077     // to resume. Restore the original task.
1078     JobEntry* job_entry = job_map_.Lookup(job_id);
1079     DCHECK(job_entry);
1080     job_entry->task = original_task;
1081   }
1082
1083   if (OnJobDone(job_id, error))
1084     callback.Run(error, resource_entry.Pass());
1085 }
1086
1087 void JobScheduler::UpdateProgress(JobID job_id, int64 progress, int64 total) {
1088   JobEntry* job_entry = job_map_.Lookup(job_id);
1089   DCHECK(job_entry);
1090
1091   job_entry->job_info.num_completed_bytes = progress;
1092   if (total != -1)
1093     job_entry->job_info.num_total_bytes = total;
1094   NotifyJobUpdated(job_entry->job_info);
1095 }
1096
1097 void JobScheduler::OnConnectionTypeChanged(
1098     net::NetworkChangeNotifier::ConnectionType type) {
1099   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1100
1101   // Resume the job loop.
1102   // Note that we don't need to check the network connection status as it will
1103   // be checked in GetCurrentAcceptedPriority().
1104   for (int i = METADATA_QUEUE; i < NUM_QUEUES; ++i)
1105     DoJobLoop(static_cast<QueueType>(i));
1106 }
1107
1108 JobScheduler::QueueType JobScheduler::GetJobQueueType(JobType type) {
1109   switch (type) {
1110     case TYPE_GET_ABOUT_RESOURCE:
1111     case TYPE_GET_APP_LIST:
1112     case TYPE_GET_ALL_RESOURCE_LIST:
1113     case TYPE_GET_RESOURCE_LIST_IN_DIRECTORY:
1114     case TYPE_SEARCH:
1115     case TYPE_GET_CHANGE_LIST:
1116     case TYPE_GET_REMAINING_CHANGE_LIST:
1117     case TYPE_GET_REMAINING_FILE_LIST:
1118     case TYPE_GET_RESOURCE_ENTRY:
1119     case TYPE_GET_SHARE_URL:
1120     case TYPE_TRASH_RESOURCE:
1121     case TYPE_COPY_RESOURCE:
1122     case TYPE_UPDATE_RESOURCE:
1123     case TYPE_RENAME_RESOURCE:
1124     case TYPE_ADD_RESOURCE_TO_DIRECTORY:
1125     case TYPE_REMOVE_RESOURCE_FROM_DIRECTORY:
1126     case TYPE_ADD_NEW_DIRECTORY:
1127     case TYPE_CREATE_FILE:
1128     case TYPE_GET_RESOURCE_LIST_IN_DIRECTORY_BY_WAPI:
1129     case TYPE_GET_REMAINING_RESOURCE_LIST:
1130       return METADATA_QUEUE;
1131
1132     case TYPE_DOWNLOAD_FILE:
1133     case TYPE_UPLOAD_NEW_FILE:
1134     case TYPE_UPLOAD_EXISTING_FILE:
1135       return FILE_QUEUE;
1136   }
1137   NOTREACHED();
1138   return FILE_QUEUE;
1139 }
1140
1141 void JobScheduler::AbortNotRunningJob(JobEntry* job,
1142                                       google_apis::GDataErrorCode error) {
1143   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1144
1145   const base::TimeDelta elapsed = base::Time::Now() - job->job_info.start_time;
1146   const QueueType queue_type = GetJobQueueType(job->job_info.job_type);
1147   util::Log(logging::LOG_INFO,
1148             "Job aborted: %s => %s (elapsed time: %sms) - %s",
1149             job->job_info.ToString().c_str(),
1150             GDataErrorCodeToString(error).c_str(),
1151             base::Int64ToString(elapsed.InMilliseconds()).c_str(),
1152             GetQueueInfo(queue_type).c_str());
1153
1154   base::Callback<void(google_apis::GDataErrorCode)> callback =
1155       job->abort_callback;
1156   queue_[GetJobQueueType(job->job_info.job_type)]->Remove(job->job_info.job_id);
1157   NotifyJobDone(job->job_info, error);
1158   job_map_.Remove(job->job_info.job_id);
1159   base::MessageLoopProxy::current()->PostTask(FROM_HERE,
1160                                               base::Bind(callback, error));
1161 }
1162
1163 void JobScheduler::NotifyJobAdded(const JobInfo& job_info) {
1164   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1165   FOR_EACH_OBSERVER(JobListObserver, observer_list_, OnJobAdded(job_info));
1166 }
1167
1168 void JobScheduler::NotifyJobDone(const JobInfo& job_info,
1169                                  google_apis::GDataErrorCode error) {
1170   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1171   FOR_EACH_OBSERVER(JobListObserver, observer_list_,
1172                     OnJobDone(job_info, GDataToFileError(error)));
1173 }
1174
1175 void JobScheduler::NotifyJobUpdated(const JobInfo& job_info) {
1176   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
1177   FOR_EACH_OBSERVER(JobListObserver, observer_list_, OnJobUpdated(job_info));
1178 }
1179
1180 std::string JobScheduler::GetQueueInfo(QueueType type) const {
1181   return QueueTypeToString(type) + " " + queue_[type]->ToString();
1182 }
1183
1184 // static
1185 std::string JobScheduler::QueueTypeToString(QueueType type) {
1186   switch (type) {
1187     case METADATA_QUEUE:
1188       return "METADATA_QUEUE";
1189     case FILE_QUEUE:
1190       return "FILE_QUEUE";
1191     case NUM_QUEUES:
1192       break;  // This value is just a sentinel. Should never be used.
1193   }
1194   NOTREACHED();
1195   return "";
1196 }
1197
1198 }  // namespace drive