path: root/components/drive/change_list_loader.cc

// Copyright (c) 2012 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 "components/drive/change_list_loader.h"

#include <stddef.h>
#include <set>
#include <utility>

#include "base/callback.h"
#include "base/callback_helpers.h"
#include "base/macros.h"
#include "base/metrics/histogram.h"
#include "base/strings/string_number_conversions.h"
#include "base/synchronization/cancellation_flag.h"
#include "base/thread_task_runner_handle.h"
#include "base/time/time.h"
#include "components/drive/change_list_loader_observer.h"
#include "components/drive/change_list_processor.h"
#include "components/drive/event_logger.h"
#include "components/drive/file_system_core_util.h"
#include "components/drive/job_scheduler.h"
#include "components/drive/resource_metadata.h"
#include "google_apis/drive/drive_api_parser.h"
#include "url/gurl.h"

namespace drive {
namespace internal {

typedef base::Callback<void(FileError, ScopedVector<ChangeList>)>
    FeedFetcherCallback;

class ChangeListLoader::FeedFetcher {
 public:
  virtual ~FeedFetcher() {}
  virtual void Run(const FeedFetcherCallback& callback) = 0;
};

namespace {

// Fetches all the (currently available) resource entries from the server.
class FullFeedFetcher : public ChangeListLoader::FeedFetcher {
 public:
  explicit FullFeedFetcher(JobScheduler* scheduler)
      : scheduler_(scheduler),
        weak_ptr_factory_(this) {
  }

  ~FullFeedFetcher() override {}

  void Run(const FeedFetcherCallback& callback) override {
    DCHECK(thread_checker_.CalledOnValidThread());
    DCHECK(!callback.is_null());

    // Remember the time stamp for usage stats.
    start_time_ = base::TimeTicks::Now();

    // This is full resource list fetch.
    scheduler_->GetAllFileList(
        base::Bind(&FullFeedFetcher::OnFileListFetched,
                   weak_ptr_factory_.GetWeakPtr(), callback));
  }

 private:
  void OnFileListFetched(const FeedFetcherCallback& callback,
                         google_apis::DriveApiErrorCode status,
                         scoped_ptr<google_apis::FileList> file_list) {
    DCHECK(thread_checker_.CalledOnValidThread());
    DCHECK(!callback.is_null());

    FileError error = GDataToFileError(status);
    if (error != FILE_ERROR_OK) {
      callback.Run(error, ScopedVector<ChangeList>());
      return;
    }

    DCHECK(file_list);
    change_lists_.push_back(new ChangeList(*file_list));

    if (!file_list->next_link().is_empty()) {
      // There is the remaining result so fetch it.
      scheduler_->GetRemainingFileList(
          file_list->next_link(),
          base::Bind(&FullFeedFetcher::OnFileListFetched,
                     weak_ptr_factory_.GetWeakPtr(), callback));
      return;
    }

    UMA_HISTOGRAM_LONG_TIMES("Drive.FullFeedLoadTime",
                             base::TimeTicks::Now() - start_time_);

    // Note: The fetcher is managed by ChangeListLoader, and the instance
    // will be deleted in the callback. Do not touch the fields after this
    // invocation.
    callback.Run(FILE_ERROR_OK, std::move(change_lists_));
  }

  JobScheduler* scheduler_;
  ScopedVector<ChangeList> change_lists_;
  base::TimeTicks start_time_;
  base::ThreadChecker thread_checker_;
  base::WeakPtrFactory<FullFeedFetcher> weak_ptr_factory_;
  DISALLOW_COPY_AND_ASSIGN(FullFeedFetcher);
};

// Fetches the delta changes since |start_change_id|.
class DeltaFeedFetcher : public ChangeListLoader::FeedFetcher {
 public:
  DeltaFeedFetcher(JobScheduler* scheduler, int64_t start_change_id)
      : scheduler_(scheduler),
        start_change_id_(start_change_id),
        weak_ptr_factory_(this) {}

  ~DeltaFeedFetcher() override {}

