Use ByteStream in downloads system to decouple source and sink.

content/browser/download/byte_stream.cc

 // Copyright (c) 2012 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 "content/browser/download/byte_stream.h"
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/memory/weak_ptr.h"
 #include "base/memory/ref_counted.h"
 #include "base/sequenced_task_runner.h"
 
 namespace {
 
 typedef std::deque<std::pair<scoped_refptr<net::IOBuffer>, size_t> >
 ContentVector;
 
 // The fraction of the buffer that must be ready to send on the input
 // before we ship data to the output.
 static const int kFractionBufferBeforeSending = 3;
 
 // The fraction of the buffer that must have been consumed on the output
 // before we update the input window.
 static const int kFractionReadBeforeWindowUpdate = 3;
 
-class ByteStreamOutputImpl;
+class ByteStreamReaderImpl;
 
 // A poor man's weak pointer; a RefCountedThreadSafe boolean that can be
 // cleared in an object destructor and accessed to check for object
 // existence.  We can't use weak pointers because they're tightly tied to
 // threads rather than task runners.
 // TODO(rdsmith): A better solution would be extending weak pointers
 // to support SequencedTaskRunners.
 struct LifetimeFlag : public base::RefCountedThreadSafe<LifetimeFlag> {
  public:
   LifetimeFlag() : is_alive_(true) { }
   bool is_alive_;
 
  protected:
   friend class base::RefCountedThreadSafe<LifetimeFlag>;
   virtual ~LifetimeFlag() { }
 
  private:
   DISALLOW_COPY_AND_ASSIGN(LifetimeFlag);
 };
 
-// For both ByteStreamInputImpl and ByteStreamOutputImpl, Construction and
+// For both ByteStreamWriterImpl and ByteStreamReaderImpl, Construction and
 // SetPeer may happen anywhere; all other operations on each class must
 // happen in the context of their SequencedTaskRunner.
-class ByteStreamInputImpl : public content::ByteStreamInput {
+class ByteStreamWriterImpl : public content::ByteStreamWriter {
  public:
-  ByteStreamInputImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
+  ByteStreamWriterImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
                       scoped_refptr<LifetimeFlag> lifetime_flag,
                       size_t buffer_size);
-  virtual ~ByteStreamInputImpl();
+  virtual ~ByteStreamWriterImpl();
 
   // Must be called before any operations are performed.
-  void SetPeer(ByteStreamOutputImpl* peer,
+  void SetPeer(ByteStreamReaderImpl* peer,
                scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
                scoped_refptr<LifetimeFlag> peer_lifetime_flag);
 
-  // Overridden from ByteStreamInput.
+  // Overridden from ByteStreamWriter.
   virtual bool Write(scoped_refptr<net::IOBuffer> buffer,
                      size_t byte_count) OVERRIDE;
   virtual void Close(content::DownloadInterruptReason status) OVERRIDE;
   virtual void RegisterCallback(const base::Closure& source_callback) OVERRIDE;
 
-  // PostTask target from |ByteStreamOutputImpl::MaybeUpdateInput|.
+  // PostTask target from |ByteStreamReaderImpl::MaybeUpdateInput|.
   static void UpdateWindow(scoped_refptr<LifetimeFlag> lifetime_flag,
-                           ByteStreamInputImpl* target,
+                           ByteStreamWriterImpl* target,
                            size_t bytes_consumed);
 
  private:
   // Called from UpdateWindow when object existence has been validated.
   void UpdateWindowInternal(size_t bytes_consumed);
 
   void PostToPeer(bool complete, content::DownloadInterruptReason status);
 
   const size_t total_buffer_size_;
 
@@ skipping 14 matching lines @@
 
   // How much we've sent to the output that for flow control purposes we
   // must assume hasn't been read yet.
   size_t output_size_used_;
 
   // Only valid to access on peer_task_runner_.
   scoped_refptr<LifetimeFlag> peer_lifetime_flag_;
 
   // Only valid to access on peer_task_runner_ if
   // |*peer_lifetime_flag_ == true|
-  ByteStreamOutputImpl* peer_;
+  ByteStreamReaderImpl* peer_;
 };
 
-class ByteStreamOutputImpl : public content::ByteStreamOutput {
+class ByteStreamReaderImpl : public content::ByteStreamReader {
  public:
-  ByteStreamOutputImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
+  ByteStreamReaderImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
                        scoped_refptr<LifetimeFlag> lifetime_flag,
                        size_t buffer_size);
-  virtual ~ByteStreamOutputImpl();
+  virtual ~ByteStreamReaderImpl();
 
   // Must be called before any operations are performed.
-  void SetPeer(ByteStreamInputImpl* peer,
+  void SetPeer(ByteStreamWriterImpl* peer,
                scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
                scoped_refptr<LifetimeFlag> peer_lifetime_flag);
 
-  // Overridden from ByteStreamOutput.
+  // Overridden from ByteStreamReader.
   virtual StreamState Read(scoped_refptr<net::IOBuffer>* data,
                            size_t* length) OVERRIDE;
   virtual content::DownloadInterruptReason GetStatus() const OVERRIDE;
   virtual void RegisterCallback(const base::Closure& sink_callback) OVERRIDE;
 
-  // PostTask target from |ByteStreamInputImpl::MaybePostToPeer| and
-  // |ByteStreamInputImpl::Close|.
+  // PostTask target from |ByteStreamWriterImpl::MaybePostToPeer| and
+  // |ByteStreamWriterImpl::Close|.
   // Receive data from our peer.
   // static because it may be called after the object it is targeting
   // has been destroyed.  It may not access |*target|
   // if |*object_lifetime_flag| is false.
   static void TransferData(
       scoped_refptr<LifetimeFlag> object_lifetime_flag,
-      ByteStreamOutputImpl* target,
+      ByteStreamReaderImpl* target,
       scoped_ptr<ContentVector> transfer_buffer,
       size_t transfer_buffer_bytes,
       bool source_complete,
       content::DownloadInterruptReason status);
 
  private:
   // Called from TransferData once object existence has been validated.
   void TransferDataInternal(
       scoped_ptr<ContentVector> transfer_buffer,
       size_t transfer_buffer_bytes,
@@ skipping 26 matching lines @@
 
   // How much has been removed from this class that we haven't told
   // the input about yet.
   size_t unreported_consumed_bytes_;
 
   // Only valid to access on peer_task_runner_.
   scoped_refptr<LifetimeFlag> peer_lifetime_flag_;
 
   // Only valid to access on peer_task_runner_ if
   // |*peer_lifetime_flag_ == true|
-  ByteStreamInputImpl* peer_;
+  ByteStreamWriterImpl* peer_;
 };
 
-ByteStreamInputImpl::ByteStreamInputImpl(
+ByteStreamWriterImpl::ByteStreamWriterImpl(
     scoped_refptr<base::SequencedTaskRunner> task_runner,
     scoped_refptr<LifetimeFlag> lifetime_flag,
     size_t buffer_size)
     : total_buffer_size_(buffer_size),
       my_task_runner_(task_runner),
       my_lifetime_flag_(lifetime_flag),
       input_contents_size_(0),
       output_size_used_(0),
       peer_(NULL) {
   DCHECK(my_lifetime_flag_.get());
   my_lifetime_flag_->is_alive_ = true;
 }
 
-ByteStreamInputImpl::~ByteStreamInputImpl() {
+ByteStreamWriterImpl::~ByteStreamWriterImpl() {
   my_lifetime_flag_->is_alive_ = false;
 }
 
-void ByteStreamInputImpl::SetPeer(
-    ByteStreamOutputImpl* peer,
+void ByteStreamWriterImpl::SetPeer(
+    ByteStreamReaderImpl* peer,
     scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
     scoped_refptr<LifetimeFlag> peer_lifetime_flag) {
   peer_ = peer;
   peer_task_runner_ = peer_task_runner;
   peer_lifetime_flag_ = peer_lifetime_flag;
 }
 
-bool ByteStreamInputImpl::Write(
+bool ByteStreamWriterImpl::Write(
     scoped_refptr<net::IOBuffer> buffer, size_t byte_count) {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
 
   input_contents_.push_back(std::make_pair(buffer, byte_count));
   input_contents_size_ += byte_count;
 
   // Arbitrarily, we buffer to a third of the total size before sending.
   if (input_contents_size_ > total_buffer_size_ / kFractionBufferBeforeSending)
     PostToPeer(false, content::DOWNLOAD_INTERRUPT_REASON_NONE);
 
   return (input_contents_size_ + output_size_used_ <= total_buffer_size_);
 }
 
-void ByteStreamInputImpl::Close(
+void ByteStreamWriterImpl::Close(
     content::DownloadInterruptReason status) {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
   PostToPeer(true, status);
 }
 
-void ByteStreamInputImpl::RegisterCallback(
+void ByteStreamWriterImpl::RegisterCallback(
     const base::Closure& source_callback) {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
   space_available_callback_ = source_callback;
 }
 
 // static
-void ByteStreamInputImpl::UpdateWindow(
-    scoped_refptr<LifetimeFlag> lifetime_flag, ByteStreamInputImpl* target,
+void ByteStreamWriterImpl::UpdateWindow(
+    scoped_refptr<LifetimeFlag> lifetime_flag, ByteStreamWriterImpl* target,
     size_t bytes_consumed) {
   // If the target object isn't alive anymore, we do nothing.
   if (!lifetime_flag->is_alive_) return;
 
   target->UpdateWindowInternal(bytes_consumed);
 }
 
-void ByteStreamInputImpl::UpdateWindowInternal(size_t bytes_consumed) {
+void ByteStreamWriterImpl::UpdateWindowInternal(size_t bytes_consumed) {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
   DCHECK_GE(output_size_used_, bytes_consumed);
   output_size_used_ -= bytes_consumed;
 
   // Callback if we were above the limit and we're now <= to it.
   size_t total_known_size_used =
       input_contents_size_ + output_size_used_;
 
   if (total_known_size_used <= total_buffer_size_ &&
       (total_known_size_used + bytes_consumed > total_buffer_size_) &&
       !space_available_callback_.is_null())
     space_available_callback_.Run();
 }
 
-void ByteStreamInputImpl::PostToPeer(
+void ByteStreamWriterImpl::PostToPeer(
     bool complete, content::DownloadInterruptReason status) {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
   // Valid contexts in which to call.
   DCHECK(complete || 0 != input_contents_size_);
 
   scoped_ptr<ContentVector> transfer_buffer(new ContentVector);
   size_t buffer_size = 0;
   if (0 != input_contents_size_) {
     transfer_buffer.reset(new ContentVector);
     transfer_buffer->swap(input_contents_);
     buffer_size = input_contents_size_;
     output_size_used_ += input_contents_size_;
     input_contents_size_ = 0;
   }
   peer_task_runner_->PostTask(
       FROM_HERE, base::Bind(
-          &ByteStreamOutputImpl::TransferData,
+          &ByteStreamReaderImpl::TransferData,
           peer_lifetime_flag_,
           peer_,
           base::Passed(transfer_buffer.Pass()),
           buffer_size,
           complete,
           status));
 }
 
-ByteStreamOutputImpl::ByteStreamOutputImpl(
+ByteStreamReaderImpl::ByteStreamReaderImpl(
     scoped_refptr<base::SequencedTaskRunner> task_runner,
     scoped_refptr<LifetimeFlag> lifetime_flag,
     size_t buffer_size)
     : total_buffer_size_(buffer_size),
       my_task_runner_(task_runner),
       my_lifetime_flag_(lifetime_flag),
       received_status_(false),
       status_(content::DOWNLOAD_INTERRUPT_REASON_NONE),
       unreported_consumed_bytes_(0),
       peer_(NULL) {
   DCHECK(my_lifetime_flag_.get());
   my_lifetime_flag_->is_alive_ = true;
 }
 
-ByteStreamOutputImpl::~ByteStreamOutputImpl() {
+ByteStreamReaderImpl::~ByteStreamReaderImpl() {
   my_lifetime_flag_->is_alive_ = false;
 }
 
-void ByteStreamOutputImpl::SetPeer(
-    ByteStreamInputImpl* peer,
+void ByteStreamReaderImpl::SetPeer(
+    ByteStreamWriterImpl* peer,
     scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
     scoped_refptr<LifetimeFlag> peer_lifetime_flag) {
   peer_ = peer;
   peer_task_runner_ = peer_task_runner;
   peer_lifetime_flag_ = peer_lifetime_flag;
 }
 
-ByteStreamOutputImpl::StreamState
-ByteStreamOutputImpl::Read(scoped_refptr<net::IOBuffer>* data,
+ByteStreamReaderImpl::StreamState
+ByteStreamReaderImpl::Read(scoped_refptr<net::IOBuffer>* data,
                            size_t* length) {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
 
   if (available_contents_.size()) {
     *data = available_contents_.front().first;
     *length = available_contents_.front().second;
     available_contents_.pop_front();
     unreported_consumed_bytes_ += *length;
 
     MaybeUpdateInput();
     return STREAM_HAS_DATA;
   }
   if (received_status_) {
     return STREAM_COMPLETE;
   }
   return STREAM_EMPTY;
 }
 
 content::DownloadInterruptReason
-ByteStreamOutputImpl::GetStatus() const {
+ByteStreamReaderImpl::GetStatus() const {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
   DCHECK(received_status_);
   return status_;
 }
 
-void ByteStreamOutputImpl::RegisterCallback(
+void ByteStreamReaderImpl::RegisterCallback(
     const base::Closure& sink_callback) {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
 
   data_available_callback_ = sink_callback;
 }
 
 // static
-void ByteStreamOutputImpl::TransferData(
+void ByteStreamReaderImpl::TransferData(
     scoped_refptr<LifetimeFlag> object_lifetime_flag,
-    ByteStreamOutputImpl* target,
+    ByteStreamReaderImpl* target,
     scoped_ptr<ContentVector> transfer_buffer,
     size_t buffer_size,
     bool source_complete,
     content::DownloadInterruptReason status) {
   // If our target is no longer alive, do nothing.
   if (!object_lifetime_flag->is_alive_) return;
 
   target->TransferDataInternal(
       transfer_buffer.Pass(), buffer_size, source_complete, status);
 }
 
-void ByteStreamOutputImpl::TransferDataInternal(
+void ByteStreamReaderImpl::TransferDataInternal(
     scoped_ptr<ContentVector> transfer_buffer,
     size_t buffer_size,
     bool source_complete,
     content::DownloadInterruptReason status) {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
 
   bool was_empty = available_contents_.empty();
 
   if (transfer_buffer.get()) {
     available_contents_.insert(available_contents_.end(),
@@ skipping 10 matching lines @@
   // source complete.
   if (((was_empty && !available_contents_.empty()) ||
        source_complete) &&
       !data_available_callback_.is_null())
     data_available_callback_.Run();
 }
 
 // Decide whether or not to send the input a window update.
 // Currently we do that whenever we've got unreported consumption
 // greater than 1/3 of total size.
-void ByteStreamOutputImpl::MaybeUpdateInput() {
+void ByteStreamReaderImpl::MaybeUpdateInput() {
   DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
 
   if (unreported_consumed_bytes_ <=
       total_buffer_size_ / kFractionReadBeforeWindowUpdate)
     return;
 
   peer_task_runner_->PostTask(
       FROM_HERE, base::Bind(
-          &ByteStreamInputImpl::UpdateWindow,
+          &ByteStreamWriterImpl::UpdateWindow,
           peer_lifetime_flag_,
           peer_,
           unreported_consumed_bytes_));
   unreported_consumed_bytes_ = 0;
 }
 
 }  // namespace
 
 namespace content {
 
-ByteStreamOutput::~ByteStreamOutput() { }
+ByteStreamReader::~ByteStreamReader() { }
 
-ByteStreamInput::~ByteStreamInput() { }
+ByteStreamWriter::~ByteStreamWriter() { }
 
 void CreateByteStream(
     scoped_refptr<base::SequencedTaskRunner> input_task_runner,
     scoped_refptr<base::SequencedTaskRunner> output_task_runner,
     size_t buffer_size,
-    scoped_ptr<ByteStreamInput>* input,
-    scoped_ptr<ByteStreamOutput>* output) {
+    scoped_ptr<ByteStreamWriter>* input,
+    scoped_ptr<ByteStreamReader>* output) {
   scoped_refptr<LifetimeFlag> input_flag(new LifetimeFlag());
   scoped_refptr<LifetimeFlag> output_flag(new LifetimeFlag());
 
-  ByteStreamInputImpl* in = new ByteStreamInputImpl(
+  ByteStreamWriterImpl* in = new ByteStreamWriterImpl(
       input_task_runner, input_flag, buffer_size);
-  ByteStreamOutputImpl* out = new ByteStreamOutputImpl(
+  ByteStreamReaderImpl* out = new ByteStreamReaderImpl(
       output_task_runner, output_flag, buffer_size);
 
   in->SetPeer(out, output_task_runner, output_flag);
   out->SetPeer(in, input_task_runner, input_flag);
   input->reset(in);
   output->reset(out);
 }
 
 }  // namespace content