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(
 111     mojo::InterfacePtr<serial::DataSource> source,
 112     mojo::InterfaceRequest<serial::DataSourceClient> client,
 113     uint32_t buffer_size,
 114     int32_t fatal_error_value)
 115     : source_(source.Pass()),
 116       client_(this, client.Pass()),
 117       fatal_error_value_(fatal_error_value),
 118       shut_down_(false),
 119       weak_factory_(this) {
 120   source_.set_error_handler(this);
 121   source_->Init(buffer_size);
 122   client_.set_error_handler(this);
 123 }
 124
 125 bool DataReceiver::Receive(const ReceiveDataCallback& callback,
 126                            const ReceiveErrorCallback& error_callback) {
 127   DCHECK(!callback.is_null() && !error_callback.is_null());
 128   if (pending_receive_ || shut_down_)
 129     return false;
 130   // When the DataSource encounters an error, it pauses transmission. When the
 131   // user starts a new receive following notification of the error (via
 132   // |error_callback| of the previous Receive call) of the error we can tell the
 133   // DataSource to resume transmission of data.
 134   if (!pending_data_frames_.empty() && pending_data_frames_.front()->is_error &&
 135       pending_data_frames_.front()->dispatched) {
 136     source_->Resume();
 137     pending_data_frames_.pop();
 138   }
 139
 140   pending_receive_.reset(
 141       new PendingReceive(this, callback, error_callback, fatal_error_value_));
 142   ReceiveInternal();
 143   return true;
 144 }
 145
 146 DataReceiver::~DataReceiver() {
 147   ShutDown();
 148 }
 149
 150 void DataReceiver::OnError(int32_t error) {
 151   if (shut_down_)
 152     return;
 153
 154   pending_data_frames_.push(linked_ptr<DataFrame>(new DataFrame(error)));
 155   if (pending_receive_)
 156     ReceiveInternal();
 157 }
 158
 159 void DataReceiver::OnData(mojo::Array<uint8_t> data) {
 160   pending_data_frames_.push(linked_ptr<DataFrame>(new DataFrame(data.Pass())));
 161   if (pending_receive_)
 162     ReceiveInternal();
 163 }
 164
 165 void DataReceiver::OnConnectionError() {
 166   ShutDown();
 167 }
 168
 169 void DataReceiver::Done(uint32_t bytes_consumed) {
 170   if (shut_down_)
 171     return;
 172
 173   DCHECK(pending_receive_);
 174   DataFrame& pending_data = *pending_data_frames_.front();
 175   pending_data.offset += bytes_consumed;
 176   DCHECK_LE(pending_data.offset, pending_data.data.size());
 177   if (pending_data.offset == pending_data.data.size()) {
 178     source_->ReportBytesReceived(
 179         static_cast<uint32_t>(pending_data.data.size()));
 180     pending_data_frames_.pop();
 181   }
 182   pending_receive_.reset();
 183 }
 184
 185 void DataReceiver::ReceiveInternal() {
 186   if (shut_down_)
 187     return;
 188   DCHECK(pending_receive_);
 189   if (pending_receive_->buffer_in_use())
 190     return;
 191
 192   if (!pending_data_frames_.empty() &&
 193       pending_receive_->DispatchDataFrame(pending_data_frames_.front().get())) {
 194     pending_receive_.reset();
 195   }
 196 }
 197
 198 void DataReceiver::ShutDown() {
 199   shut_down_ = true;
 200   if (pending_receive_)
 201     pending_receive_->DispatchFatalError();
 202 }
 203
 204 DataReceiver::PendingReceive::PendingReceive(
 205     DataReceiver* receiver,
 206     const ReceiveDataCallback& callback,
 207     const ReceiveErrorCallback& error_callback,
 208     int32_t fatal_error_value)
 209     : receiver_(receiver),
 210       receive_callback_(callback),
 211       receive_error_callback_(error_callback),
 212       fatal_error_value_(fatal_error_value),
 213       buffer_in_use_(false) {
 214 }
 215
 216 bool DataReceiver::PendingReceive::DispatchDataFrame(
 217     DataReceiver::DataFrame* data) {
 218   DCHECK(!buffer_in_use_);
 219   DCHECK(!data->dispatched);
 220
 221   if (data->is_error) {
 222     data->dispatched = true;
 223     base::MessageLoop::current()->PostTask(
 224         FROM_HERE, base::Bind(receive_error_callback_, data->error));
 225     return true;
 226   }
 227   buffer_in_use_ = true;
 228   base::MessageLoop::current()->PostTask(
 229       FROM_HERE,
 230       base::Bind(
 231           receive_callback_,
 232           base::Passed(scoped_ptr<ReadOnlyBuffer>(new Buffer(
 233               receiver_,
 234               this,
 235               reinterpret_cast<char*>(&data->data[0]) + data->offset,
 236               static_cast<uint32_t>(data->data.size() - data->offset))))));
 237   return false;
 238 }
 239
 240 void DataReceiver::PendingReceive::DispatchFatalError() {
 241   receive_error_callback_.Run(fatal_error_value_);
 242 }
 243
 244 void DataReceiver::PendingReceive::Done(uint32_t bytes_consumed) {
 245   DCHECK(buffer_in_use_);
 246   buffer_in_use_ = false;
 247   receiver_->Done(bytes_consumed);
 248 }
 249
 250 DataReceiver::PendingReceive::Buffer::Buffer(
 251     scoped_refptr<DataReceiver> receiver,
 252     PendingReceive* receive,
 253     const char* buffer,
 254     uint32_t buffer_size)
 255     : receiver_(receiver),
 256       pending_receive_(receive),
 257       buffer_(buffer),
 258       buffer_size_(buffer_size) {
 259 }
 260
 261 DataReceiver::PendingReceive::Buffer::~Buffer() {
 262   if (pending_receive_)
 263     pending_receive_->Done(0);
 264 }
 265
 266 const char* DataReceiver::PendingReceive::Buffer::GetData() {
 267   return buffer_;
 268 }
 269
 270 uint32_t DataReceiver::PendingReceive::Buffer::GetSize() {
 271   return buffer_size_;
 272 }
 273
 274 void DataReceiver::PendingReceive::Buffer::Done(uint32_t bytes_consumed) {
 275   pending_receive_->Done(bytes_consumed);
 276   pending_receive_ = NULL;
 277   receiver_ = NULL;
 278   buffer_ = NULL;
 279   buffer_size_ = 0;
 280 }
 281
 282 void DataReceiver::PendingReceive::Buffer::DoneWithError(
 283     uint32_t bytes_consumed,
 284     int32_t error) {
 285   Done(bytes_consumed);
 286 }
 287
 288 }  // namespace device