  void Run(const FeedFetcherCallback& callback) override {
    DCHECK(thread_checker_.CalledOnValidThread());
    DCHECK(!callback.is_null());

    scheduler_->GetChangeList(
        start_change_id_,
        base::Bind(&DeltaFeedFetcher::OnChangeListFetched,
                   weak_ptr_factory_.GetWeakPtr(), callback));
  }

 private:
  void OnChangeListFetched(const FeedFetcherCallback& callback,
                           google_apis::DriveApiErrorCode status,
                           scoped_ptr<google_apis::ChangeList> change_list) {
    DCHECK(thread_checker_.CalledOnValidThread());
    DCHECK(!callback.is_null());

    FileError error = GDataToFileError(status);
    if (error != FILE_ERROR_OK) {
      callback.Run(error, ScopedVector<ChangeList>());
      return;
    }

    DCHECK(change_list);
    change_lists_.push_back(new ChangeList(*change_list));

    if (!change_list->next_link().is_empty()) {
      // There is the remaining result so fetch it.
      scheduler_->GetRemainingChangeList(
          change_list->next_link(),
          base::Bind(&DeltaFeedFetcher::OnChangeListFetched,
                     weak_ptr_factory_.GetWeakPtr(), callback));
      return;
    }

    // Note: The fetcher is managed by ChangeListLoader, and the instance
    // will be deleted in the callback. Do not touch the fields after this
    // invocation.
    callback.Run(FILE_ERROR_OK, std::move(change_lists_));
  }

  JobScheduler* scheduler_;
  int64_t start_change_id_;
  ScopedVector<ChangeList> change_lists_;
  base::ThreadChecker thread_checker_;
  base::WeakPtrFactory<DeltaFeedFetcher> weak_ptr_factory_;
  DISALLOW_COPY_AND_ASSIGN(DeltaFeedFetcher);
};

}  // namespace

LoaderController::LoaderController()
    : lock_count_(0),
      weak_ptr_factory_(this) {
}

LoaderController::~LoaderController() {
  DCHECK(thread_checker_.CalledOnValidThread());
}

scoped_ptr<base::ScopedClosureRunner> LoaderController::GetLock() {
  DCHECK(thread_checker_.CalledOnValidThread());

  ++lock_count_;
  return make_scoped_ptr(new base::ScopedClosureRunner(
      base::Bind(&LoaderController::Unlock,
                 weak_ptr_factory_.GetWeakPtr())));
}

void LoaderController::ScheduleRun(const base::Closure& task) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(!task.is_null());

  if (lock_count_ > 0) {
    pending_tasks_.push_back(task);
  } else {
    task.Run();
  }
}

void LoaderController::Unlock() {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK_LT(0, lock_count_);

  if (--lock_count_ > 0)
    return;

  std::vector<base::Closure> tasks;
  tasks.swap(pending_tasks_);
  for (size_t i = 0; i < tasks.size(); ++i)
    tasks[i].Run();
}

AboutResourceLoader::AboutResourceLoader(JobScheduler* scheduler)
    : scheduler_(scheduler),
      current_update_task_id_(-1),
      weak_ptr_factory_(this) {
}

AboutResourceLoader::~AboutResourceLoader() {}

void AboutResourceLoader::GetAboutResource(
    const google_apis::AboutResourceCallback& callback) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(!callback.is_null());

  // If the latest UpdateAboutResource task is still running. Wait for it,
  if (pending_callbacks_.count(current_update_task_id_)) {
    pending_callbacks_[current_update_task_id_].push_back(callback);
    return;
  }

  if (cached_about_resource_) {
    base::ThreadTaskRunnerHandle::Get()->PostTask(
        FROM_HERE,
        base::Bind(
            callback,
            google_apis::HTTP_NO_CONTENT,
            base::Passed(scoped_ptr<google_apis::AboutResource>(
                new google_apis::AboutResource(*cached_about_resource_)))));
  } else {
    UpdateAboutResource(callback);
  }
}

void AboutResourceLoader::UpdateAboutResource(
    const google_apis::AboutResourceCallback& callback) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(!callback.is_null());

  ++current_update_task_id_;
  pending_callbacks_[current_update_task_id_].push_back(callback);

  scheduler_->GetAboutResource(
      base::Bind(&AboutResourceLoader::UpdateAboutResourceAfterGetAbout,
                 weak_ptr_factory_.GetWeakPtr(),
                 current_update_task_id_));
}

