src/jingle/notifier/listener/non_blocking_push_client.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 "jingle/notifier/listener/non_blocking_push_client.h"
   6
   7 #include "base/bind.h"
   8 #include "base/location.h"
   9 #include "base/logging.h"
  10 #include "base/message_loop/message_loop_proxy.h"
  11 #include "jingle/notifier/listener/push_client_observer.h"
  12
  13 namespace notifier {
  14
  15 // All methods are called on the delegate thread unless specified
  16 // otherwise.
  17 class NonBlockingPushClient::Core
  18     : public base::RefCountedThreadSafe<NonBlockingPushClient::Core>,
  19       public PushClientObserver {
  20  public:
  21   // Called on the parent thread.
  22   explicit Core(
  23       const scoped_refptr<base::SingleThreadTaskRunner>&
  24           delegate_task_runner,
  25       const base::WeakPtr<NonBlockingPushClient>& parent_push_client);
  26
  27   // Must be called after being created.
  28   //
  29   // This is separated out from the constructor since posting tasks
  30   // from the constructor is dangerous.
  31   void CreateOnDelegateThread(
  32       const CreateBlockingPushClientCallback&
  33           create_blocking_push_client_callback);
  34
  35   // Must be called before being destroyed.
  36   void DestroyOnDelegateThread();
  37
  38   void UpdateSubscriptions(const SubscriptionList& subscriptions);
  39   void UpdateCredentials(const std::string& email, const std::string& token);
  40   void SendNotification(const Notification& data);
  41   void SendPing();
  42
  43   virtual void OnNotificationsEnabled() OVERRIDE;
  44   virtual void OnNotificationsDisabled(
  45       NotificationsDisabledReason reason) OVERRIDE;
  46   virtual void OnIncomingNotification(
  47       const Notification& notification) OVERRIDE;
  48   virtual void OnPingResponse() OVERRIDE;
  49
  50  private:
  51   friend class base::RefCountedThreadSafe<NonBlockingPushClient::Core>;
  52
  53   // Called on either the parent thread or the delegate thread.
  54   virtual ~Core();
  55
  56   const scoped_refptr<base::SingleThreadTaskRunner> parent_task_runner_;
  57   const scoped_refptr<base::SingleThreadTaskRunner> delegate_task_runner_;
  58
  59   const base::WeakPtr<NonBlockingPushClient> parent_push_client_;
  60   scoped_ptr<PushClient> delegate_push_client_;
  61
  62   DISALLOW_COPY_AND_ASSIGN(Core);
  63 };
  64
  65 NonBlockingPushClient::Core::Core(
  66     const scoped_refptr<base::SingleThreadTaskRunner>& delegate_task_runner,
  67     const base::WeakPtr<NonBlockingPushClient>& parent_push_client)
  68     : parent_task_runner_(base::MessageLoopProxy::current()),
  69       delegate_task_runner_(delegate_task_runner),
  70       parent_push_client_(parent_push_client) {}
  71
  72 NonBlockingPushClient::Core::~Core() {
  73   DCHECK(parent_task_runner_->BelongsToCurrentThread() ||
  74          delegate_task_runner_->BelongsToCurrentThread());
  75   DCHECK(!delegate_push_client_.get());
  76 }
  77
  78 void NonBlockingPushClient::Core::CreateOnDelegateThread(
  79     const CreateBlockingPushClientCallback&
  80         create_blocking_push_client_callback) {
  81   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
  82   DCHECK(!delegate_push_client_.get());
  83   delegate_push_client_ = create_blocking_push_client_callback.Run();
  84   delegate_push_client_->AddObserver(this);
  85 }
  86
  87 void NonBlockingPushClient::Core::DestroyOnDelegateThread() {
  88   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
  89   DCHECK(delegate_push_client_.get());
  90   delegate_push_client_->RemoveObserver(this);
  91   delegate_push_client_.reset();
  92 }
  93
  94 void NonBlockingPushClient::Core::UpdateSubscriptions(
  95     const SubscriptionList& subscriptions) {
  96   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
  97   DCHECK(delegate_push_client_.get());
  98   delegate_push_client_->UpdateSubscriptions(subscriptions);
  99 }
 100
 101 void NonBlockingPushClient::Core::UpdateCredentials(
 102       const std::string& email, const std::string& token) {
 103   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
 104   DCHECK(delegate_push_client_.get());
 105   delegate_push_client_->UpdateCredentials(email, token);
 106 }
 107
 108 void NonBlockingPushClient::Core::SendNotification(
 109     const Notification& notification) {
 110   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
 111   DCHECK(delegate_push_client_.get());
 112   delegate_push_client_->SendNotification(notification);
 113 }
 114
 115 void NonBlockingPushClient::Core::SendPing() {
 116   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
 117   DCHECK(delegate_push_client_.get());
 118   delegate_push_client_->SendPing();
 119 }
 120
 121 void NonBlockingPushClient::Core::OnNotificationsEnabled() {
 122   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
 123   parent_task_runner_->PostTask(
 124       FROM_HERE,
 125       base::Bind(&NonBlockingPushClient::OnNotificationsEnabled,
 126                  parent_push_client_));
 127 }
 128
 129 void NonBlockingPushClient::Core::OnNotificationsDisabled(
 130     NotificationsDisabledReason reason) {
 131   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
 132   parent_task_runner_->PostTask(
 133       FROM_HERE,
 134       base::Bind(&NonBlockingPushClient::OnNotificationsDisabled,
 135                  parent_push_client_, reason));
 136 }
 137
 138 void NonBlockingPushClient::Core::OnIncomingNotification(
 139     const Notification& notification) {
 140   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
 141   parent_task_runner_->PostTask(
 142       FROM_HERE,
 143       base::Bind(&NonBlockingPushClient::OnIncomingNotification,
 144                  parent_push_client_, notification));
 145 }
 146
 147 void NonBlockingPushClient::Core::OnPingResponse() {
 148   DCHECK(delegate_task_runner_->BelongsToCurrentThread());
 149   parent_task_runner_->PostTask(
 150       FROM_HERE,
 151       base::Bind(&NonBlockingPushClient::OnPingResponse,
 152                  parent_push_client_));
 153 }
 154
 155 NonBlockingPushClient::NonBlockingPushClient(
 156     const scoped_refptr<base::SingleThreadTaskRunner>& delegate_task_runner,
 157     const CreateBlockingPushClientCallback&
 158         create_blocking_push_client_callback)
 159     : weak_ptr_factory_(this),
 160       delegate_task_runner_(delegate_task_runner),
 161       core_(new Core(delegate_task_runner_,
 162                      weak_ptr_factory_.GetWeakPtr())) {
 163   delegate_task_runner_->PostTask(
 164       FROM_HERE,
 165       base::Bind(&NonBlockingPushClient::Core::CreateOnDelegateThread,
 166                  core_.get(), create_blocking_push_client_callback));
 167 }
 168
 169 NonBlockingPushClient::~NonBlockingPushClient() {
 170   DCHECK(thread_checker_.CalledOnValidThread());
 171   delegate_task_runner_->PostTask(
 172       FROM_HERE,
 173       base::Bind(&NonBlockingPushClient::Core::DestroyOnDelegateThread,
 174                  core_.get()));
 175 }
 176
 177 void NonBlockingPushClient::AddObserver(PushClientObserver* observer) {
 178   DCHECK(thread_checker_.CalledOnValidThread());
 179   observers_.AddObserver(observer);
 180 }
 181
 182 void NonBlockingPushClient::RemoveObserver(PushClientObserver* observer) {
 183   DCHECK(thread_checker_.CalledOnValidThread());
 184   observers_.RemoveObserver(observer);
 185 }
 186
 187 void NonBlockingPushClient::UpdateSubscriptions(
 188     const SubscriptionList& subscriptions) {
 189   DCHECK(thread_checker_.CalledOnValidThread());
 190   delegate_task_runner_->PostTask(
 191       FROM_HERE,
 192       base::Bind(&NonBlockingPushClient::Core::UpdateSubscriptions,
 193                  core_.get(), subscriptions));
 194 }
 195
 196 void NonBlockingPushClient::UpdateCredentials(
 197     const std::string& email, const std::string& token) {
 198   DCHECK(thread_checker_.CalledOnValidThread());
 199   delegate_task_runner_->PostTask(
 200       FROM_HERE,
 201       base::Bind(&NonBlockingPushClient::Core::UpdateCredentials,
 202                  core_.get(), email, token));
 203 }
 204
 205 void NonBlockingPushClient::SendNotification(
 206     const Notification& notification) {
 207   DCHECK(thread_checker_.CalledOnValidThread());
 208   delegate_task_runner_->PostTask(
 209       FROM_HERE,
 210       base::Bind(&NonBlockingPushClient::Core::SendNotification, core_.get(),
 211                  notification));
 212 }
 213
 214 void NonBlockingPushClient::SendPing() {
 215   DCHECK(thread_checker_.CalledOnValidThread());
 216   delegate_task_runner_->PostTask(
 217       FROM_HERE,
 218       base::Bind(&NonBlockingPushClient::Core::SendPing, core_.get()));
 219 }
 220
 221 void NonBlockingPushClient::OnNotificationsEnabled() {
 222   DCHECK(thread_checker_.CalledOnValidThread());
 223   FOR_EACH_OBSERVER(PushClientObserver, observers_,
 224                     OnNotificationsEnabled());
 225 }
 226
 227 void NonBlockingPushClient::OnNotificationsDisabled(
 228     NotificationsDisabledReason reason) {
 229   DCHECK(thread_checker_.CalledOnValidThread());
 230   FOR_EACH_OBSERVER(PushClientObserver, observers_,
 231                     OnNotificationsDisabled(reason));
 232 }
 233
 234 void NonBlockingPushClient::OnIncomingNotification(
 235     const Notification& notification) {
 236   DCHECK(thread_checker_.CalledOnValidThread());
 237   FOR_EACH_OBSERVER(PushClientObserver, observers_,
 238                     OnIncomingNotification(notification));
 239 }
 240
 241 void NonBlockingPushClient::OnPingResponse() {
 242   DCHECK(thread_checker_.CalledOnValidThread());
 243   FOR_EACH_OBSERVER(PushClientObserver, observers_, OnPingResponse());
 244 }
 245
 246 }  // namespace notifier