src/device/serial/data_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_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 // Represents a receive that is not yet fulfilled.
  15 class DataReceiver::PendingReceive {
  16  public:
  17   PendingReceive(DataReceiver* receiver,
  18                  const ReceiveDataCallback& callback,
  19                  const ReceiveErrorCallback& error_callback,
  20                  int32_t fatal_error_value);
  21
  22   // Dispatches |data| to |receive_callback_|. Returns whether this
  23   // PendingReceive is finished by this call.
  24   bool DispatchDataFrame(DataReceiver::DataFrame* data);
  25
  26   // Reports |fatal_error_value_| to |receive_error_callback_|.
  27   void DispatchFatalError();
  28
  29   bool buffer_in_use() { return buffer_in_use_; }
  30
  31  private:
  32   class Buffer;
  33
  34   // Invoked when the user is finished with the ReadOnlyBuffer provided to
  35   // |receive_callback_|.
  36   void Done(uint32_t num_bytes);
  37
  38   // The DataReceiver that owns this.
  39   DataReceiver* receiver_;
  40
  41   // The callback to dispatch data.
  42   ReceiveDataCallback receive_callback_;
  43
  44   // The callback to report errors.
  45   ReceiveErrorCallback receive_error_callback_;
  46
  47   // The error value to report when DispatchFatalError() is called.
  48   const int32_t fatal_error_value_;
  49
  50   // True if the user owns a buffer passed to |receive_callback_| as part of
  51   // DispatchDataFrame().
  52   bool buffer_in_use_;
  53 };
  54
  55 // A ReadOnlyBuffer implementation that provides a view of a buffer owned by a
  56 // DataReceiver.
  57 class DataReceiver::PendingReceive::Buffer : public ReadOnlyBuffer {
  58  public:
  59   Buffer(scoped_refptr<DataReceiver> pipe,
  60          PendingReceive* receive,
  61          const char* buffer,
  62          uint32_t buffer_size);
  63   ~Buffer() override;
  64
  65   // ReadOnlyBuffer overrides.
  66   const char* GetData() override;
  67   uint32_t GetSize() override;
  68   void Done(uint32_t bytes_consumed) override;
  69   void DoneWithError(uint32_t bytes_consumed, int32_t error) override;
  70
  71  private:
  72   // The DataReceiver of whose buffer we are providing a view.
  73   scoped_refptr<DataReceiver> receiver_;
  74
  75   // The PendingReceive to which this buffer has been created in response.
  76   PendingReceive* pending_receive_;
  77
  78   const char* buffer_;
  79   uint32_t buffer_size_;
  80 };
  81
  82 // A buffer of data or an error received from the DataSource.
  83 struct DataReceiver::DataFrame {
  84   explicit DataFrame(mojo::Array<uint8_t> data)
  85       : is_error(false),
  86         data(data.Pass()),
  87         offset(0),
  88         error(0),
  89         dispatched(false) {}
  90
  91   explicit DataFrame(int32_t error)
  92       : is_error(true), offset(0), error(error), dispatched(false) {}
  93
  94   // Whether this DataFrame represents an error.
  95   bool is_error;
  96
  97   // The data received from the DataSource.
  98   mojo::Array<uint8_t> data;
  99
 100   // The offset within |data| at which the next read should begin.
 101   uint32_t offset;
 102
 103   // The value of the error that occurred.
 104   const int32_t error;
 105
 106   // Whether the error has been dispatched to the user.
 107   bool dispatched;
 108 };
 109
 110 DataReceiver::DataReceiver(mojo::InterfacePtr<serial::DataSource> source,
 111                            uint32_t buffer_size,
 112                            int32_t fatal_error_value)
 113     : source_(source.Pass()),
 114       fatal_error_value_(fatal_error_value),
 115       shut_down_(false),
 116       weak_factory_(this) {
 117   source_.set_client(this);
 118   source_.set_error_handler(this);
 119   source_->Init(buffer_size);
 120 }
 121
 122 bool DataReceiver::Receive(const ReceiveDataCallback& callback,
 123                            const ReceiveErrorCallback& error_callback) {
 124   DCHECK(!callback.is_null() && !error_callback.is_null());
 125   if (pending_receive_ || shut_down_)
 126     return false;
 127   // When the DataSource encounters an error, it pauses transmission. When the
 128   // user starts a new receive following notification of the error (via
 129   // |error_callback| of the previous Receive call) of the error we can tell the
 130   // DataSource to resume transmission of data.
 131   if (!pending_data_frames_.empty() && pending_data_frames_.front()->is_error &&
 132       pending_data_frames_.front()->dispatched) {
 133     source_->Resume();
 134     pending_data_frames_.pop();
 135   }
 136
 137   pending_receive_.reset(
 138       new PendingReceive(this, callback, error_callback, fatal_error_value_));
 139   ReceiveInternal();
 140   return true;
 141 }
 142
 143 DataReceiver::~DataReceiver() {
 144   ShutDown();
 145 }
 146
 147 void DataReceiver::OnError(int32_t error) {
 148   if (shut_down_)
 149     return;
 150
 151   pending_data_frames_.push(linked_ptr<DataFrame>(new DataFrame(error)));
 152   if (pending_receive_)
 153     ReceiveInternal();
 154 }
 155
 156 void DataReceiver::OnData(mojo::Array<uint8_t> data) {
 157   pending_data_frames_.push(linked_ptr<DataFrame>(new DataFrame(data.Pass())));
 158   if (pending_receive_)
 159     ReceiveInternal();
 160 }
 161
 162 void DataReceiver::OnConnectionError() {
 163   ShutDown();
 164 }
 165
 166 void DataReceiver::Done(uint32_t bytes_consumed) {
 167   if (shut_down_)
 168     return;
 169
 170   DCHECK(pending_receive_);
 171   DataFrame& pending_data = *pending_data_frames_.front();
 172   pending_data.offset += bytes_consumed;
 173   DCHECK_LE(pending_data.offset, pending_data.data.size());
 174   if (pending_data.offset == pending_data.data.size()) {
 175     source_->ReportBytesReceived(
 176         static_cast<uint32_t>(pending_data.data.size()));
 177     pending_data_frames_.pop();
 178   }
 179   pending_receive_.reset();
 180 }
 181
 182 void DataReceiver::ReceiveInternal() {
 183   if (shut_down_)
 184     return;
 185   DCHECK(pending_receive_);
 186   if (pending_receive_->buffer_in_use())
 187     return;
 188
 189   if (!pending_data_frames_.empty() &&
 190       pending_receive_->DispatchDataFrame(pending_data_frames_.front().get())) {
 191     pending_receive_.reset();
 192   }
 193 }
 194
 195 void DataReceiver::ShutDown() {
 196   shut_down_ = true;
 197   if (pending_receive_)
 198     pending_receive_->DispatchFatalError();
 199 }
 200
 201 DataReceiver::PendingReceive::PendingReceive(
 202     DataReceiver* receiver,
 203     const ReceiveDataCallback& callback,
 204     const ReceiveErrorCallback& error_callback,
 205     int32_t fatal_error_value)
 206     : receiver_(receiver),
 207       receive_callback_(callback),
 208       receive_error_callback_(error_callback),
 209       fatal_error_value_(fatal_error_value),
 210       buffer_in_use_(false) {
 211 }
 212
 213 bool DataReceiver::PendingReceive::DispatchDataFrame(
 214     DataReceiver::DataFrame* data) {
 215   DCHECK(!buffer_in_use_);
 216   DCHECK(!data->dispatched);
 217
 218   if (data->is_error) {
 219     data->dispatched = true;
 220     base::MessageLoop::current()->PostTask(
 221         FROM_HERE, base::Bind(receive_error_callback_, data->error));
 222     return true;
 223   }
 224   buffer_in_use_ = true;
 225   base::MessageLoop::current()->PostTask(
 226       FROM_HERE,
 227       base::Bind(
 228           receive_callback_,
 229           base::Passed(scoped_ptr<ReadOnlyBuffer>(new Buffer(
 230               receiver_,
 231               this,
 232               reinterpret_cast<char*>(&data->data[0]) + data->offset,
 233               static_cast<uint32_t>(data->data.size() - data->offset))))));
 234   return false;
 235 }
 236
 237 void DataReceiver::PendingReceive::DispatchFatalError() {
 238   receive_error_callback_.Run(fatal_error_value_);
 239 }
 240
 241 void DataReceiver::PendingReceive::Done(uint32_t bytes_consumed) {
 242   DCHECK(buffer_in_use_);
 243   buffer_in_use_ = false;
 244   receiver_->Done(bytes_consumed);
 245 }
 246
 247 DataReceiver::PendingReceive::Buffer::Buffer(
 248     scoped_refptr<DataReceiver> receiver,
 249     PendingReceive* receive,
 250     const char* buffer,
 251     uint32_t buffer_size)
 252     : receiver_(receiver),
 253       pending_receive_(receive),
 254       buffer_(buffer),
 255       buffer_size_(buffer_size) {
 256 }
 257
 258 DataReceiver::PendingReceive::Buffer::~Buffer() {
 259   if (pending_receive_)
 260     pending_receive_->Done(0);
 261 }
 262
 263 const char* DataReceiver::PendingReceive::Buffer::GetData() {
 264   return buffer_;
 265 }
 266
 267 uint32_t DataReceiver::PendingReceive::Buffer::GetSize() {
 268   return buffer_size_;
 269 }
 270
 271 void DataReceiver::PendingReceive::Buffer::Done(uint32_t bytes_consumed) {
 272   pending_receive_->Done(bytes_consumed);
 273   pending_receive_ = NULL;
 274   receiver_ = NULL;
 275   buffer_ = NULL;
 276   buffer_size_ = 0;
 277 }
 278
 279 void DataReceiver::PendingReceive::Buffer::DoneWithError(
 280     uint32_t bytes_consumed,
 281     int32_t error) {
 282   Done(bytes_consumed);
 283 }
 284
 285 }  // namespace device