device/serial/data_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_sender.h"
   6
   7 #include <algorithm>
   8
   9 #include "base/bind.h"
  10 #include "base/message_loop/message_loop.h"
  11
  12 namespace device {
  13
  14 // Represents a send that is not yet fulfilled.
  15 class DataSender::PendingSend {
  16  public:
  17   PendingSend(const base::StringPiece& data,
  18               const DataSentCallback& callback,
  19               const SendErrorCallback& error_callback,
  20               DataSender* sender);
  21
  22   // Reports |fatal_error_value_| to |receive_error_callback_|.
  23   void DispatchFatalError();
  24
  25   // Attempts to send any data not yet sent to |sink|.
  26   void SendData();
  27
  28  private:
  29   // Invoked to report that |num_bytes| of data have been sent and then an
  30   // error, |error| was encountered. Subtracts the number of bytes that were
  31   // part of this send from |num_bytes|. If this send was not completed before
  32   // the error, this calls |error_callback_| to report the error. Otherwise,
  33   // this calls |callback_|. Returns the number of bytes sent but not acked.
  34   void OnDataSent(uint32_t num_bytes, int32_t error);
  35
  36   // The data to send.
  37   const base::StringPiece data_;
  38
  39   // The callback to report success.
  40   const DataSentCallback callback_;
  41
  42   // The callback to report errors.
  43   const SendErrorCallback error_callback_;
  44
  45   // The DataSender that owns this PendingSend.
  46   DataSender* sender_;
  47 };
  48
  49 DataSender::DataSender(mojo::InterfacePtr<serial::DataSink> sink,
  50                        uint32_t buffer_size,
  51                        int32_t fatal_error_value)
  52     : sink_(sink.Pass()),
  53       fatal_error_value_(fatal_error_value),
  54       shut_down_(false) {
  55   sink_.set_error_handler(this);
  56 }
  57
  58 DataSender::~DataSender() {
  59   ShutDown();
  60 }
  61
  62 bool DataSender::Send(const base::StringPiece& data,
  63                       const DataSentCallback& callback,
  64                       const SendErrorCallback& error_callback) {
  65   DCHECK(!callback.is_null() && !error_callback.is_null());
  66   if (!pending_cancel_.is_null() || shut_down_)
  67     return false;
  68
  69   linked_ptr<PendingSend> pending_send(
  70       new PendingSend(data, callback, error_callback, this));
  71   pending_send->SendData();
  72   sends_awaiting_ack_.push(pending_send);
  73   return true;
  74 }
  75
  76 bool DataSender::Cancel(int32_t error, const CancelCallback& callback) {
  77   DCHECK(!callback.is_null());
  78   if (!pending_cancel_.is_null() || shut_down_)
  79     return false;
  80   if (sends_awaiting_ack_.empty()) {
  81     base::MessageLoop::current()->PostTask(FROM_HERE, callback);
  82     return true;
  83   }
  84
  85   pending_cancel_ = callback;
  86   sink_->Cancel(error);
  87   return true;
  88 }
  89
  90 void DataSender::SendComplete() {
  91   if (shut_down_)
  92     return;
  93
  94   DCHECK(!sends_awaiting_ack_.empty());
  95   sends_awaiting_ack_.pop();
  96   if (sends_awaiting_ack_.empty())
  97     RunCancelCallback();
  98 }
  99
 100 void DataSender::SendFailed(int32_t error) {
 101   if (shut_down_)
 102     return;
 103
 104   DCHECK(!sends_awaiting_ack_.empty());
 105   sends_awaiting_ack_.pop();
 106   if (!sends_awaiting_ack_.empty())
 107     return;
 108   sink_->ClearError();
 109   RunCancelCallback();
 110 }
 111
 112 void DataSender::OnConnectionError() {
 113   ShutDown();
 114 }
 115
 116 void DataSender::RunCancelCallback() {
 117   DCHECK(sends_awaiting_ack_.empty());
 118   if (pending_cancel_.is_null())
 119     return;
 120
 121   base::MessageLoop::current()->PostTask(FROM_HERE,
 122                                          base::Bind(pending_cancel_));
 123   pending_cancel_.Reset();
 124 }
 125
 126 void DataSender::ShutDown() {
 127   shut_down_ = true;
 128   while (!sends_awaiting_ack_.empty()) {
 129     sends_awaiting_ack_.front()->DispatchFatalError();
 130     sends_awaiting_ack_.pop();
 131   }
 132   RunCancelCallback();
 133 }
 134
 135 DataSender::PendingSend::PendingSend(const base::StringPiece& data,
 136                                      const DataSentCallback& callback,
 137                                      const SendErrorCallback& error_callback,
 138                                      DataSender* sender)
 139     : data_(data),
 140       callback_(callback),
 141       error_callback_(error_callback),
 142       sender_(sender) {
 143 }
 144
 145 void DataSender::PendingSend::OnDataSent(uint32_t num_bytes, int32_t error) {
 146   if (error) {
 147     base::MessageLoop::current()->PostTask(
 148         FROM_HERE, base::Bind(error_callback_, num_bytes, error));
 149     sender_->SendFailed(error);
 150   } else {
 151     DCHECK(num_bytes == data_.size());
 152     base::MessageLoop::current()->PostTask(FROM_HERE,
 153                                            base::Bind(callback_, num_bytes));
 154     sender_->SendComplete();
 155   }
 156 }
 157
 158 void DataSender::PendingSend::DispatchFatalError() {
 159   base::MessageLoop::current()->PostTask(
 160       FROM_HERE, base::Bind(error_callback_, 0, sender_->fatal_error_value_));
 161 }
 162
 163 void DataSender::PendingSend::SendData() {
 164   uint32_t num_bytes_to_send = static_cast<uint32_t>(data_.size());
 165   mojo::Array<uint8_t> bytes(num_bytes_to_send);
 166   memcpy(&bytes[0], data_.data(), num_bytes_to_send);
 167   sender_->sink_->OnData(
 168       bytes.Pass(),
 169       base::Bind(&DataSender::PendingSend::OnDataSent, base::Unretained(this)));
 170 }
 171
 172 }  // namespace device