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