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_connection_error_handler(
  83       base::Bind(&DataSinkReceiver::OnConnectionError, base::Unretained(this)));
  84 }
  85
  86 void DataSinkReceiver::ShutDown() {
  87   shut_down_ = true;
  88 }
  89
  90 DataSinkReceiver::~DataSinkReceiver() {
  91 }
  92
  93 void DataSinkReceiver::Cancel(int32_t error) {
  94   // If we have sent a ReportBytesSentAndError but have not received the
  95   // response, that ReportBytesSentAndError message will appear to the
  96   // DataSinkClient to be caused by this Cancel message. In that case, we ignore
  97   // the cancel.
  98   if (current_error_)
  99     return;
 100
 101   // If there is a buffer is in use, mark the buffer as cancelled and notify the
 102   // client by calling |cancel_callback_|. The sink implementation may or may
 103   // not take the cancellation into account when deciding what error (if any) to
 104   // return. If the sink returns an error, we ignore the cancellation error.
 105   // Otherwise, if the sink does not report an error, we override that with the
 106   // cancellation error. Once a cancellation has been received, the next report
 107   // sent to the client will always contain an error; the error returned by the
 108   // sink or the cancellation error if the sink does not return an error.
 109   if (buffer_in_use_) {
 110     buffer_in_use_->Cancel(error);
 111     if (!cancel_callback_.is_null())
 112       cancel_callback_.Run(error);
 113     return;
 114   }
 115   ReportError(0, error);
 116 }
 117
 118 void DataSinkReceiver::OnData(
 119     mojo::Array<uint8_t> data,
 120     const mojo::Callback<void(uint32_t, int32_t)>& callback) {
 121   if (current_error_) {
 122     callback.Run(0, current_error_);
 123     return;
 124   }
 125   pending_data_buffers_.push(
 126       linked_ptr<DataFrame>(new DataFrame(data.Pass(), callback)));
 127   if (!buffer_in_use_)
 128     RunReadyCallback();
 129 }
 130
 131 void DataSinkReceiver::OnConnectionError() {
 132   DispatchFatalError();
 133 }
 134
 135 void DataSinkReceiver::RunReadyCallback() {
 136   DCHECK(!shut_down_ && !current_error_);
 137   // If data arrives while a call to RunReadyCallback() is posted, we can be
 138   // called with buffer_in_use_ already set.
 139   if (buffer_in_use_)
 140     return;
 141   buffer_in_use_ =
 142       new Buffer(this,
 143                  pending_data_buffers_.front()->GetData(),
 144                  pending_data_buffers_.front()->GetRemainingBytes());
 145   ready_callback_.Run(scoped_ptr<ReadOnlyBuffer>(buffer_in_use_));
 146 }
 147
 148 void DataSinkReceiver::Done(uint32_t bytes_read) {
 149   if (!DoneInternal(bytes_read))
 150     return;
 151   pending_data_buffers_.front()->OnDataConsumed(bytes_read);
 152   if (pending_data_buffers_.front()->GetRemainingBytes() == 0)
 153     pending_data_buffers_.pop();
 154   if (!pending_data_buffers_.empty()) {
 155     base::MessageLoop::current()->PostTask(
 156         FROM_HERE,
 157         base::Bind(&DataSinkReceiver::RunReadyCallback,
 158                    weak_factory_.GetWeakPtr()));
 159   }
 160 }
 161
 162 void DataSinkReceiver::DoneWithError(uint32_t bytes_read, int32_t error) {
 163   if (!DoneInternal(bytes_read))
 164     return;
 165   ReportError(bytes_read, error);
 166 }
 167
 168 bool DataSinkReceiver::DoneInternal(uint32_t bytes_read) {
 169   if (shut_down_)
 170     return false;
 171
 172   DCHECK(buffer_in_use_);
 173   buffer_in_use_ = NULL;
 174   return true;
 175 }
 176
 177 void DataSinkReceiver::ReportError(uint32_t bytes_read, int32_t error) {
 178   // When we encounter an error, we must discard the data from any send buffers
 179   // transmitted by the DataSink client before it receives this error.
 180   DCHECK(error);
 181   current_error_ = error;
 182   while (!pending_data_buffers_.empty()) {
 183     pending_data_buffers_.front()->ReportError(bytes_read, error);
 184     pending_data_buffers_.pop();
 185     bytes_read = 0;
 186   }
 187 }
 188
 189 void DataSinkReceiver::ClearError() {
 190   current_error_ = 0;
 191 }
 192
 193 void DataSinkReceiver::DispatchFatalError() {
 194   if (shut_down_)
 195     return;
 196
 197   ShutDown();
 198   if (!error_callback_.is_null())
 199     error_callback_.Run();
 200 }
 201
 202 DataSinkReceiver::Buffer::Buffer(scoped_refptr<DataSinkReceiver> receiver,
 203                                  const char* buffer,
 204                                  uint32_t buffer_size)
 205     : receiver_(receiver),
 206       buffer_(buffer),
 207       buffer_size_(buffer_size),
 208       cancelled_(false),
 209       cancellation_error_(0) {
 210 }
 211
 212 DataSinkReceiver::Buffer::~Buffer() {
 213   if (!receiver_.get())
 214     return;
 215   if (cancelled_)
 216     receiver_->DoneWithError(0, cancellation_error_);
 217   else
 218     receiver_->Done(0);
 219 }
 220
 221 void DataSinkReceiver::Buffer::Cancel(int32_t error) {
 222   cancelled_ = true;
 223   cancellation_error_ = error;
 224 }
 225
 226 const char* DataSinkReceiver::Buffer::GetData() {
 227   return buffer_;
 228 }
 229
 230 uint32_t DataSinkReceiver::Buffer::GetSize() {
 231   return buffer_size_;
 232 }
 233
 234 void DataSinkReceiver::Buffer::Done(uint32_t bytes_read) {
 235   scoped_refptr<DataSinkReceiver> receiver = receiver_;
 236   receiver_ = nullptr;
 237   if (cancelled_)
 238     receiver->DoneWithError(bytes_read, cancellation_error_);
 239   else
 240     receiver->Done(bytes_read);
 241   buffer_ = NULL;
 242   buffer_size_ = 0;
 243 }
 244
 245 void DataSinkReceiver::Buffer::DoneWithError(uint32_t bytes_read,
 246                                              int32_t error) {
 247   scoped_refptr<DataSinkReceiver> receiver = receiver_;
 248   receiver_ = nullptr;
 249   receiver->DoneWithError(bytes_read, error);
 250   buffer_ = NULL;
 251   buffer_size_ = 0;
 252 }
 253
 254 DataSinkReceiver::DataFrame::DataFrame(
 255     mojo::Array<uint8_t> data,
 256     const mojo::Callback<void(uint32_t, int32_t)>& callback)
 257     : data_(data.Pass()), offset_(0), callback_(callback) {
 258   DCHECK_LT(0u, data_.size());
 259 }
 260
 261 // Returns the number of uncomsumed bytes remaining of this data frame.
 262 uint32_t DataSinkReceiver::DataFrame::GetRemainingBytes() {
 263   return static_cast<uint32_t>(data_.size() - offset_);
 264 }
 265
 266 // Returns a pointer to the remaining data to be consumed.
 267 const char* DataSinkReceiver::DataFrame::GetData() {
 268   DCHECK_LT(offset_, data_.size());
 269   return reinterpret_cast<const char*>(&data_[0]) + offset_;
 270 }
 271
 272 void DataSinkReceiver::DataFrame::OnDataConsumed(uint32_t bytes_read) {
 273   offset_ += bytes_read;
 274   DCHECK_LE(offset_, data_.size());
 275   if (offset_ == data_.size())
 276     callback_.Run(offset_, 0);
 277 }
 278 void DataSinkReceiver::DataFrame::ReportError(uint32_t bytes_read,
 279                                               int32_t error) {
 280   offset_ += bytes_read;
 281   DCHECK_LE(offset_, data_.size());
 282   callback_.Run(offset_, error);
 283 }
 284
 285 }  // namespace device