sync/notifier/non_blocking_invalidator.cc

   1 // Copyright (c) 2012 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 "sync/notifier/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 "jingle/notifier/listener/push_client.h"
  16 #include "sync/notifier/invalidation_notifier.h"
  17
  18 namespace syncer {
  19
  20 class NonBlockingInvalidator::Core
  21     : public base::RefCountedThreadSafe<NonBlockingInvalidator::Core>,
  22       // InvalidationHandler to observe the InvalidationNotifier we create.
  23       public InvalidationHandler {
  24  public:
  25   // Called on parent thread.  |delegate_observer| should be
  26   // initialized.
  27   explicit Core(
  28       const WeakHandle<InvalidationHandler>& delegate_observer);
  29
  30   // Helpers called on I/O thread.
  31   void Initialize(
  32       const notifier::NotifierOptions& notifier_options,
  33       const std::string& invalidator_client_id,
  34       const InvalidationStateMap& initial_invalidation_state_map,
  35       const std::string& invalidation_bootstrap_data,
  36       const WeakHandle<InvalidationStateTracker>& invalidation_state_tracker,
  37       const std::string& client_info);
  38   void Teardown();
  39   void UpdateRegisteredIds(const ObjectIdSet& ids);
  40   void Acknowledge(const invalidation::ObjectId& id,
  41                    const AckHandle& ack_handle);
  42   void UpdateCredentials(const std::string& email, const std::string& token);
  43
  44   // InvalidationHandler implementation (all called on I/O thread by
  45   // InvalidationNotifier).
  46   virtual void OnInvalidatorStateChange(InvalidatorState reason) OVERRIDE;
  47   virtual void OnIncomingInvalidation(
  48       const ObjectIdInvalidationMap& invalidation_map) OVERRIDE;
  49
  50  private:
  51   friend class
  52       base::RefCountedThreadSafe<NonBlockingInvalidator::Core>;
  53   // Called on parent or I/O thread.
  54   virtual ~Core();
  55
  56   // The variables below should be used only on the I/O thread.
  57   const WeakHandle<InvalidationHandler> delegate_observer_;
  58   scoped_ptr<InvalidationNotifier> invalidation_notifier_;
  59   scoped_refptr<base::SingleThreadTaskRunner> network_task_runner_;
  60
  61   DISALLOW_COPY_AND_ASSIGN(Core);
  62 };
  63
  64 NonBlockingInvalidator::Core::Core(
  65     const WeakHandle<InvalidationHandler>& delegate_observer)
  66     : delegate_observer_(delegate_observer) {
  67   DCHECK(delegate_observer_.IsInitialized());
  68 }
  69
  70 NonBlockingInvalidator::Core::~Core() {
  71 }
  72
  73 void NonBlockingInvalidator::Core::Initialize(
  74     const notifier::NotifierOptions& notifier_options,
  75     const std::string& invalidator_client_id,
  76     const InvalidationStateMap& initial_invalidation_state_map,
  77     const std::string& invalidation_bootstrap_data,
  78     const WeakHandle<InvalidationStateTracker>& invalidation_state_tracker,
  79     const std::string& client_info) {
  80   DCHECK(notifier_options.request_context_getter.get());
  81   DCHECK_EQ(notifier::NOTIFICATION_SERVER,
  82             notifier_options.notification_method);
  83   network_task_runner_ = notifier_options.request_context_getter->
  84       GetNetworkTaskRunner();
  85   DCHECK(network_task_runner_->BelongsToCurrentThread());
  86   invalidation_notifier_.reset(
  87       new InvalidationNotifier(
  88           notifier::PushClient::CreateDefaultOnIOThread(notifier_options),
  89           invalidator_client_id,
  90           initial_invalidation_state_map,
  91           invalidation_bootstrap_data,
  92           invalidation_state_tracker,
  93           client_info));
  94   invalidation_notifier_->RegisterHandler(this);
  95 }
  96
  97 void NonBlockingInvalidator::Core::Teardown() {
  98   DCHECK(network_task_runner_->BelongsToCurrentThread());
  99   invalidation_notifier_->UnregisterHandler(this);
 100   invalidation_notifier_.reset();
 101   network_task_runner_ = NULL;
 102 }
 103
 104 void NonBlockingInvalidator::Core::UpdateRegisteredIds(const ObjectIdSet& ids) {
 105   DCHECK(network_task_runner_->BelongsToCurrentThread());
 106   invalidation_notifier_->UpdateRegisteredIds(this, ids);
 107 }
 108
 109 void NonBlockingInvalidator::Core::Acknowledge(const invalidation::ObjectId& id,
 110                                                const AckHandle& ack_handle) {
 111   DCHECK(network_task_runner_->BelongsToCurrentThread());
 112   invalidation_notifier_->Acknowledge(id, ack_handle);
 113 }
 114
 115 void NonBlockingInvalidator::Core::UpdateCredentials(const std::string& email,
 116                                                      const std::string& token) {
 117   DCHECK(network_task_runner_->BelongsToCurrentThread());
 118   invalidation_notifier_->UpdateCredentials(email, token);
 119 }
 120
 121 void NonBlockingInvalidator::Core::OnInvalidatorStateChange(
 122     InvalidatorState reason) {
 123   DCHECK(network_task_runner_->BelongsToCurrentThread());
 124   delegate_observer_.Call(
 125       FROM_HERE, &InvalidationHandler::OnInvalidatorStateChange, reason);
 126 }
 127
 128 void NonBlockingInvalidator::Core::OnIncomingInvalidation(
 129     const ObjectIdInvalidationMap& invalidation_map) {
 130   DCHECK(network_task_runner_->BelongsToCurrentThread());
 131   delegate_observer_.Call(FROM_HERE,
 132                           &InvalidationHandler::OnIncomingInvalidation,
 133                           invalidation_map);
 134 }
 135
 136 NonBlockingInvalidator::NonBlockingInvalidator(
 137     const notifier::NotifierOptions& notifier_options,
 138     const std::string& invalidator_client_id,
 139     const InvalidationStateMap& initial_invalidation_state_map,
 140     const std::string& invalidation_bootstrap_data,
 141     const WeakHandle<InvalidationStateTracker>&
 142         invalidation_state_tracker,
 143     const std::string& client_info)
 144         : weak_ptr_factory_(this),
 145           core_(
 146               new Core(MakeWeakHandle(weak_ptr_factory_.GetWeakPtr()))),
 147           parent_task_runner_(
 148               base::ThreadTaskRunnerHandle::Get()),
 149           network_task_runner_(notifier_options.request_context_getter->
 150               GetNetworkTaskRunner()) {
 151   if (!network_task_runner_->PostTask(
 152           FROM_HERE,
 153           base::Bind(
 154               &NonBlockingInvalidator::Core::Initialize,
 155               core_.get(),
 156               notifier_options,
 157               invalidator_client_id,
 158               initial_invalidation_state_map,
 159               invalidation_bootstrap_data,
 160               invalidation_state_tracker,
 161               client_info))) {
 162     NOTREACHED();
 163   }
 164 }
 165
 166 NonBlockingInvalidator::~NonBlockingInvalidator() {
 167   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 168   if (!network_task_runner_->PostTask(
 169           FROM_HERE,
 170           base::Bind(&NonBlockingInvalidator::Core::Teardown,
 171                      core_.get()))) {
 172     DVLOG(1) << "Network thread stopped before invalidator is destroyed.";
 173   }
 174 }
 175
 176 void NonBlockingInvalidator::RegisterHandler(InvalidationHandler* handler) {
 177   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 178   registrar_.RegisterHandler(handler);
 179 }
 180
 181 void NonBlockingInvalidator::UpdateRegisteredIds(InvalidationHandler* handler,
 182                                                  const ObjectIdSet& ids) {
 183   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 184   registrar_.UpdateRegisteredIds(handler, ids);
 185   if (!network_task_runner_->PostTask(
 186           FROM_HERE,
 187           base::Bind(
 188               &NonBlockingInvalidator::Core::UpdateRegisteredIds,
 189               core_.get(),
 190               registrar_.GetAllRegisteredIds()))) {
 191     NOTREACHED();
 192   }
 193 }
 194
 195 void NonBlockingInvalidator::UnregisterHandler(InvalidationHandler* handler) {
 196   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 197   registrar_.UnregisterHandler(handler);
 198 }
 199
 200 void NonBlockingInvalidator::Acknowledge(const invalidation::ObjectId& id,
 201                                          const AckHandle& ack_handle) {
 202   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 203   if (!network_task_runner_->PostTask(
 204           FROM_HERE,
 205           base::Bind(
 206               &NonBlockingInvalidator::Core::Acknowledge,
 207               core_.get(),
 208               id,
 209               ack_handle))) {
 210     NOTREACHED();
 211   }
 212 }
 213
 214 InvalidatorState NonBlockingInvalidator::GetInvalidatorState() const {
 215   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 216   return registrar_.GetInvalidatorState();
 217 }
 218
 219 void NonBlockingInvalidator::UpdateCredentials(const std::string& email,
 220                                                const std::string& token) {
 221   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 222   if (!network_task_runner_->PostTask(
 223           FROM_HERE,
 224           base::Bind(&NonBlockingInvalidator::Core::UpdateCredentials,
 225                      core_.get(), email, token))) {
 226     NOTREACHED();
 227   }
 228 }
 229
 230 void NonBlockingInvalidator::OnInvalidatorStateChange(InvalidatorState state) {
 231   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 232   registrar_.UpdateInvalidatorState(state);
 233 }
 234
 235 void NonBlockingInvalidator::OnIncomingInvalidation(
 236         const ObjectIdInvalidationMap& invalidation_map) {
 237   DCHECK(parent_task_runner_->BelongsToCurrentThread());
 238   registrar_.DispatchInvalidationsToHandlers(invalidation_map);
 239 }
 240
 241 }  // namespace syncer