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