sync/notifier/sync_system_resources.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 "sync/notifier/sync_system_resources.h"

#include <cstdlib>
#include <cstring>
#include <string>

#include "base/bind.h"
#include "base/logging.h"
#include "base/message_loop/message_loop.h"
#include "base/stl_util.h"
#include "base/strings/string_util.h"
#include "base/strings/stringprintf.h"
#include "google/cacheinvalidation/client_gateway.pb.h"
#include "google/cacheinvalidation/deps/callback.h"
#include "google/cacheinvalidation/include/types.h"
#include "sync/notifier/invalidation_util.h"

namespace syncer {

SyncLogger::SyncLogger() {}
SyncLogger::~SyncLogger() {}

void SyncLogger::Log(LogLevel level, const char* file, int line,
                     const char* format, ...) {
  logging::LogSeverity log_severity = -2;  // VLOG(2)
  bool emit_log = false;
  switch (level) {
    case FINE_LEVEL:
      log_severity = -2;  // VLOG(2)
      emit_log = VLOG_IS_ON(2);
      break;
    case INFO_LEVEL:
      log_severity = -1;  // VLOG(1)
      emit_log = VLOG_IS_ON(1);
      break;
    case WARNING_LEVEL:
      log_severity = logging::LOG_WARNING;
      emit_log = LOG_IS_ON(WARNING);
      break;
    case SEVERE_LEVEL:
      log_severity = logging::LOG_ERROR;
      emit_log = LOG_IS_ON(ERROR);
      break;
  }
  if (emit_log) {
    va_list ap;
    va_start(ap, format);
    std::string result;
    base::StringAppendV(&result, format, ap);
    logging::LogMessage(file, line, log_severity).stream() << result;
    va_end(ap);
  }
}

void SyncLogger::SetSystemResources(invalidation::SystemResources* resources) {
  // Do nothing.
}

SyncInvalidationScheduler::SyncInvalidationScheduler()
    : created_on_loop_(base::MessageLoop::current()),
      is_started_(false),
      is_stopped_(false),
      weak_factory_(this) {
  CHECK(created_on_loop_);
}

SyncInvalidationScheduler::~SyncInvalidationScheduler() {
  CHECK_EQ(created_on_loop_, base::MessageLoop::current());
  CHECK(is_stopped_);
}

void SyncInvalidationScheduler::Start() {
  CHECK_EQ(created_on_loop_, base::MessageLoop::current());
  CHECK(!is_started_);
  is_started_ = true;
  is_stopped_ = false;
  weak_factory_.InvalidateWeakPtrs();
}

void SyncInvalidationScheduler::Stop() {
  CHECK_EQ(created_on_loop_, base::MessageLoop::current());
  is_stopped_ = true;
  is_started_ = false;
  weak_factory_.InvalidateWeakPtrs();
  STLDeleteElements(&posted_tasks_);
  posted_tasks_.clear();
}

void SyncInvalidationScheduler::Schedule(invalidation::TimeDelta delay,
                                         invalidation::Closure* task) {
  DCHECK(invalidation::IsCallbackRepeatable(task));
  CHECK_EQ(created_on_loop_, base::MessageLoop::current());

  if (!is_started_) {
    delete task;
    return;
  }

  posted_tasks_.insert(task);
  base::MessageLoop::current()->PostDelayedTask(
      FROM_HERE, base::Bind(&SyncInvalidationScheduler::RunPostedTask,
                            weak_factory_.GetWeakPtr(), task),
      delay);
}

bool SyncInvalidationScheduler::IsRunningOnThread() const {
  return created_on_loop_ == base::MessageLoop::current();
}

invalidation::Time SyncInvalidationScheduler::GetCurrentTime() const {
  CHECK_EQ(created_on_loop_, base::MessageLoop::current());
  return base::Time::Now();
}

void SyncInvalidationScheduler::SetSystemResources(
    invalidation::SystemResources* resources) {
  // Do nothing.
}

void SyncInvalidationScheduler::RunPostedTask(invalidation::Closure* task) {
  CHECK_EQ(created_on_loop_, base::MessageLoop::current());
  task->Run();
  posted_tasks_.erase(task);
  delete task;
}

SyncNetworkChannel::SyncNetworkChannel()
    : invalidator_state_(DEFAULT_INVALIDATION_ERROR),
      scheduling_hash_(0) {
}

SyncNetworkChannel::~SyncNetworkChannel() {
  STLDeleteElements(&network_status_receivers_);
}

void SyncNetworkChannel::SendMessage(const std::string& outgoing_message) {
  std::string encoded_message;
  EncodeMessage(&encoded_message, outgoing_message, service_context_,
      scheduling_hash_);
  SendEncodedMessage(encoded_message);
}

void SyncNetworkChannel::SetMessageReceiver(
    invalidation::MessageCallback* incoming_receiver) {
  incoming_receiver_.reset(incoming_receiver);
}

void SyncNetworkChannel::AddNetworkStatusReceiver(
    invalidation::NetworkStatusCallback* network_status_receiver) {
  network_status_receiver->Run(invalidator_state_ == INVALIDATIONS_ENABLED);
  network_status_receivers_.push_back(network_status_receiver);
}

void SyncNetworkChannel::SetSystemResources(
    invalidation::SystemResources* resources) {
  // Do nothing.
}

void SyncNetworkChannel::AddObserver(Observer* observer) {
  observers_.AddObserver(observer);
}

void SyncNetworkChannel::RemoveObserver(Observer* observer) {
  observers_.RemoveObserver(observer);
}

const std::string& SyncNetworkChannel::GetServiceContextForTest() const {
  return service_context_;
}

int64 SyncNetworkChannel::GetSchedulingHashForTest() const {
  return scheduling_hash_;
}

std::string SyncNetworkChannel::EncodeMessageForTest(
    const std::string& message, const std::string& service_context,
    int64 scheduling_hash) {
  std::string encoded_message;
  EncodeMessage(&encoded_message, message, service_context, scheduling_hash);
  return encoded_message;
}

bool SyncNetworkChannel::DecodeMessageForTest(
    const std::string& data,
    std::string* message,
    std::string* service_context,
    int64* scheduling_hash) {
  return DecodeMessage(data, message, service_context, scheduling_hash);
}

void SyncNetworkChannel::NotifyStateChange(InvalidatorState invalidator_state) {
  // Remember state for future NetworkStatusReceivers.
  invalidator_state_ = invalidator_state;
  // Notify NetworkStatusReceivers in cacheinvalidation.
  for (NetworkStatusReceiverList::const_iterator it =
           network_status_receivers_.begin();
       it != network_status_receivers_.end(); ++it) {
    (*it)->Run(invalidator_state_ == INVALIDATIONS_ENABLED);
  }
  // Notify observers.
  FOR_EACH_OBSERVER(Observer, observers_,
                    OnNetworkChannelStateChanged(invalidator_state_));
}

void SyncNetworkChannel::DeliverIncomingMessage(const std::string& data) {
  if (!incoming_receiver_) {
    DLOG(ERROR) << "No receiver for incoming notification";
    return;
  }
  std::string message;
  if (!DecodeMessage(data,
                     &message, &service_context_, &scheduling_hash_)) {
    DLOG(ERROR) << "Could not parse ClientGatewayMessage";
    return;
  }
  incoming_receiver_->Run(message);
}

void SyncNetworkChannel::EncodeMessage(
    std::string* encoded_message,
    const std::string& message,
    const std::string& service_context,
    int64 scheduling_hash) {
  ipc::invalidation::ClientGatewayMessage envelope;
  envelope.set_is_client_to_server(true);
  if (!service_context.empty()) {
    envelope.set_service_context(service_context);
    envelope.set_rpc_scheduling_hash(scheduling_hash);
  }
  envelope.set_network_message(message);
  envelope.SerializeToString(encoded_message);
}


bool SyncNetworkChannel::DecodeMessage(
    const std::string& data,
    std::string* message,
    std::string* service_context,
    int64* scheduling_hash) {
  ipc::invalidation::ClientGatewayMessage envelope;
  if (!envelope.ParseFromString(data)) {
    return false;
  }
  *message = envelope.network_message();
  if (envelope.has_service_context()) {
    *service_context = envelope.service_context();
  }
  if (envelope.has_rpc_scheduling_hash()) {
    *scheduling_hash = envelope.rpc_scheduling_hash();
  }
  return true;
}


SyncStorage::SyncStorage(StateWriter* state_writer,
                         invalidation::Scheduler* scheduler)
    : state_writer_(state_writer),
      scheduler_(scheduler) {
  DCHECK(state_writer_);
  DCHECK(scheduler_);
}

SyncStorage::~SyncStorage() {}

void SyncStorage::WriteKey(const std::string& key, const std::string& value,
                           invalidation::WriteKeyCallback* done) {
  CHECK(state_writer_);
  // TODO(ghc): actually write key,value associations, and don't invoke the
  // callback until the operation completes.
  state_writer_->WriteState(value);
  cached_state_ = value;
  // According to the cache invalidation API folks, we can do this as
  // long as we make sure to clear the persistent state that we start
  // up the cache invalidation client with.  However, we musn't do it
  // right away, as we may be called under a lock that the callback
  // uses.
  scheduler_->Schedule(
      invalidation::Scheduler::NoDelay(),
      invalidation::NewPermanentCallback(
          this, &SyncStorage::RunAndDeleteWriteKeyCallback,
          done));
}

void SyncStorage::ReadKey(const std::string& key,
                          invalidation::ReadKeyCallback* done) {
  DCHECK(scheduler_->IsRunningOnThread()) << "not running on scheduler thread";
  RunAndDeleteReadKeyCallback(done, cached_state_);
}

void SyncStorage::DeleteKey(const std::string& key,
                            invalidation::DeleteKeyCallback* done) {
  // TODO(ghc): Implement.
  LOG(WARNING) << "ignoring call to DeleteKey(" << key << ", callback)";
}

void SyncStorage::ReadAllKeys(invalidation::ReadAllKeysCallback* done) {
  // TODO(ghc): Implement.
  LOG(WARNING) << "ignoring call to ReadAllKeys(callback)";
}

void SyncStorage::SetSystemResources(
    invalidation::SystemResources* resources) {
  // Do nothing.
}

void SyncStorage::RunAndDeleteWriteKeyCallback(
    invalidation::WriteKeyCallback* callback) {
  callback->Run(
      invalidation::Status(invalidation::Status::SUCCESS, std::string()));
  delete callback;
}

void SyncStorage::RunAndDeleteReadKeyCallback(
    invalidation::ReadKeyCallback* callback, const std::string& value) {
  callback->Run(std::make_pair(
      invalidation::Status(invalidation::Status::SUCCESS, std::string()),
      value));
  delete callback;
}

SyncSystemResources::SyncSystemResources(
    SyncNetworkChannel* sync_network_channel,
    StateWriter* state_writer)
    : is_started_(false),
      logger_(new SyncLogger()),
      internal_scheduler_(new SyncInvalidationScheduler()),
      listener_scheduler_(new SyncInvalidationScheduler()),
      storage_(new SyncStorage(state_writer, internal_scheduler_.get())),
      sync_network_channel_(sync_network_channel) {
}

SyncSystemResources::~SyncSystemResources() {
  Stop();
}

void SyncSystemResources::Start() {
  internal_scheduler_->Start();
  listener_scheduler_->Start();
  is_started_ = true;
}

void SyncSystemResources::Stop() {
  internal_scheduler_->Stop();
  listener_scheduler_->Stop();
}

bool SyncSystemResources::IsStarted() const {
  return is_started_;
}

void SyncSystemResources::set_platform(const std::string& platform) {
  platform_ = platform;
}

std::string SyncSystemResources::platform() const {
  return platform_;
}

SyncLogger* SyncSystemResources::logger() {
  return logger_.get();
}

SyncStorage* SyncSystemResources::storage() {
  return storage_.get();
}

SyncNetworkChannel* SyncSystemResources::network() {
  return sync_network_channel_;
}

SyncInvalidationScheduler* SyncSystemResources::internal_scheduler() {
  return internal_scheduler_.get();
}

SyncInvalidationScheduler* SyncSystemResources::listener_scheduler() {
  return listener_scheduler_.get();
}

}  // namespace syncer