// 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