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