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.
5 #include "device/serial/data_source_sender.h"
10 #include "base/bind.h"
11 #include "base/message_loop/message_loop.h"
15 // Represents a send that is not yet fulfilled.
16 class DataSourceSender::PendingSend {
18 PendingSend(DataSourceSender* sender, const ReadyCallback& callback);
20 // Asynchronously fills |data_| with up to |num_bytes| of data. Following
21 // this, one of Done() and DoneWithError() will be called with the result.
22 void GetData(uint32_t num_bytes);
26 // Reports a successful write of |bytes_written|.
27 void Done(uint32_t bytes_written);
29 // Reports a partially successful or unsuccessful write of |bytes_written|
30 // with an error of |error|.
31 void DoneWithError(uint32_t bytes_written, int32_t error);
33 // The DataSourceSender that owns this.
34 DataSourceSender* sender_;
36 // The callback to call to get data.
37 ReadyCallback callback_;
39 // Whether the buffer specified by GetData() has been passed to |callback_|,
40 // but has not yet called Done() or DoneWithError().
43 // The data obtained using |callback_| to be dispatched to the client.
44 std::vector<char> data_;
47 // A Writable implementation that provides a view of a buffer owned by a
49 class DataSourceSender::PendingSend::Buffer : public WritableBuffer {
51 Buffer(scoped_refptr<DataSourceSender> sender,
54 uint32_t buffer_size);
57 // WritableBuffer overrides.
58 char* GetData() override;
59 uint32_t GetSize() override;
60 void Done(uint32_t bytes_written) override;
61 void DoneWithError(uint32_t bytes_written, int32_t error) override;
64 // The DataSourceSender of whose buffer we are providing a view.
65 scoped_refptr<DataSourceSender> sender_;
67 // The PendingSend to which this buffer has been created in response.
68 PendingSend* pending_send_;
71 uint32_t buffer_size_;
74 DataSourceSender::DataSourceSender(const ReadyCallback& ready_callback,
75 const ErrorCallback& error_callback)
76 : ready_callback_(ready_callback),
77 error_callback_(error_callback),
78 available_buffer_capacity_(0),
82 DCHECK(!ready_callback.is_null() && !error_callback.is_null());
85 void DataSourceSender::ShutDown() {
87 ready_callback_.Reset();
88 error_callback_.Reset();
91 DataSourceSender::~DataSourceSender() {
95 void DataSourceSender::Init(uint32_t buffer_size) {
96 available_buffer_capacity_ = buffer_size;
100 void DataSourceSender::Resume() {
102 DispatchFatalError();
110 void DataSourceSender::ReportBytesReceived(uint32_t bytes_sent) {
111 available_buffer_capacity_ += bytes_sent;
112 if (!pending_send_ && !paused_)
116 void DataSourceSender::OnConnectionError() {
117 DispatchFatalError();
120 void DataSourceSender::GetMoreData() {
121 if (shut_down_ || paused_ || pending_send_ || !available_buffer_capacity_)
124 pending_send_.reset(new PendingSend(this, ready_callback_));
125 pending_send_->GetData(available_buffer_capacity_);
128 void DataSourceSender::Done(const std::vector<char>& data) {
130 if (!shut_down_ && available_buffer_capacity_) {
131 base::MessageLoop::current()->PostTask(
133 base::Bind(&DataSourceSender::GetMoreData, weak_factory_.GetWeakPtr()));
137 void DataSourceSender::DoneWithError(const std::vector<char>& data,
141 client()->OnError(error);
143 // We don't call GetMoreData here so we don't send any additional data until
144 // Resume() is called.
147 void DataSourceSender::DoneInternal(const std::vector<char>& data) {
148 DCHECK(pending_send_);
152 available_buffer_capacity_ -= static_cast<uint32_t>(data.size());
154 mojo::Array<uint8_t> data_to_send(data.size());
155 std::copy(data.begin(), data.end(), &data_to_send[0]);
156 client()->OnData(data_to_send.Pass());
158 pending_send_.reset();
161 void DataSourceSender::DispatchFatalError() {
165 error_callback_.Run();
169 DataSourceSender::PendingSend::PendingSend(DataSourceSender* sender,
170 const ReadyCallback& callback)
171 : sender_(sender), callback_(callback), buffer_in_use_(false) {
174 void DataSourceSender::PendingSend::GetData(uint32_t num_bytes) {
176 DCHECK(!buffer_in_use_);
177 buffer_in_use_ = true;
178 data_.resize(num_bytes);
179 callback_.Run(scoped_ptr<WritableBuffer>(
180 new Buffer(sender_, this, &data_[0], num_bytes)));
183 void DataSourceSender::PendingSend::Done(uint32_t bytes_written) {
184 DCHECK(buffer_in_use_);
185 DCHECK_LE(bytes_written, data_.size());
186 buffer_in_use_ = false;
187 data_.resize(bytes_written);
188 sender_->Done(data_);
191 void DataSourceSender::PendingSend::DoneWithError(uint32_t bytes_written,
193 DCHECK(buffer_in_use_);
194 DCHECK_LE(bytes_written, data_.size());
195 buffer_in_use_ = false;
196 data_.resize(bytes_written);
197 sender_->DoneWithError(data_, error);
200 DataSourceSender::PendingSend::Buffer::Buffer(
201 scoped_refptr<DataSourceSender> sender,
204 uint32_t buffer_size)
208 buffer_size_(buffer_size) {
211 DataSourceSender::PendingSend::Buffer::~Buffer() {
213 pending_send_->Done(0);
216 char* DataSourceSender::PendingSend::Buffer::GetData() {
220 uint32_t DataSourceSender::PendingSend::Buffer::GetSize() {
224 void DataSourceSender::PendingSend::Buffer::Done(uint32_t bytes_written) {
225 DCHECK(sender_.get());
226 PendingSend* send = pending_send_;
227 pending_send_ = nullptr;
228 send->Done(bytes_written);
232 void DataSourceSender::PendingSend::Buffer::DoneWithError(
233 uint32_t bytes_written,
235 DCHECK(sender_.get());
236 PendingSend* send = pending_send_;
237 pending_send_ = nullptr;
238 send->DoneWithError(bytes_written, error);
242 } // namespace device