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_connection_error_handler(
  56       base::Bind(&DataSender::OnConnectionError, base::Unretained(this)));
  57 }
  58
  59 DataSender::~DataSender() {
  60   ShutDown();
  61 }
  62
  63 bool DataSender::Send(const base::StringPiece& data,
  64                       const DataSentCallback& callback,
  65                       const SendErrorCallback& error_callback) {
  66   DCHECK(!callback.is_null() && !error_callback.is_null());
  67   if (!pending_cancel_.is_null() || shut_down_)
  68     return false;
  69
  70   linked_ptr<PendingSend> pending_send(
  71       new PendingSend(data, callback, error_callback, this));
  72   pending_send->SendData();
  73   sends_awaiting_ack_.push(pending_send);
  74   return true;
  75 }
  76
  77 bool DataSender::Cancel(int32_t error, const CancelCallback& callback) {
  78   DCHECK(!callback.is_null());
  79   if (!pending_cancel_.is_null() || shut_down_)
  80     return false;
  81   if (sends_awaiting_ack_.empty()) {
  82     base::MessageLoop::current()->PostTask(FROM_HERE, callback);
  83     return true;
  84   }
  85
  86   pending_cancel_ = callback;
  87   sink_->Cancel(error);
  88   return true;
  89 }
  90
  91 void DataSender::SendComplete() {
  92   if (shut_down_)
  93     return;
  94
  95   DCHECK(!sends_awaiting_ack_.empty());
  96   sends_awaiting_ack_.pop();
  97   if (sends_awaiting_ack_.empty())
  98     RunCancelCallback();
  99 }
 100
 101 void DataSender::SendFailed(int32_t error) {
 102   if (shut_down_)
 103     return;
 104
 105   DCHECK(!sends_awaiting_ack_.empty());
 106   sends_awaiting_ack_.pop();
 107   if (!sends_awaiting_ack_.empty())
 108     return;
 109   sink_->ClearError();
 110   RunCancelCallback();
 111 }
 112
 113 void DataSender::OnConnectionError() {
 114   ShutDown();
 115 }
 116
 117 void DataSender::RunCancelCallback() {
 118   DCHECK(sends_awaiting_ack_.empty());
 119   if (pending_cancel_.is_null())
 120     return;
 121
 122   base::MessageLoop::current()->PostTask(FROM_HERE,
 123                                          base::Bind(pending_cancel_));
 124   pending_cancel_.Reset();
 125 }
 126
 127 void DataSender::ShutDown() {
 128   shut_down_ = true;
 129   while (!sends_awaiting_ack_.empty()) {
 130     sends_awaiting_ack_.front()->DispatchFatalError();
 131     sends_awaiting_ack_.pop();
 132   }
 133   RunCancelCallback();
 134 }
 135
 136 DataSender::PendingSend::PendingSend(const base::StringPiece& data,
 137                                      const DataSentCallback& callback,
 138                                      const SendErrorCallback& error_callback,
 139                                      DataSender* sender)
 140     : data_(data),
 141       callback_(callback),
 142       error_callback_(error_callback),
 143       sender_(sender) {
 144 }
 145
 146 void DataSender::PendingSend::OnDataSent(uint32_t num_bytes, int32_t error) {
 147   if (error) {
 148     base::MessageLoop::current()->PostTask(
 149         FROM_HERE, base::Bind(error_callback_, num_bytes, error));
 150     sender_->SendFailed(error);
 151   } else {
 152     DCHECK(num_bytes == data_.size());
 153     base::MessageLoop::current()->PostTask(FROM_HERE,
 154                                            base::Bind(callback_, num_bytes));
 155     sender_->SendComplete();
 156   }
 157 }
 158
 159 void DataSender::PendingSend::DispatchFatalError() {
 160   base::MessageLoop::current()->PostTask(
 161       FROM_HERE, base::Bind(error_callback_, 0, sender_->fatal_error_value_));
 162 }
 163
 164 void DataSender::PendingSend::SendData() {
 165   uint32_t num_bytes_to_send = static_cast<uint32_t>(data_.size());
 166   mojo::Array<uint8_t> bytes(num_bytes_to_send);
 167   memcpy(&bytes[0], data_.data(), num_bytes_to_send);
 168   sender_->sink_->OnData(
 169       bytes.Pass(),
 170       base::Bind(&DataSender::PendingSend::OnDataSent, base::Unretained(this)));
 171 }
 172
 173 }  // namespace device