void AboutResourceLoader::UpdateAboutResourceAfterGetAbout(
    int task_id,
    google_apis::DriveApiErrorCode status,
    scoped_ptr<google_apis::AboutResource> about_resource) {
  DCHECK(thread_checker_.CalledOnValidThread());
  FileError error = GDataToFileError(status);

  const std::vector<google_apis::AboutResourceCallback> callbacks =
      pending_callbacks_[task_id];
  pending_callbacks_.erase(task_id);

  if (error != FILE_ERROR_OK) {
    for (size_t i = 0; i < callbacks.size(); ++i)
      callbacks[i].Run(status, scoped_ptr<google_apis::AboutResource>());
    return;
  }

  // Updates the cache when the resource is successfully obtained.
  if (cached_about_resource_ &&
      cached_about_resource_->largest_change_id() >
      about_resource->largest_change_id()) {
    LOG(WARNING) << "Local cached about resource is fresher than server, "
                 << "local = " << cached_about_resource_->largest_change_id()
                 << ", server = " << about_resource->largest_change_id();
  }
  cached_about_resource_.reset(new google_apis::AboutResource(*about_resource));

  for (size_t i = 0; i < callbacks.size(); ++i) {
    callbacks[i].Run(
        status,
        make_scoped_ptr(new google_apis::AboutResource(*about_resource)));
  }
}

ChangeListLoader::ChangeListLoader(
    EventLogger* logger,
    base::SequencedTaskRunner* blocking_task_runner,
    ResourceMetadata* resource_metadata,
    JobScheduler* scheduler,
    AboutResourceLoader* about_resource_loader,
    LoaderController* loader_controller)
    : logger_(logger),
      blocking_task_runner_(blocking_task_runner),
      in_shutdown_(new base::CancellationFlag),
      resource_metadata_(resource_metadata),
      scheduler_(scheduler),
      about_resource_loader_(about_resource_loader),
      loader_controller_(loader_controller),
      loaded_(false),
      weak_ptr_factory_(this) {
}

ChangeListLoader::~ChangeListLoader() {
  in_shutdown_->Set();
  // Delete |in_shutdown_| with the blocking task runner so that it gets deleted
  // after all active ChangeListProcessors.
  blocking_task_runner_->DeleteSoon(FROM_HERE, in_shutdown_.release());
}

bool ChangeListLoader::IsRefreshing() const {
  // Callback for change list loading is stored in pending_load_callback_.
  // It is non-empty if and only if there is an in-flight loading operation.
  return !pending_load_callback_.empty();
}

void ChangeListLoader::AddObserver(ChangeListLoaderObserver* observer) {
  DCHECK(thread_checker_.CalledOnValidThread());
  observers_.AddObserver(observer);
}

void ChangeListLoader::RemoveObserver(ChangeListLoaderObserver* observer) {
  DCHECK(thread_checker_.CalledOnValidThread());
  observers_.RemoveObserver(observer);
}

void ChangeListLoader::CheckForUpdates(const FileOperationCallback& callback) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(!callback.is_null());

  // We only start to check for updates iff the load is done.
  // I.e., we ignore checking updates if not loaded to avoid starting the
  // load without user's explicit interaction (such as opening Drive).
  if (!loaded_ && !IsRefreshing())
    return;

  // For each CheckForUpdates() request, always refresh the changestamp info.
  about_resource_loader_->UpdateAboutResource(
      base::Bind(&ChangeListLoader::OnAboutResourceUpdated,
                 weak_ptr_factory_.GetWeakPtr()));

  if (IsRefreshing()) {
    // There is in-flight loading. So keep the callback here, and check for
    // updates when the in-flight loading is completed.
    pending_update_check_callback_ = callback;
    return;
  }

  DCHECK(loaded_);
  logger_->Log(logging::LOG_INFO, "Checking for updates");
  Load(callback);
}

