components/invalidation/non_blocking_invalidator.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/non_blocking_invalidator.h"
   6
   7 #include <cstddef>
   8
   9 #include "base/location.h"
  10 #include "base/logging.h"
  11 #include "base/memory/scoped_ptr.h"
  12 #include "base/profiler/scoped_tracker.h"
  13 #include "base/single_thread_task_runner.h"
  14 #include "base/thread_task_runner_handle.h"
  15 #include "base/threading/thread.h"
  16 #include "components/invalidation/gcm_network_channel_delegate.h"
  17 #include "components/invalidation/invalidation_handler.h"
  18 #include "components/invalidation/invalidation_notifier.h"
  19 #include "components/invalidation/object_id_invalidation_map.h"
  20 #include "components/invalidation/sync_system_resources.h"
  21 #include "jingle/notifier/listener/push_client.h"
  22
  23 namespace syncer {
  24
  25 struct NonBlockingInvalidator::InitializeOptions {
  26   InitializeOptions(
  27       NetworkChannelCreator network_channel_creator,
  28       const std::string& invalidator_client_id,
  29       const UnackedInvalidationsMap& saved_invalidations,
  30       const std::string& invalidation_bootstrap_data,
  31       const base::WeakPtr<InvalidationStateTracker>& invalidation_state_tracker,
  32       const scoped_refptr<base::SingleThreadTaskRunner>&
  33           invalidation_state_tracker_task_runner,
  34       const std::string& client_info,
  35       scoped_refptr<net::URLRequestContextGetter> request_context_getter)
  36       : network_channel_creator(network_channel_creator),
  37         invalidator_client_id(invalidator_client_id),
  38         saved_invalidations(saved_invalidations),
  39         invalidation_bootstrap_data(invalidation_bootstrap_data),
  40         invalidation_state_tracker(invalidation_state_tracker),
  41         invalidation_state_tracker_task_runner(
  42             invalidation_state_tracker_task_runner),
  43         client_info(client_info),
  44         request_context_getter(request_context_getter) {}
  45
  46   NetworkChannelCreator network_channel_creator;
  47   std::string invalidator_client_id;
  48   UnackedInvalidationsMap saved_invalidations;
  49   std::string invalidation_bootstrap_data;
  50   base::WeakPtr<InvalidationStateTracker> invalidation_state_tracker;
  51   scoped_refptr<base::SingleThreadTaskRunner>
  52       invalidation_state_tracker_task_runner;
  53   std::string client_info;
  54   scoped_refptr<net::URLRequestContextGetter> request_context_getter;
  55 };
  56
  57 namespace {
  58 // This class provides a wrapper for a logging class in order to receive
  59 // callbacks across threads, without having to worry about owner threads.
  60 class CallbackProxy {
  61  public:
  62   explicit CallbackProxy(
  63       base::Callback<void(const base::DictionaryValue&)> callback);
  64   ~CallbackProxy();
  65
  66   void Run(const base::DictionaryValue& value);
  67
  68  private:
  69   static void DoRun(base::Callback<void(const base::DictionaryValue&)> callback,
  70                     scoped_ptr<base::DictionaryValue> value);
  71
  72   base::Callback<void(const base::DictionaryValue&)> callback_;
  73   scoped_refptr<base::SingleThreadTaskRunner> running_thread_;
  74
  75   DISALLOW_COPY_AND_ASSIGN(CallbackProxy);
  76 };
  77
  78 CallbackProxy::CallbackProxy(
  79     base::Callback<void(const base::DictionaryValue&)> callback)
  80     : callback_(callback),
  81       running_thread_(base::ThreadTaskRunnerHandle::Get()) {}
  82
  83 CallbackProxy::~CallbackProxy() {}
  84
  85 void CallbackProxy::DoRun(
  86     base::Callback<void(const base::DictionaryValue&)> callback,
  87     scoped_ptr<base::DictionaryValue> value) {
  88   callback.Run(*value);
  89 }
  90
  91 void CallbackProxy::Run(const base::DictionaryValue& value) {
  92   scoped_ptr<base::DictionaryValue> copied(value.DeepCopy());
  93   running_thread_->PostTask(
  94       FROM_HERE,
  95       base::Bind(&CallbackProxy::DoRun, callback_, base::Passed(&copied)));
  96 }
  97 }
  98
  99 class NonBlockingInvalidator::Core
 100     : public base::RefCountedThreadSafe<NonBlockingInvalidator::Core>,
 101       // InvalidationHandler to observe the InvalidationNotifier we create.
 102       public InvalidationHandler {
 103  public:
 104   // Called on parent thread.  |delegate_observer| should be initialized.
 105   Core(const base::WeakPtr<NonBlockingInvalidator>& delegate_observer,
 106        const scoped_refptr<base::SingleThreadTaskRunner>&
 107            delegate_observer_task_runner);
 108
 109   // Helpers called on I/O thread.
 110   void Initialize(
 111       const NonBlockingInvalidator::InitializeOptions& initialize_options);
 112   void Teardown();
 113   void UpdateRegisteredIds(const ObjectIdSet& ids);
 114   void UpdateCredentials(const std::string& email, const std::string& token);
 115   void RequestDetailedStatus(
 116       base::Callback<void(const base::DictionaryValue&)> callback) const;
 117
 118   // InvalidationHandler implementation (all called on I/O thread by
 119   // InvalidationNotifier).
 120   void OnInvalidatorStateChange(InvalidatorState reason) override;
 121   void OnIncomingInvalidation(
 122       const ObjectIdInvalidationMap& invalidation_map) override;
 123   std::string GetOwnerName() const override;
 124
 125  private:
 126   friend class
 127       base::RefCountedThreadSafe<NonBlockingInvalidator::Core>;
 128   // Called on parent or I/O thread.
 129   ~Core() override;
 130
 131   // The variables below should be used only on the I/O thread.
 132   const base::WeakPtr<NonBlockingInvalidator> delegate_observer_;
 133   scoped_refptr<base::SingleThreadTaskRunner> delegate_observer_task_runner_;
 134   scoped_ptr<InvalidationNotifier> invalidation_notifier_;
 135   scoped_refptr<base::SingleThreadTaskRunner> network_task_runner_;
 136
 137   DISALLOW_COPY_AND_ASSIGN(Core);
 138 };
 139
 140 NonBlockingInvalidator::Core::Core(
 141     const base::WeakPtr<NonBlockingInvalidator>& delegate_observer,
 142     const scoped_refptr<base::SingleThreadTaskRunner>&
 143         delegate_observer_task_runner)
 144     : delegate_observer_(delegate_observer),
 145       delegate_observer_task_runner_(delegate_observer_task_runner) {
 146   DCHECK(delegate_observer_);
 147   DCHECK(delegate_observer_task_runner_.get());
 148 }
 149
 150 NonBlockingInvalidator::Core::~Core() {
 151 }
 152
 153 void NonBlockingInvalidator::Core::Initialize(
 154     const NonBlockingInvalidator::InitializeOptions& initialize_options) {
 155   DCHECK(initialize_options.request_context_getter.get());
 156   network_task_runner_ =
 157       initialize_options.request_context_getter->GetNetworkTaskRunner();
 158   DCHECK(network_task_runner_->BelongsToCurrentThread());
 159   scoped_ptr<SyncNetworkChannel> network_channel =
 160       initialize_options.network_channel_creator.Run();
 161   invalidation_notifier_.reset(new InvalidationNotifier(
 162       network_channel.Pass(),
 163       initialize_options.invalidator_client_id,
 164       initialize_options.saved_invalidations,
 165       initialize_options.invalidation_bootstrap_data,
 166       initialize_options.invalidation_state_tracker,
 167       initialize_options.invalidation_state_tracker_task_runner,
 168       initialize_options.client_info));
 169   invalidation_notifier_->RegisterHandler(this);
 170 }
 171
 172 void NonBlockingInvalidator::Core::Teardown() {
 173   DCHECK(network_task_runner_->BelongsToCurrentThread());
 174   invalidation_notifier_->UnregisterHandler(this);
 175   invalidation_notifier_.reset();
 176   network_task_runner_ = NULL;
 177 }
 178
 179 void NonBlockingInvalidator::Core::UpdateRegisteredIds(const ObjectIdSet& ids) {
 180   DCHECK(network_task_runner_->BelongsToCurrentThread());
 181   invalidation_notifier_->UpdateRegisteredIds(this, ids);
 182 }
 183
 184 void NonBlockingInvalidator::Core::UpdateCredentials(const std::string& email,
 185                                                      const std::string& token) {
 186   // TODO(pkasting): Remove ScopedTracker below once crbug.com/477117 is fixed.
 187   tracked_objects::ScopedTracker tracking_profile(
 188       FROM_HERE_WITH_EXPLICIT_FUNCTION(
 189           "477117 NonBlockingInvalidator::Core::UpdateCredentials"));
 190   DCHECK(network_task_runner_->BelongsToCurrentThread());
 191   invalidation_notifier_->UpdateCredentials(email, token);
 192 }
 193
 194 void NonBlockingInvalidator::Core::RequestDetailedStatus(
 195     base::Callback<void(const base::DictionaryValue&)> callback) const {
 196   DCHECK(network_task_runner_->BelongsToCurrentThread());
 197   invalidation_notifier_->RequestDetailedStatus(callback);
 198 }
 199
 200 void NonBlockingInvalidator::Core::OnInvalidatorStateChange(
 201     InvalidatorState reason) {
 202   DCHECK(network_task_runner_->BelongsToCurrentThread());
 203   delegate_observer_task_runner_->PostTask(
 204       FROM_HERE,
 205       base::Bind(&NonBlockingInvalidator::OnInvalidatorStateChange,
 206                  delegate_observer_,
 207                  reason));
 208 }
 209
 210 void NonBlockingInvalidator::Core::OnIncomingInvalidation(
 211     const ObjectIdInvalidationMap& invalidation_map) {
 212   DCHECK(network_task_runner_->BelongsToCurrentThread());
 213   delegate_observer_task_runner_->PostTask(
 214       FROM_HERE,
 215       base::Bind(&NonBlockingInvalidator::OnIncomingInvalidation,
 216                  delegate_observer_,
 217                  invalidation_map));
 218 }
 219
 220 std::string NonBlockingInvalidator::Core::GetOwnerName() const {
 221   return "Sync";
 222 }
 223
 224 NonBlockingInvalidator::NonBlockingInvalidator(
 225     NetworkChannelCreator network_channel_creator,
 226     const std::string& invalidator_client_id,
 227     const UnackedInvalidationsMap& saved_invalidations,
 228     const std::string& invalidation_bootstrap_data,
 229     InvalidationStateTracker* invalidation_state_tracker,
 230     const std::string& client_info,
 231     const scoped_refptr<net::URLRequestContextGetter>& request_context_getter)
 232     : invalidation_state_tracker_(invalidation_state_tracker),
 233       parent_task_runner_(base::ThreadTaskRunnerHandle::Get()),
 234       network_task_runner_(request_context_getter->GetNetworkTaskRunner()),
 235       weak_ptr_factory_(this) {
 236   base::WeakPtr<NonBlockingInvalidator> weak_ptr_this =
 237       weak_ptr_factory_.GetWeakPtr();
 238   weak_ptr_this.get();  // Bind to this thread.
 239
 240   core_ = new Core(weak_ptr_this,
 241                    base::MessageLoopProxy::current());
 242
 243   InitializeOptions initialize_options(network_channel_creator,
 244                                        invalidator_client_id,
 245                                        saved_invalidations,
 246                                        invalidation_bootstrap_data,
 247                                        weak_ptr_this,
 248                                        base::MessageLoopProxy::current(),
 249                                        client_info,
 250                                        request_context_getter);
 251
 252   if (!network_task_runner_->PostTask(
 253           FROM_HERE,
 254           base::Bind(
 255               &NonBlockingInvalidator::Core::Initialize,
 256               core_.get(),
 257               initialize_options))) {
 258     NOTREACHED();
 259   }
 260 }
 261
 262 NonBlockingInvalidator::~NonBlockingInvalidator() {
 263   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 264   if (!network_task_runner_->PostTask(
 265           FROM_HERE,
 266           base::Bind(&NonBlockingInvalidator::Core::Teardown,
 267                      core_.get()))) {
 268     DVLOG(1) << "Network thread stopped before invalidator is destroyed.";
 269   }
 270 }
 271
 272 void NonBlockingInvalidator::RegisterHandler(InvalidationHandler* handler) {
 273   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 274   registrar_.RegisterHandler(handler);
 275 }
 276
 277 bool NonBlockingInvalidator::UpdateRegisteredIds(InvalidationHandler* handler,
 278                                                  const ObjectIdSet& ids) {
 279   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 280   if (!registrar_.UpdateRegisteredIds(handler, ids))
 281     return false;
 282   if (!network_task_runner_->PostTask(
 283           FROM_HERE,
 284           base::Bind(
 285               &NonBlockingInvalidator::Core::UpdateRegisteredIds,
 286               core_.get(),
 287               registrar_.GetAllRegisteredIds()))) {
 288     NOTREACHED();
 289   }
 290   return true;
 291 }
 292
 293 void NonBlockingInvalidator::UnregisterHandler(InvalidationHandler* handler) {
 294   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 295   registrar_.UnregisterHandler(handler);
 296 }
 297
 298 InvalidatorState NonBlockingInvalidator::GetInvalidatorState() const {
 299   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 300   return registrar_.GetInvalidatorState();
 301 }
 302
 303 void NonBlockingInvalidator::UpdateCredentials(const std::string& email,
 304                                                const std::string& token) {
 305   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 306   if (!network_task_runner_->PostTask(
 307           FROM_HERE,
 308           base::Bind(&NonBlockingInvalidator::Core::UpdateCredentials,
 309                      core_.get(), email, token))) {
 310     NOTREACHED();
 311   }
 312 }
 313
 314 void NonBlockingInvalidator::RequestDetailedStatus(
 315     base::Callback<void(const base::DictionaryValue&)> callback) const {
 316   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 317   base::Callback<void(const base::DictionaryValue&)> proxy_callback =
 318       base::Bind(&CallbackProxy::Run, base::Owned(new CallbackProxy(callback)));
 319   if (!network_task_runner_->PostTask(
 320           FROM_HERE,
 321           base::Bind(&NonBlockingInvalidator::Core::RequestDetailedStatus,
 322                      core_.get(),
 323                      proxy_callback))) {
 324     NOTREACHED();
 325   }
 326 }
 327
 328 NetworkChannelCreator
 329     NonBlockingInvalidator::MakePushClientChannelCreator(
 330         const notifier::NotifierOptions& notifier_options) {
 331   return base::Bind(SyncNetworkChannel::CreatePushClientChannel,
 332       notifier_options);
 333 }
 334
 335 NetworkChannelCreator NonBlockingInvalidator::MakeGCMNetworkChannelCreator(
 336     scoped_refptr<net::URLRequestContextGetter> request_context_getter,
 337     scoped_ptr<GCMNetworkChannelDelegate> delegate) {
 338   return base::Bind(&SyncNetworkChannel::CreateGCMNetworkChannel,
 339                     request_context_getter,
 340                     base::Passed(&delegate));
 341 }
 342
 343 void NonBlockingInvalidator::ClearAndSetNewClientId(const std::string& data) {
 344   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 345   invalidation_state_tracker_->ClearAndSetNewClientId(data);
 346 }
 347
 348 std::string NonBlockingInvalidator::GetInvalidatorClientId() const {
 349   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 350   return invalidation_state_tracker_->GetInvalidatorClientId();
 351 }
 352
 353 void NonBlockingInvalidator::SetBootstrapData(const std::string& data) {
 354   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 355   invalidation_state_tracker_->SetBootstrapData(data);
 356 }
 357
 358 std::string NonBlockingInvalidator::GetBootstrapData() const {
 359   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 360   return invalidation_state_tracker_->GetBootstrapData();
 361 }
 362
 363 void NonBlockingInvalidator::SetSavedInvalidations(
 364       const UnackedInvalidationsMap& states) {
 365   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 366   invalidation_state_tracker_->SetSavedInvalidations(states);
 367 }
 368
 369 UnackedInvalidationsMap NonBlockingInvalidator::GetSavedInvalidations() const {
 370   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 371   return invalidation_state_tracker_->GetSavedInvalidations();
 372 }
 373
 374 void NonBlockingInvalidator::Clear() {
 375   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 376   invalidation_state_tracker_->Clear();
 377 }
 378
 379 void NonBlockingInvalidator::OnInvalidatorStateChange(InvalidatorState state) {
 380   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 381   registrar_.UpdateInvalidatorState(state);
 382 }
 383
 384 void NonBlockingInvalidator::OnIncomingInvalidation(
 385         const ObjectIdInvalidationMap& invalidation_map) {
 386   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 387   registrar_.DispatchInvalidationsToHandlers(invalidation_map);
 388 }
 389
 390 std::string NonBlockingInvalidator::GetOwnerName() const { return "Sync"; }
 391
 392 }  // namespace syncer