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(
  75     mojo::InterfaceRequest<serial::DataSource> source,
  76     mojo::InterfacePtr<serial::DataSourceClient> client,
  77     const ReadyCallback& ready_callback,
  78     const ErrorCallback& error_callback)
  79     : binding_(this, source.Pass()),
  80       client_(client.Pass()),
  81       ready_callback_(ready_callback),
  82       error_callback_(error_callback),
  83       available_buffer_capacity_(0),
  84       paused_(false),
  85       shut_down_(false),
  86       weak_factory_(this) {
  87   DCHECK(!ready_callback.is_null() && !error_callback.is_null());
  88   binding_.set_error_handler(this);
  89   client_.set_error_handler(this);
  90 }
  91
  92 void DataSourceSender::ShutDown() {
  93   shut_down_ = true;
  94   ready_callback_.Reset();
  95   error_callback_.Reset();
  96 }
  97
  98 DataSourceSender::~DataSourceSender() {
  99   DCHECK(shut_down_);
 100 }
 101
 102 void DataSourceSender::Init(uint32_t buffer_size) {
 103   available_buffer_capacity_ = buffer_size;
 104   GetMoreData();
 105 }
 106
 107 void DataSourceSender::Resume() {
 108   if (pending_send_) {
 109     DispatchFatalError();
 110     return;
 111   }
 112
 113   paused_ = false;
 114   GetMoreData();
 115 }
 116
 117 void DataSourceSender::ReportBytesReceived(uint32_t bytes_sent) {
 118   available_buffer_capacity_ += bytes_sent;
 119   if (!pending_send_ && !paused_)
 120     GetMoreData();
 121 }
 122
 123 void DataSourceSender::OnConnectionError() {
 124   DispatchFatalError();
 125 }
 126
 127 void DataSourceSender::GetMoreData() {
 128   if (shut_down_ || paused_ || pending_send_ || !available_buffer_capacity_)
 129     return;
 130
 131   pending_send_.reset(new PendingSend(this, ready_callback_));
 132   pending_send_->GetData(available_buffer_capacity_);
 133 }
 134
 135 void DataSourceSender::Done(const std::vector<char>& data) {
 136   DoneInternal(data);
 137   if (!shut_down_ && available_buffer_capacity_) {
 138     base::MessageLoop::current()->PostTask(
 139         FROM_HERE,
 140         base::Bind(&DataSourceSender::GetMoreData, weak_factory_.GetWeakPtr()));
 141   }
 142 }
 143
 144 void DataSourceSender::DoneWithError(const std::vector<char>& data,
 145                                      int32_t error) {
 146   DoneInternal(data);
 147   if (!shut_down_)
 148     client_->OnError(error);
 149   paused_ = true;
 150   // We don't call GetMoreData here so we don't send any additional data until
 151   // Resume() is called.
 152 }
 153
 154 void DataSourceSender::DoneInternal(const std::vector<char>& data) {
 155   DCHECK(pending_send_);
 156   if (shut_down_)
 157     return;
 158
 159   available_buffer_capacity_ -= static_cast<uint32_t>(data.size());
 160   if (!data.empty()) {
 161     mojo::Array<uint8_t> data_to_send(data.size());
 162     std::copy(data.begin(), data.end(), &data_to_send[0]);
 163     client_->OnData(data_to_send.Pass());
 164   }
 165   pending_send_.reset();
 166 }
 167
 168 void DataSourceSender::DispatchFatalError() {
 169   if (shut_down_)
 170     return;
 171
 172   error_callback_.Run();
 173   ShutDown();
 174 }
 175
 176 DataSourceSender::PendingSend::PendingSend(DataSourceSender* sender,
 177                                            const ReadyCallback& callback)
 178     : sender_(sender), callback_(callback), buffer_in_use_(false) {
 179 }
 180
 181 void DataSourceSender::PendingSend::GetData(uint32_t num_bytes) {
 182   DCHECK(num_bytes);
 183   DCHECK(!buffer_in_use_);
 184   buffer_in_use_ = true;
 185   data_.resize(num_bytes);
 186   callback_.Run(scoped_ptr<WritableBuffer>(
 187       new Buffer(sender_, this, &data_[0], num_bytes)));
 188 }
 189
 190 void DataSourceSender::PendingSend::Done(uint32_t bytes_written) {
 191   DCHECK(buffer_in_use_);
 192   DCHECK_LE(bytes_written, data_.size());
 193   buffer_in_use_ = false;
 194   data_.resize(bytes_written);
 195   sender_->Done(data_);
 196 }
 197
 198 void DataSourceSender::PendingSend::DoneWithError(uint32_t bytes_written,
 199                                                   int32_t error) {
 200   DCHECK(buffer_in_use_);
 201   DCHECK_LE(bytes_written, data_.size());
 202   buffer_in_use_ = false;
 203   data_.resize(bytes_written);
 204   sender_->DoneWithError(data_, error);
 205 }
 206
 207 DataSourceSender::PendingSend::Buffer::Buffer(
 208     scoped_refptr<DataSourceSender> sender,
 209     PendingSend* send,
 210     char* buffer,
 211     uint32_t buffer_size)
 212     : sender_(sender),
 213       pending_send_(send),
 214       buffer_(buffer),
 215       buffer_size_(buffer_size) {
 216 }
 217
 218 DataSourceSender::PendingSend::Buffer::~Buffer() {
 219   if (pending_send_)
 220     pending_send_->Done(0);
 221 }
 222
 223 char* DataSourceSender::PendingSend::Buffer::GetData() {
 224   return buffer_;
 225 }
 226
 227 uint32_t DataSourceSender::PendingSend::Buffer::GetSize() {
 228   return buffer_size_;
 229 }
 230
 231 void DataSourceSender::PendingSend::Buffer::Done(uint32_t bytes_written) {
 232   DCHECK(sender_.get());
 233   PendingSend* send = pending_send_;
 234   pending_send_ = nullptr;
 235   send->Done(bytes_written);
 236   sender_ = nullptr;
 237 }
 238
 239 void DataSourceSender::PendingSend::Buffer::DoneWithError(
 240     uint32_t bytes_written,
 241     int32_t error) {
 242   DCHECK(sender_.get());
 243   PendingSend* send = pending_send_;
 244   pending_send_ = nullptr;
 245   send->DoneWithError(bytes_written, error);
 246   sender_ = nullptr;
 247 }
 248
 249 }  // namespace device