Remove Client= from device/serial/data_stream.mojom.

device/serial/data_sink_receiver.cc

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "device/serial/data_sink_receiver.h"
 
 #include <limits>
 
 #include "base/bind.h"
 #include "base/message_loop/message_loop.h"
@@ skipping 27 matching lines @@
   // Whether this receive has been cancelled.
   bool cancelled_;
 
   // If |cancelled_|, contains the cancellation error to report.
   int32_t cancellation_error_;
 };
 
 // A frame of data received from the client.
 class DataSinkReceiver::DataFrame {
  public:
-  explicit DataFrame(mojo::Array<uint8_t> data);
+  explicit DataFrame(mojo::Array<uint8_t> data,
+                     const mojo::Callback<void(uint32_t, int32_t)>& callback);
 
-  // Returns the number of uncomsumed bytes remaining of this data frame.
+  // Returns the number of unconsumed bytes remaining of this data frame.
   uint32_t GetRemainingBytes();
 
   // Returns a pointer to the remaining data to be consumed.
   const char* GetData();
 
   // Reports that |bytes_read| bytes have been consumed.
   void OnDataConsumed(uint32_t bytes_read);
 
+  // Reports that an error occurred.
+  void ReportError(uint32_t bytes_read, int32_t error);
+
  private:
   mojo::Array<uint8_t> data_;
   uint32_t offset_;
+  const mojo::Callback<void(uint32_t, int32_t)> callback_;
 };
 
-DataSinkReceiver::DataSinkReceiver(const ReadyCallback& ready_callback,
-                                   const CancelCallback& cancel_callback,
-                                   const ErrorCallback& error_callback)
-    : ready_callback_(ready_callback),
+DataSinkReceiver::DataSinkReceiver(
+    mojo::InterfaceRequest<serial::DataSink> request,
+    const ReadyCallback& ready_callback,
+    const CancelCallback& cancel_callback,
+    const ErrorCallback& error_callback)
+    : binding_(this, request.Pass()),
+      ready_callback_(ready_callback),
       cancel_callback_(cancel_callback),
       error_callback_(error_callback),
-      flush_pending_(false),
+      current_error_(0),
       buffer_in_use_(NULL),
-      initialized_(false),
-      available_buffer_capacity_(0),
       shut_down_(false),
       weak_factory_(this) {
+  binding_.set_error_handler(this);
 }
 
 void DataSinkReceiver::ShutDown() {
   shut_down_ = true;
 }
 
 DataSinkReceiver::~DataSinkReceiver() {
 }
 
-void DataSinkReceiver::Init(uint32_t buffer_size) {
-  if (initialized_) {
-    ShutDown();
-    return;
-  }
-  initialized_ = true;
-  available_buffer_capacity_ = buffer_size;
-}
-
 void DataSinkReceiver::Cancel(int32_t error) {
   // If we have sent a ReportBytesSentAndError but have not received the
   // response, that ReportBytesSentAndError message will appear to the
   // DataSinkClient to be caused by this Cancel message. In that case, we ignore
   // the cancel.
-  if (flush_pending_)
+  if (current_error_)
     return;
 
   // If there is a buffer is in use, mark the buffer as cancelled and notify the
   // client by calling |cancel_callback_|. The sink implementation may or may
   // not take the cancellation into account when deciding what error (if any) to
   // return. If the sink returns an error, we ignore the cancellation error.
   // Otherwise, if the sink does not report an error, we override that with the
   // cancellation error. Once a cancellation has been received, the next report
   // sent to the client will always contain an error; the error returned by the
   // sink or the cancellation error if the sink does not return an error.
   if (buffer_in_use_) {
     buffer_in_use_->Cancel(error);
     if (!cancel_callback_.is_null())
       cancel_callback_.Run(error);
     return;
   }
-  ReportBytesSentAndError(0, error);
+  ReportError(0, error);
 }
 