void ChangeListLoader::LoadIfNeeded(const FileOperationCallback& callback) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(!callback.is_null());

  // If the metadata is not yet loaded, start loading.
  if (!loaded_ && !IsRefreshing())
    Load(callback);
}

void ChangeListLoader::Load(const FileOperationCallback& callback) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(!callback.is_null());

  // Check if this is the first time this ChangeListLoader do loading.
  // Note: IsRefreshing() depends on pending_load_callback_ so check in advance.
  const bool is_initial_load = (!loaded_ && !IsRefreshing());

  // Register the callback function to be called when it is loaded.
  pending_load_callback_.push_back(callback);

  // If loading task is already running, do nothing.
  if (pending_load_callback_.size() > 1)
    return;

  // Check the current status of local metadata, and start loading if needed.
  int64_t* local_changestamp = new int64_t(0);
  base::PostTaskAndReplyWithResult(
      blocking_task_runner_.get(),
      FROM_HERE,
      base::Bind(&ResourceMetadata::GetLargestChangestamp,
                 base::Unretained(resource_metadata_),
                 local_changestamp),
      base::Bind(&ChangeListLoader::LoadAfterGetLargestChangestamp,
                 weak_ptr_factory_.GetWeakPtr(),
                 is_initial_load,
                 base::Owned(local_changestamp)));
}

void ChangeListLoader::LoadAfterGetLargestChangestamp(
    bool is_initial_load,
    const int64_t* local_changestamp,
    FileError error) {
  DCHECK(thread_checker_.CalledOnValidThread());

  if (error != FILE_ERROR_OK) {
    OnChangeListLoadComplete(error);
    return;
  }

  if (is_initial_load && *local_changestamp > 0) {
    // The local data is usable. Flush callbacks to tell loading was successful.
    OnChangeListLoadComplete(FILE_ERROR_OK);

    // Continues to load from server in background.
    // Put dummy callbacks to indicate that fetching is still continuing.
    pending_load_callback_.push_back(
        base::Bind(&util::EmptyFileOperationCallback));
  }

  about_resource_loader_->GetAboutResource(
      base::Bind(&ChangeListLoader::LoadAfterGetAboutResource,
                 weak_ptr_factory_.GetWeakPtr(),
                 *local_changestamp));
}

void ChangeListLoader::LoadAfterGetAboutResource(
    int64_t local_changestamp,
    google_apis::DriveApiErrorCode status,
    scoped_ptr<google_apis::AboutResource> about_resource) {
  DCHECK(thread_checker_.CalledOnValidThread());

  FileError error = GDataToFileError(status);
  if (error != FILE_ERROR_OK) {
    OnChangeListLoadComplete(error);
    return;
  }

  DCHECK(about_resource);

  int64_t remote_changestamp = about_resource->largest_change_id();
  int64_t start_changestamp = local_changestamp > 0 ? local_changestamp + 1 : 0;
  if (local_changestamp >= remote_changestamp) {
    if (local_changestamp > remote_changestamp) {
      LOG(WARNING) << "Local resource metadata is fresher than server, "
                   << "local = " << local_changestamp
                   << ", server = " << remote_changestamp;
    }

    // No changes detected, tell the client that the loading was successful.
    OnChangeListLoadComplete(FILE_ERROR_OK);
  } else {
    // Start loading the change list.
    LoadChangeListFromServer(start_changestamp);
  }
}

void ChangeListLoader::OnChangeListLoadComplete(FileError error) {
  DCHECK(thread_checker_.CalledOnValidThread());

  if (!loaded_ && error == FILE_ERROR_OK) {
    loaded_ = true;
    FOR_EACH_OBSERVER(ChangeListLoaderObserver,
                      observers_,
                      OnInitialLoadComplete());
  }

  for (size_t i = 0; i < pending_load_callback_.size(); ++i) {
    base::ThreadTaskRunnerHandle::Get()->PostTask(
        FROM_HERE,
        base::Bind(pending_load_callback_[i], error));
  }
  pending_load_callback_.clear();

  // If there is pending update check, try to load the change from the server
  // again, because there may exist an update during the completed loading.
  if (!pending_update_check_callback_.is_null()) {
    Load(base::ResetAndReturn(&pending_update_check_callback_));
  }
}

