device/serial/data_source_sender.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 "device/serial/data_source_sender.h"
   6
   7 #include <algorithm>
   8 #include <limits>
   9
  10 #include "base/bind.h"
  11 #include "base/message_loop/message_loop.h"
  12
  13 namespace device {
  14
  15 // Represents a send that is not yet fulfilled.
  16 class DataSourceSender::PendingSend {
  17  public:
  18   PendingSend(DataSourceSender* sender, const ReadyCallback& callback);
  19
  20   // Asynchronously fills |data_| with up to |num_bytes| of data. Following
  21   // this, one of Done() and DoneWithError() will be called with the result.
  22   void GetData(uint32_t num_bytes);
  23
  24  private:
  25   class Buffer;
  26   // Reports a successful write of |bytes_written|.
  27   void Done(uint32_t bytes_written);
  28
  29   // Reports a partially successful or unsuccessful write of |bytes_written|
  30   // with an error of |error|.
  31   void DoneWithError(uint32_t bytes_written, int32_t error);
  32
  33   // The DataSourceSender that owns this.
  34   DataSourceSender* sender_;
  35
  36   // The callback to call to get data.
  37   ReadyCallback callback_;
  38
  39   // Whether the buffer specified by GetData() has been passed to |callback_|,
  40   // but has not yet called Done() or DoneWithError().
  41   bool buffer_in_use_;
  42
  43   // The data obtained using |callback_| to be dispatched to the client.
  44   std::vector<char> data_;
  45 };
  46
  47 // A Writable implementation that provides a view of a buffer owned by a
  48 // DataSourceSender.
  49 class DataSourceSender::PendingSend::Buffer : public WritableBuffer {
  50  public:
  51   Buffer(scoped_refptr<DataSourceSender> sender,
  52          PendingSend* send,
  53          char* buffer,
  54          uint32_t buffer_size);
  55   ~Buffer() override;
  56
  57   // WritableBuffer overrides.
  58   char* GetData() override;
  59   uint32_t GetSize() override;
  60   void Done(uint32_t bytes_written) override;
  61   void DoneWithError(uint32_t bytes_written, int32_t error) override;
  62
  63  private:
  64   // The DataSourceSender of whose buffer we are providing a view.
  65   scoped_refptr<DataSourceSender> sender_;
  66
  67   // The PendingSend to which this buffer has been created in response.
  68   PendingSend* pending_send_;
  69
  70   char* buffer_;
  71   uint32_t buffer_size_;
  72 };
  73
  74 DataSourceSender::DataSourceSender(const ReadyCallback& ready_callback,
  75                                    const ErrorCallback& error_callback)
  76     : ready_callback_(ready_callback),
  77       error_callback_(error_callback),
  78       available_buffer_capacity_(0),
  79       paused_(false),
  80       shut_down_(false),
  81       weak_factory_(this) {
  82   DCHECK(!ready_callback.is_null() && !error_callback.is_null());
  83 }
  84
  85 void DataSourceSender::ShutDown() {
  86   shut_down_ = true;
  87   ready_callback_.Reset();
  88   error_callback_.Reset();
  89 }
  90
  91 DataSourceSender::~DataSourceSender() {
  92   DCHECK(shut_down_);
  93 }
  94
  95 void DataSourceSender::Init(uint32_t buffer_size) {
  96   available_buffer_capacity_ = buffer_size;
  97   GetMoreData();
  98 }
  99
 100 void DataSourceSender::Resume() {
 101   if (pending_send_) {
 102     DispatchFatalError();
 103     return;
 104   }
 105
 106   paused_ = false;
 107   GetMoreData();
 108 }
 109
 110 void DataSourceSender::ReportBytesReceived(uint32_t bytes_sent) {
 111   available_buffer_capacity_ += bytes_sent;
 112   if (!pending_send_ && !paused_)
 113     GetMoreData();
 114 }
 115
 116 void DataSourceSender::OnConnectionError() {
 117   DispatchFatalError();
 118 }
 119
 120 void DataSourceSender::GetMoreData() {
 121   if (shut_down_ || paused_ || pending_send_ || !available_buffer_capacity_)
 122     return;
 123
 124   pending_send_.reset(new PendingSend(this, ready_callback_));
 125   pending_send_->GetData(available_buffer_capacity_);
 126 }
 127
 128 void DataSourceSender::Done(const std::vector<char>& data) {
 129   DoneInternal(data);
 130   if (!shut_down_ && available_buffer_capacity_) {
 131     base::MessageLoop::current()->PostTask(
 132         FROM_HERE,
 133         base::Bind(&DataSourceSender::GetMoreData, weak_factory_.GetWeakPtr()));
 134   }
 135 }
 136
 137 void DataSourceSender::DoneWithError(const std::vector<char>& data,
 138                                      int32_t error) {
 139   DoneInternal(data);
 140   if (!shut_down_)
 141     client()->OnError(error);
 142   paused_ = true;
 143   // We don't call GetMoreData here so we don't send any additional data until
 144   // Resume() is called.
 145 }
 146
 147 void DataSourceSender::DoneInternal(const std::vector<char>& data) {
 148   DCHECK(pending_send_);
 149   if (shut_down_)
 150     return;
 151
 152   available_buffer_capacity_ -= static_cast<uint32_t>(data.size());
 153   if (!data.empty()) {
 154     mojo::Array<uint8_t> data_to_send(data.size());
 155     std::copy(data.begin(), data.end(), &data_to_send[0]);
 156     client()->OnData(data_to_send.Pass());
 157   }
 158   pending_send_.reset();
 159 }
 160
 161 void DataSourceSender::DispatchFatalError() {
 162   if (shut_down_)
 163     return;
 164
 165   error_callback_.Run();
 166   ShutDown();
 167 }
 168
 169 DataSourceSender::PendingSend::PendingSend(DataSourceSender* sender,
 170                                            const ReadyCallback& callback)
 171     : sender_(sender), callback_(callback), buffer_in_use_(false) {
 172 }
 173
 174 void DataSourceSender::PendingSend::GetData(uint32_t num_bytes) {
 175   DCHECK(num_bytes);
 176   DCHECK(!buffer_in_use_);
 177   buffer_in_use_ = true;
 178   data_.resize(num_bytes);
 179   callback_.Run(scoped_ptr<WritableBuffer>(
 180       new Buffer(sender_, this, &data_[0], num_bytes)));
 181 }
 182
 183 void DataSourceSender::PendingSend::Done(uint32_t bytes_written) {
 184   DCHECK(buffer_in_use_);
 185   DCHECK_LE(bytes_written, data_.size());
 186   buffer_in_use_ = false;
 187   data_.resize(bytes_written);
 188   sender_->Done(data_);
 189 }
 190
 191 void DataSourceSender::PendingSend::DoneWithError(uint32_t bytes_written,
 192                                                   int32_t error) {
 193   DCHECK(buffer_in_use_);
 194   DCHECK_LE(bytes_written, data_.size());
 195   buffer_in_use_ = false;
 196   data_.resize(bytes_written);
 197   sender_->DoneWithError(data_, error);
 198 }
 199
 200 DataSourceSender::PendingSend::Buffer::Buffer(
 201     scoped_refptr<DataSourceSender> sender,
 202     PendingSend* send,
 203     char* buffer,
 204     uint32_t buffer_size)
 205     : sender_(sender),
 206       pending_send_(send),
 207       buffer_(buffer),
 208       buffer_size_(buffer_size) {
 209 }
 210
 211 DataSourceSender::PendingSend::Buffer::~Buffer() {
 212   if (pending_send_)
 213     pending_send_->Done(0);
 214 }
 215
 216 char* DataSourceSender::PendingSend::Buffer::GetData() {
 217   return buffer_;
 218 }
 219
 220 uint32_t DataSourceSender::PendingSend::Buffer::GetSize() {
 221   return buffer_size_;
 222 }
 223
 224 void DataSourceSender::PendingSend::Buffer::Done(uint32_t bytes_written) {
 225   DCHECK(sender_.get());
 226   PendingSend* send = pending_send_;
 227   pending_send_ = nullptr;
 228   send->Done(bytes_written);
 229   sender_ = nullptr;
 230 }
 231
 232 void DataSourceSender::PendingSend::Buffer::DoneWithError(
 233     uint32_t bytes_written,
 234     int32_t error) {
 235   DCHECK(sender_.get());
 236   PendingSend* send = pending_send_;
 237   pending_send_ = nullptr;
 238   send->DoneWithError(bytes_written, error);
 239   sender_ = nullptr;
 240 }
 241
 242 }  // namespace device