-void DataSinkReceiver::OnData(mojo::Array<uint8_t> data) {
-  if (!initialized_) {
-    ShutDown();
+void DataSinkReceiver::OnData(
+    mojo::Array<uint8_t> data,
+    const mojo::Callback<void(uint32_t, int32_t)>& callback) {
+  if (current_error_) {
+    callback.Run(0, current_error_);
     return;
   }
-  if (data.size() > available_buffer_capacity_) {
-    ShutDown();
-    return;
-  }
-  available_buffer_capacity_ -= static_cast<uint32_t>(data.size());
-  pending_data_buffers_.push(linked_ptr<DataFrame>(new DataFrame(data.Pass())));
-  if (!buffer_in_use_ && !flush_pending_)
+  pending_data_buffers_.push(
+      linked_ptr<DataFrame>(new DataFrame(data.Pass(), callback)));
+  if (!buffer_in_use_)
     RunReadyCallback();
 }
 
 void DataSinkReceiver::OnConnectionError() {
   DispatchFatalError();
 }
 
 void DataSinkReceiver::RunReadyCallback() {
-  DCHECK(!shut_down_ && !flush_pending_);
+  DCHECK(!shut_down_ && !current_error_);
   // If data arrives while a call to RunReadyCallback() is posted, we can be
   // called with buffer_in_use_ already set.
   if (buffer_in_use_)
     return;
   buffer_in_use_ =
       new Buffer(this,
                  pending_data_buffers_.front()->GetData(),
                  pending_data_buffers_.front()->GetRemainingBytes());
   ready_callback_.Run(scoped_ptr<ReadOnlyBuffer>(buffer_in_use_));
 }
 
 void DataSinkReceiver::Done(uint32_t bytes_read) {
   if (!DoneInternal(bytes_read))
     return;
-  client()->ReportBytesSent(bytes_read);
+  pending_data_buffers_.front()->OnDataConsumed(bytes_read);
+  if (pending_data_buffers_.front()->GetRemainingBytes() == 0)
+    pending_data_buffers_.pop();
   if (!pending_data_buffers_.empty()) {
     base::MessageLoop::current()->PostTask(
         FROM_HERE,
         base::Bind(&DataSinkReceiver::RunReadyCallback,
                    weak_factory_.GetWeakPtr()));
   }
 }
 
 void DataSinkReceiver::DoneWithError(uint32_t bytes_read, int32_t error) {
   if (!DoneInternal(bytes_read))
     return;
-  ReportBytesSentAndError(bytes_read, error);
+  ReportError(bytes_read, error);
 }
 
 bool DataSinkReceiver::DoneInternal(uint32_t bytes_read) {
   if (shut_down_)
     return false;
 
   DCHECK(buffer_in_use_);
   buffer_in_use_ = NULL;
-  available_buffer_capacity_ += bytes_read;
-  pending_data_buffers_.front()->OnDataConsumed(bytes_read);
-  if (pending_data_buffers_.front()->GetRemainingBytes() == 0)
-    pending_data_buffers_.pop();
   return true;
 }
 
-void DataSinkReceiver::ReportBytesSentAndError(uint32_t bytes_read,
-                                               int32_t error) {
+void DataSinkReceiver::ReportError(uint32_t bytes_read, int32_t error) {
   // When we encounter an error, we must discard the data from any send buffers
-  // transmitted by the DataSinkClient before it receives this error.
-  flush_pending_ = true;
-  client()->ReportBytesSentAndError(
-      bytes_read,
-      error,
-      base::Bind(&DataSinkReceiver::DoFlush, weak_factory_.GetWeakPtr()));
+  // transmitted by the DataSink client before it receives this error.
+  DCHECK(error);
+  current_error_ = error;
+  while (!pending_data_buffers_.empty()) {
+    pending_data_buffers_.front()->ReportError(bytes_read, error);
+    pending_data_buffers_.pop();
+    bytes_read = 0;
+  }
 }
 
-void DataSinkReceiver::DoFlush() {
-  DCHECK(flush_pending_);
-  flush_pending_ = false;
-  while (!pending_data_buffers_.empty()) {
-    available_buffer_capacity_ +=
-        pending_data_buffers_.front()->GetRemainingBytes();
-    pending_data_buffers_.pop();
-  }
+void DataSinkReceiver::ClearError() {
+  current_error_ = 0;
 }
 
 void DataSinkReceiver::DispatchFatalError() {
   if (shut_down_)
     return;
 
   ShutDown();
   if (!error_callback_.is_null())
     error_callback_.Run();
 }
@@ skipping 43 matching lines @@
 
 void DataSinkReceiver::Buffer::DoneWithError(uint32_t bytes_read,
                                              int32_t error) {
   scoped_refptr<DataSinkReceiver> receiver = receiver_;
   receiver_ = nullptr;
   receiver->DoneWithError(bytes_read, error);
   buffer_ = NULL;
   buffer_size_ = 0;
 }
 
-DataSinkReceiver::DataFrame::DataFrame(mojo::Array<uint8_t> data)
-    : data_(data.Pass()), offset_(0) {
+DataSinkReceiver::DataFrame::DataFrame(
+    mojo::Array<uint8_t> data,
+    const mojo::Callback<void(uint32_t, int32_t)>& callback)
+    : data_(data.Pass()), offset_(0), callback_(callback) {
   DCHECK_LT(0u, data_.size());
 }
 
 // Returns the number of uncomsumed bytes remaining of this data frame.
 uint32_t DataSinkReceiver::DataFrame::GetRemainingBytes() {
   return static_cast<uint32_t>(data_.size() - offset_);
 }
 
 // Returns a pointer to the remaining data to be consumed.
 const char* DataSinkReceiver::DataFrame::GetData() {
   DCHECK_LT(offset_, data_.size());
   return reinterpret_cast<const char*>(&data_[0]) + offset_;
 }
 
 void DataSinkReceiver::DataFrame::OnDataConsumed(uint32_t bytes_read) {
   offset_ += bytes_read;
   DCHECK_LE(offset_, data_.size());
+  if (offset_ == data_.size())
+    callback_.Run(offset_, 0);
+}
+void DataSinkReceiver::DataFrame::ReportError(uint32_t bytes_read,
+                                              int32_t error) {
+  offset_ += bytes_read;
+  DCHECK_LE(offset_, data_.size());
+  callback_.Run(offset_, error);
 }
 
 }  // namespace device