components/invalidation/impl/sync_system_resources.cc

   1 // Copyright 2014 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 "components/invalidation/impl/sync_system_resources.h"
   6
   7 #include <cstdlib>
   8 #include <cstring>
   9 #include <string>
  10
  11 #include "base/bind.h"
  12 #include "base/location.h"
  13 #include "base/logging.h"
  14 #include "base/single_thread_task_runner.h"
  15 #include "base/stl_util.h"
  16 #include "base/strings/string_util.h"
  17 #include "base/strings/stringprintf.h"
  18 #include "base/thread_task_runner_handle.h"
  19 #include "components/invalidation/impl/gcm_network_channel.h"
  20 #include "components/invalidation/impl/gcm_network_channel_delegate.h"
  21 #include "components/invalidation/impl/push_client_channel.h"
  22 #include "components/invalidation/public/invalidation_util.h"
  23 #include "google/cacheinvalidation/deps/callback.h"
  24 #include "google/cacheinvalidation/include/types.h"
  25 #include "jingle/notifier/listener/push_client.h"
  26
  27 namespace syncer {
  28
  29 SyncLogger::SyncLogger() {}
  30 SyncLogger::~SyncLogger() {}
  31
  32 void SyncLogger::Log(LogLevel level, const char* file, int line,
  33                      const char* format, ...) {
  34   logging::LogSeverity log_severity = -2;  // VLOG(2)
  35   bool emit_log = false;
  36   switch (level) {
  37     case FINE_LEVEL:
  38       log_severity = -2;  // VLOG(2)
  39       emit_log = VLOG_IS_ON(2);
  40       break;
  41     case INFO_LEVEL:
  42       log_severity = -1;  // VLOG(1)
  43       emit_log = VLOG_IS_ON(1);
  44       break;
  45     case WARNING_LEVEL:
  46       log_severity = logging::LOG_WARNING;
  47       emit_log = LOG_IS_ON(WARNING);
  48       break;
  49     case SEVERE_LEVEL:
  50       log_severity = logging::LOG_ERROR;
  51       emit_log = LOG_IS_ON(ERROR);
  52       break;
  53   }
  54   if (emit_log) {
  55     va_list ap;
  56     va_start(ap, format);
  57     std::string result;
  58     base::StringAppendV(&result, format, ap);
  59     logging::LogMessage(file, line, log_severity).stream() << result;
  60     va_end(ap);
  61   }
  62 }
  63
  64 void SyncLogger::SetSystemResources(invalidation::SystemResources* resources) {
  65   // Do nothing.
  66 }
  67
  68 SyncInvalidationScheduler::SyncInvalidationScheduler()
  69     : created_on_loop_(base::MessageLoop::current()),
  70       is_started_(false),
  71       is_stopped_(false),
  72       weak_factory_(this) {
  73   CHECK(created_on_loop_);
  74 }
  75
  76 SyncInvalidationScheduler::~SyncInvalidationScheduler() {
  77   CHECK_EQ(created_on_loop_, base::MessageLoop::current());
  78   CHECK(is_stopped_);
  79 }
  80
  81 void SyncInvalidationScheduler::Start() {
  82   CHECK_EQ(created_on_loop_, base::MessageLoop::current());
  83   CHECK(!is_started_);
  84   is_started_ = true;
  85   is_stopped_ = false;
  86   weak_factory_.InvalidateWeakPtrs();
  87 }
  88
  89 void SyncInvalidationScheduler::Stop() {
  90   CHECK_EQ(created_on_loop_, base::MessageLoop::current());
  91   is_stopped_ = true;
  92   is_started_ = false;
  93   weak_factory_.InvalidateWeakPtrs();
  94   STLDeleteElements(&posted_tasks_);
  95 }
  96
  97 void SyncInvalidationScheduler::Schedule(invalidation::TimeDelta delay,
  98                                          invalidation::Closure* task) {
  99   DCHECK(invalidation::IsCallbackRepeatable(task));
 100   CHECK_EQ(created_on_loop_, base::MessageLoop::current());
 101
 102   if (!is_started_) {
 103     delete task;
 104     return;
 105   }
 106
 107   posted_tasks_.insert(task);
 108   base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
 109       FROM_HERE, base::Bind(&SyncInvalidationScheduler::RunPostedTask,
 110                             weak_factory_.GetWeakPtr(), task),
 111       delay);
 112 }
 113
 114 bool SyncInvalidationScheduler::IsRunningOnThread() const {
 115   return created_on_loop_ == base::MessageLoop::current();
 116 }
 117
 118 invalidation::Time SyncInvalidationScheduler::GetCurrentTime() const {
 119   CHECK_EQ(created_on_loop_, base::MessageLoop::current());
 120   return base::Time::Now();
 121 }
 122
 123 void SyncInvalidationScheduler::SetSystemResources(
 124     invalidation::SystemResources* resources) {
 125   // Do nothing.
 126 }
 127
 128 void SyncInvalidationScheduler::RunPostedTask(invalidation::Closure* task) {
 129   CHECK_EQ(created_on_loop_, base::MessageLoop::current());
 130   task->Run();
 131   posted_tasks_.erase(task);
 132   delete task;
 133 }
 134
 135 SyncNetworkChannel::SyncNetworkChannel()
 136     : last_network_status_(false),
 137       received_messages_count_(0) {}
 138
 139 SyncNetworkChannel::~SyncNetworkChannel() {
 140   STLDeleteElements(&network_status_receivers_);
 141 }
 142
 143 void SyncNetworkChannel::SetMessageReceiver(
 144     invalidation::MessageCallback* incoming_receiver) {
 145   incoming_receiver_.reset(incoming_receiver);
 146 }
 147
 148 void SyncNetworkChannel::AddNetworkStatusReceiver(
 149     invalidation::NetworkStatusCallback* network_status_receiver) {
 150   network_status_receiver->Run(last_network_status_);
 151   network_status_receivers_.push_back(network_status_receiver);
 152 }
 153
 154 void SyncNetworkChannel::SetSystemResources(
 155     invalidation::SystemResources* resources) {
 156   // Do nothing.
 157 }
 158
 159 void SyncNetworkChannel::AddObserver(Observer* observer) {
 160   observers_.AddObserver(observer);
 161 }
 162
 163 void SyncNetworkChannel::RemoveObserver(Observer* observer) {
 164   observers_.RemoveObserver(observer);
 165 }
 166
 167 scoped_ptr<SyncNetworkChannel> SyncNetworkChannel::CreatePushClientChannel(
 168     const notifier::NotifierOptions& notifier_options) {
 169   scoped_ptr<notifier::PushClient> push_client(
 170       notifier::PushClient::CreateDefaultOnIOThread(notifier_options));
 171   return scoped_ptr<SyncNetworkChannel>(
 172       new PushClientChannel(push_client.Pass()));
 173 }
 174
 175 scoped_ptr<SyncNetworkChannel> SyncNetworkChannel::CreateGCMNetworkChannel(
 176     scoped_refptr<net::URLRequestContextGetter> request_context_getter,
 177     scoped_ptr<GCMNetworkChannelDelegate> delegate) {
 178   return scoped_ptr<SyncNetworkChannel>(new GCMNetworkChannel(
 179       request_context_getter, delegate.Pass()));
 180 }
 181
 182 void SyncNetworkChannel::NotifyNetworkStatusChange(bool online) {
 183   // Remember network state for future NetworkStatusReceivers.
 184   last_network_status_ = online;
 185   // Notify NetworkStatusReceivers in cacheinvalidation.
 186   for (NetworkStatusReceiverList::const_iterator it =
 187            network_status_receivers_.begin();
 188        it != network_status_receivers_.end(); ++it) {
 189     (*it)->Run(online);
 190   }
 191 }
 192
 193 void SyncNetworkChannel::NotifyChannelStateChange(
 194     InvalidatorState invalidator_state) {
 195   FOR_EACH_OBSERVER(Observer, observers_,
 196                     OnNetworkChannelStateChanged(invalidator_state));
 197 }
 198
 199 bool SyncNetworkChannel::DeliverIncomingMessage(const std::string& message) {
 200   if (!incoming_receiver_) {
 201     DLOG(ERROR) << "No receiver for incoming notification";
 202     return false;
 203   }
 204   received_messages_count_++;
 205   incoming_receiver_->Run(message);
 206   return true;
 207 }
 208
 209 int SyncNetworkChannel::GetReceivedMessagesCount() const {
 210   return received_messages_count_;
 211 }
 212
 213 SyncStorage::SyncStorage(StateWriter* state_writer,
 214                          invalidation::Scheduler* scheduler)
 215     : state_writer_(state_writer),
 216       scheduler_(scheduler) {
 217   DCHECK(state_writer_);
 218   DCHECK(scheduler_);
 219 }
 220
 221 SyncStorage::~SyncStorage() {}
 222
 223 void SyncStorage::WriteKey(const std::string& key, const std::string& value,
 224                            invalidation::WriteKeyCallback* done) {
 225   CHECK(state_writer_);
 226   // TODO(ghc): actually write key,value associations, and don't invoke the
 227   // callback until the operation completes.
 228   state_writer_->WriteState(value);
 229   cached_state_ = value;
 230   // According to the cache invalidation API folks, we can do this as
 231   // long as we make sure to clear the persistent state that we start
 232   // up the cache invalidation client with.  However, we musn't do it
 233   // right away, as we may be called under a lock that the callback
 234   // uses.
 235   scheduler_->Schedule(
 236       invalidation::Scheduler::NoDelay(),
 237       invalidation::NewPermanentCallback(
 238           this, &SyncStorage::RunAndDeleteWriteKeyCallback,
 239           done));
 240 }
 241
 242 void SyncStorage::ReadKey(const std::string& key,
 243                           invalidation::ReadKeyCallback* done) {
 244   DCHECK(scheduler_->IsRunningOnThread()) << "not running on scheduler thread";
 245   RunAndDeleteReadKeyCallback(done, cached_state_);
 246 }
 247
 248 void SyncStorage::DeleteKey(const std::string& key,
 249                             invalidation::DeleteKeyCallback* done) {
 250   // TODO(ghc): Implement.
 251   LOG(WARNING) << "ignoring call to DeleteKey(" << key << ", callback)";
 252 }
 253
 254 void SyncStorage::ReadAllKeys(invalidation::ReadAllKeysCallback* done) {
 255   // TODO(ghc): Implement.
 256   LOG(WARNING) << "ignoring call to ReadAllKeys(callback)";
 257 }
 258
 259 void SyncStorage::SetSystemResources(
 260     invalidation::SystemResources* resources) {
 261   // Do nothing.
 262 }
 263
 264 void SyncStorage::RunAndDeleteWriteKeyCallback(
 265     invalidation::WriteKeyCallback* callback) {
 266   callback->Run(
 267       invalidation::Status(invalidation::Status::SUCCESS, std::string()));
 268   delete callback;
 269 }
 270
 271 void SyncStorage::RunAndDeleteReadKeyCallback(
 272     invalidation::ReadKeyCallback* callback, const std::string& value) {
 273   callback->Run(std::make_pair(
 274       invalidation::Status(invalidation::Status::SUCCESS, std::string()),
 275       value));
 276   delete callback;
 277 }
 278
 279 SyncSystemResources::SyncSystemResources(
 280     SyncNetworkChannel* sync_network_channel,
 281     StateWriter* state_writer)
 282     : is_started_(false),
 283       logger_(new SyncLogger()),
 284       internal_scheduler_(new SyncInvalidationScheduler()),
 285       listener_scheduler_(new SyncInvalidationScheduler()),
 286       storage_(new SyncStorage(state_writer, internal_scheduler_.get())),
 287       sync_network_channel_(sync_network_channel) {
 288 }
 289
 290 SyncSystemResources::~SyncSystemResources() {
 291   Stop();
 292 }
 293
 294 void SyncSystemResources::Start() {
 295   internal_scheduler_->Start();
 296   listener_scheduler_->Start();
 297   is_started_ = true;
 298 }
 299
 300 void SyncSystemResources::Stop() {
 301   internal_scheduler_->Stop();
 302   listener_scheduler_->Stop();
 303 }
 304
 305 bool SyncSystemResources::IsStarted() const {
 306   return is_started_;
 307 }
 308
 309 void SyncSystemResources::set_platform(const std::string& platform) {
 310   platform_ = platform;
 311 }
 312
 313 std::string SyncSystemResources::platform() const {
 314   return platform_;
 315 }
 316
 317 SyncLogger* SyncSystemResources::logger() {
 318   return logger_.get();
 319 }
 320
 321 SyncStorage* SyncSystemResources::storage() {
 322   return storage_.get();
 323 }
 324
 325 SyncNetworkChannel* SyncSystemResources::network() {
 326   return sync_network_channel_;
 327 }
 328
 329 SyncInvalidationScheduler* SyncSystemResources::internal_scheduler() {
 330   return internal_scheduler_.get();
 331 }
 332
 333 SyncInvalidationScheduler* SyncSystemResources::listener_scheduler() {
 334   return listener_scheduler_.get();
 335 }
 336
 337 }  // namespace syncer