device/serial/data_sink_receiver.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_sink_receiver.h"
   6
   7 #include <limits>
   8
   9 #include "base/bind.h"
  10 #include "base/message_loop/message_loop.h"
  11
  12 namespace device {
  13
  14 // A ReadOnlyBuffer implementation that provides a view of a buffer owned by a
  15 // DataSinkReceiver.
  16 class DataSinkReceiver::Buffer : public ReadOnlyBuffer {
  17  public:
  18   Buffer(scoped_refptr<DataSinkReceiver> receiver,
  19          const char* buffer,
  20          uint32_t buffer_size);
  21   ~Buffer() override;
  22
  23   void Cancel(int32_t error);
  24
  25   // ReadOnlyBuffer overrides.
  26   const char* GetData() override;
  27   uint32_t GetSize() override;
  28   void Done(uint32_t bytes_read) override;
  29   void DoneWithError(uint32_t bytes_read, int32_t error) override;
  30
  31  private:
  32   // The DataSinkReceiver of whose buffer we are providing a view.
  33   scoped_refptr<DataSinkReceiver> receiver_;
  34
  35   const char* buffer_;
  36   uint32_t buffer_size_;
  37
  38   // Whether this receive has been cancelled.
  39   bool cancelled_;
  40
  41   // If |cancelled_|, contains the cancellation error to report.
  42   int32_t cancellation_error_;
  43 };
  44
  45 // A frame of data received from the client.
  46 class DataSinkReceiver::DataFrame {
  47  public:
  48   explicit DataFrame(mojo::Array<uint8_t> data,
  49                      const mojo::Callback<void(uint32_t, int32_t)>& callback);
  50
  51   // Returns the number of unconsumed bytes remaining of this data frame.
  52   uint32_t GetRemainingBytes();
  53
  54   // Returns a pointer to the remaining data to be consumed.
  55   const char* GetData();
  56
  57   // Reports that |bytes_read| bytes have been consumed.
  58   void OnDataConsumed(uint32_t bytes_read);
  59
  60   // Reports that an error occurred.
  61   void ReportError(uint32_t bytes_read, int32_t error);
  62
  63  private:
  64   mojo::Array<uint8_t> data_;
  65   uint32_t offset_;
  66   const mojo::Callback<void(uint32_t, int32_t)> callback_;
  67 };
  68
  69 DataSinkReceiver::DataSinkReceiver(
  70     mojo::InterfaceRequest<serial::DataSink> request,
  71     const ReadyCallback& ready_callback,
  72     const CancelCallback& cancel_callback,
  73     const ErrorCallback& error_callback)
  74     : binding_(this, request.Pass()),
  75       ready_callback_(ready_callback),
  76       cancel_callback_(cancel_callback),
  77       error_callback_(error_callback),
  78       current_error_(0),
  79       buffer_in_use_(NULL),
  80       shut_down_(false),
  81       weak_factory_(this) {
  82   binding_.set_error_handler(this);
  83 }
  84
  85 void DataSinkReceiver::ShutDown() {
  86   shut_down_ = true;
  87 }
  88
  89 DataSinkReceiver::~DataSinkReceiver() {
  90 }
  91
  92 void DataSinkReceiver::Cancel(int32_t error) {
  93   // If we have sent a ReportBytesSentAndError but have not received the
  94   // response, that ReportBytesSentAndError message will appear to the
  95   // DataSinkClient to be caused by this Cancel message. In that case, we ignore
  96   // the cancel.
  97   if (current_error_)
  98     return;
  99
 100   // If there is a buffer is in use, mark the buffer as cancelled and notify the
 101   // client by calling |cancel_callback_|. The sink implementation may or may
 102   // not take the cancellation into account when deciding what error (if any) to
 103   // return. If the sink returns an error, we ignore the cancellation error.
 104   // Otherwise, if the sink does not report an error, we override that with the
 105   // cancellation error. Once a cancellation has been received, the next report
 106   // sent to the client will always contain an error; the error returned by the
 107   // sink or the cancellation error if the sink does not return an error.
 108   if (buffer_in_use_) {
 109     buffer_in_use_->Cancel(error);
 110     if (!cancel_callback_.is_null())
 111       cancel_callback_.Run(error);
 112     return;
 113   }
 114   ReportError(0, error);
 115 }
 116
 117 void DataSinkReceiver::OnData(
 118     mojo::Array<uint8_t> data,
 119     const mojo::Callback<void(uint32_t, int32_t)>& callback) {
 120   if (current_error_) {
 121     callback.Run(0, current_error_);
 122     return;
 123   }
 124   pending_data_buffers_.push(
 125       linked_ptr<DataFrame>(new DataFrame(data.Pass(), callback)));
 126   if (!buffer_in_use_)
 127     RunReadyCallback();
 128 }
 129
 130 void DataSinkReceiver::OnConnectionError() {
 131   DispatchFatalError();
 132 }
 133
 134 void DataSinkReceiver::RunReadyCallback() {
 135   DCHECK(!shut_down_ && !current_error_);
 136   // If data arrives while a call to RunReadyCallback() is posted, we can be
 137   // called with buffer_in_use_ already set.
 138   if (buffer_in_use_)
 139     return;
 140   buffer_in_use_ =
 141       new Buffer(this,
 142                  pending_data_buffers_.front()->GetData(),
 143                  pending_data_buffers_.front()->GetRemainingBytes());
 144   ready_callback_.Run(scoped_ptr<ReadOnlyBuffer>(buffer_in_use_));
 145 }
 146
 147 void DataSinkReceiver::Done(uint32_t bytes_read) {
 148   if (!DoneInternal(bytes_read))
 149     return;
 150   pending_data_buffers_.front()->OnDataConsumed(bytes_read);
 151   if (pending_data_buffers_.front()->GetRemainingBytes() == 0)
 152     pending_data_buffers_.pop();
 153   if (!pending_data_buffers_.empty()) {
 154     base::MessageLoop::current()->PostTask(
 155         FROM_HERE,
 156         base::Bind(&DataSinkReceiver::RunReadyCallback,
 157                    weak_factory_.GetWeakPtr()));
 158   }
 159 }
 160
 161 void DataSinkReceiver::DoneWithError(uint32_t bytes_read, int32_t error) {
 162   if (!DoneInternal(bytes_read))
 163     return;
 164   ReportError(bytes_read, error);
 165 }
 166
 167 bool DataSinkReceiver::DoneInternal(uint32_t bytes_read) {
 168   if (shut_down_)
 169     return false;
 170
 171   DCHECK(buffer_in_use_);
 172   buffer_in_use_ = NULL;
 173   return true;
 174 }
 175
 176 void DataSinkReceiver::ReportError(uint32_t bytes_read, int32_t error) {
 177   // When we encounter an error, we must discard the data from any send buffers
 178   // transmitted by the DataSink client before it receives this error.
 179   DCHECK(error);
 180   current_error_ = error;
 181   while (!pending_data_buffers_.empty()) {
 182     pending_data_buffers_.front()->ReportError(bytes_read, error);
 183     pending_data_buffers_.pop();
 184     bytes_read = 0;
 185   }
 186 }
 187
 188 void DataSinkReceiver::ClearError() {
 189   current_error_ = 0;
 190 }
 191
 192 void DataSinkReceiver::DispatchFatalError() {
 193   if (shut_down_)
 194     return;
 195
 196   ShutDown();
 197   if (!error_callback_.is_null())
 198     error_callback_.Run();
 199 }
 200
 201 DataSinkReceiver::Buffer::Buffer(scoped_refptr<DataSinkReceiver> receiver,
 202                                  const char* buffer,
 203                                  uint32_t buffer_size)
 204     : receiver_(receiver),
 205       buffer_(buffer),
 206       buffer_size_(buffer_size),
 207       cancelled_(false),
 208       cancellation_error_(0) {
 209 }
 210
 211 DataSinkReceiver::Buffer::~Buffer() {
 212   if (!receiver_.get())
 213     return;
 214   if (cancelled_)
 215     receiver_->DoneWithError(0, cancellation_error_);
 216   else
 217     receiver_->Done(0);
 218 }
 219
 220 void DataSinkReceiver::Buffer::Cancel(int32_t error) {
 221   cancelled_ = true;
 222   cancellation_error_ = error;
 223 }
 224
 225 const char* DataSinkReceiver::Buffer::GetData() {
 226   return buffer_;
 227 }
 228
 229 uint32_t DataSinkReceiver::Buffer::GetSize() {
 230   return buffer_size_;
 231 }
 232
 233 void DataSinkReceiver::Buffer::Done(uint32_t bytes_read) {
 234   scoped_refptr<DataSinkReceiver> receiver = receiver_;
 235   receiver_ = nullptr;
 236   if (cancelled_)
 237     receiver->DoneWithError(bytes_read, cancellation_error_);
 238   else
 239     receiver->Done(bytes_read);
 240   buffer_ = NULL;
 241   buffer_size_ = 0;
 242 }
 243
 244 void DataSinkReceiver::Buffer::DoneWithError(uint32_t bytes_read,
 245                                              int32_t error) {
 246   scoped_refptr<DataSinkReceiver> receiver = receiver_;
 247   receiver_ = nullptr;
 248   receiver->DoneWithError(bytes_read, error);
 249   buffer_ = NULL;
 250   buffer_size_ = 0;
 251 }
 252
 253 DataSinkReceiver::DataFrame::DataFrame(
 254     mojo::Array<uint8_t> data,
 255     const mojo::Callback<void(uint32_t, int32_t)>& callback)
 256     : data_(data.Pass()), offset_(0), callback_(callback) {
 257   DCHECK_LT(0u, data_.size());
 258 }
 259
 260 // Returns the number of uncomsumed bytes remaining of this data frame.
 261 uint32_t DataSinkReceiver::DataFrame::GetRemainingBytes() {
 262   return static_cast<uint32_t>(data_.size() - offset_);
 263 }
 264
 265 // Returns a pointer to the remaining data to be consumed.
 266 const char* DataSinkReceiver::DataFrame::GetData() {
 267   DCHECK_LT(offset_, data_.size());
 268   return reinterpret_cast<const char*>(&data_[0]) + offset_;
 269 }
 270
 271 void DataSinkReceiver::DataFrame::OnDataConsumed(uint32_t bytes_read) {
 272   offset_ += bytes_read;
 273   DCHECK_LE(offset_, data_.size());
 274   if (offset_ == data_.size())
 275     callback_.Run(offset_, 0);
 276 }
 277 void DataSinkReceiver::DataFrame::ReportError(uint32_t bytes_read,
 278                                               int32_t error) {
 279   offset_ += bytes_read;
 280   DCHECK_LE(offset_, data_.size());
 281   callback_.Run(offset_, error);
 282 }
 283
 284 }  // namespace device