void ChangeListLoader::OnAboutResourceUpdated(
    google_apis::DriveApiErrorCode error,
    scoped_ptr<google_apis::AboutResource> resource) {
  DCHECK(thread_checker_.CalledOnValidThread());

  if (drive::GDataToFileError(error) != drive::FILE_ERROR_OK) {
    logger_->Log(logging::LOG_ERROR,
                 "Failed to update the about resource: %s",
                 google_apis::DriveApiErrorCodeToString(error).c_str());
    return;
  }
  logger_->Log(logging::LOG_INFO,
               "About resource updated to: %s",
               base::Int64ToString(resource->largest_change_id()).c_str());
}

void ChangeListLoader::LoadChangeListFromServer(int64_t start_changestamp) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(!change_feed_fetcher_);
  DCHECK(about_resource_loader_->cached_about_resource());

  bool is_delta_update = start_changestamp != 0;

  // Set up feed fetcher.
  if (is_delta_update) {
    change_feed_fetcher_.reset(
        new DeltaFeedFetcher(scheduler_, start_changestamp));
  } else {
    change_feed_fetcher_.reset(new FullFeedFetcher(scheduler_));
  }

  // Make a copy of cached_about_resource_ to remember at which changestamp we
  // are fetching change list.
  change_feed_fetcher_->Run(
      base::Bind(&ChangeListLoader::LoadChangeListFromServerAfterLoadChangeList,
                 weak_ptr_factory_.GetWeakPtr(),
                 base::Passed(make_scoped_ptr(new google_apis::AboutResource(
                     *about_resource_loader_->cached_about_resource()))),
                 is_delta_update));
}

void ChangeListLoader::LoadChangeListFromServerAfterLoadChangeList(
    scoped_ptr<google_apis::AboutResource> about_resource,
    bool is_delta_update,
    FileError error,
    ScopedVector<ChangeList> change_lists) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(about_resource);

  // Delete the fetcher first.
  change_feed_fetcher_.reset();

  if (error != FILE_ERROR_OK) {
    OnChangeListLoadComplete(error);
    return;
  }

  ChangeListProcessor* change_list_processor =
      new ChangeListProcessor(resource_metadata_, in_shutdown_.get());
  // Don't send directory content change notification while performing
  // the initial content retrieval.
  const bool should_notify_changed_directories = is_delta_update;

  logger_->Log(logging::LOG_INFO,
               "Apply change lists (is delta: %d)",
               is_delta_update);
  loader_controller_->ScheduleRun(base::Bind(
      base::IgnoreResult(
          &base::PostTaskAndReplyWithResult<FileError, FileError>),
      base::RetainedRef(blocking_task_runner_), FROM_HERE,
      base::Bind(&ChangeListProcessor::Apply,
                 base::Unretained(change_list_processor),
                 base::Passed(&about_resource), base::Passed(&change_lists),
                 is_delta_update),
      base::Bind(&ChangeListLoader::LoadChangeListFromServerAfterUpdate,
                 weak_ptr_factory_.GetWeakPtr(),
                 base::Owned(change_list_processor),
                 should_notify_changed_directories, base::Time::Now())));
}

void ChangeListLoader::LoadChangeListFromServerAfterUpdate(
    ChangeListProcessor* change_list_processor,
    bool should_notify_changed_directories,
    const base::Time& start_time,
    FileError error) {
  DCHECK(thread_checker_.CalledOnValidThread());

  const base::TimeDelta elapsed = base::Time::Now() - start_time;
  logger_->Log(logging::LOG_INFO,
               "Change lists applied (elapsed time: %sms)",
               base::Int64ToString(elapsed.InMilliseconds()).c_str());

  if (should_notify_changed_directories) {
    FOR_EACH_OBSERVER(ChangeListLoaderObserver,
                      observers_,
                      OnFileChanged(change_list_processor->changed_files()));
  }

  OnChangeListLoadComplete(error);

  FOR_EACH_OBSERVER(ChangeListLoaderObserver,
                    observers_,
                    OnLoadFromServerComplete());
}

}  // namespace internal
}  // namespace drive