Initial implementation of the ByteStream refactor.

content/browser/download/byte_stream.cc

Index: content/browser/download/byte_stream.cc
diff --git a/content/browser/download/byte_stream.cc b/content/browser/download/byte_stream.cc
new file mode 100644
index 0000000000000000000000000000000000000000..fc41262e49c6c01838467e3aba2d4b3f57225cca
--- /dev/null
+++ b/content/browser/download/byte_stream.cc
@@ -0,0 +1,155 @@
+// 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"
+
+namespace content {
+
+ByteStream::ByteStream()
+    : buffer_size_(kDefaultBufferSize),
+      data_size_(0),
+      is_complete_(false),
+      source_status_(DOWNLOAD_INTERRUPT_REASON_NONE) { }
+
+void ByteStream::SetBufferSize(size_t buffer_size) {
+  buffer_size_ = buffer_size;
+}
+
+bool ByteStream::AddData(scoped_refptr<net::IOBuffer> buffer,
+                         size_t byte_count) {
+  base::AutoLock auto_lock(lock_);
+
+  if (data_size_ == 0 && byte_count > 0 && sink_task_runner_.get() != NULL)

[ahendrickson, 2012/04/16 15:14:27]

Nit: Add braces for multi-line 'if' bodies.

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

Ooops.  But already done in a later PS.

+    // Nothing actually touches the data on this object can be

[ahendrickson, 2012/04/16 15:14:27]

*Nothing that

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

Done.

+    // executed until we drop the lock, so it's ok to
+    // dispatch this before we actually add the data.
+    sink_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&content::ByteStream::RunSinkCallback,
+                              this, sink_task_runner_));
+
+  // Take manual (not enforced by compiler via scoped_*) ownership of data.
+  contents_.push_back(std::make_pair(buffer, byte_count));
+  data_size_ += byte_count;
+  return (data_size_ < buffer_size_);
+}
+
+void ByteStream::SourceComplete(DownloadInterruptReason status) {
+  base::AutoLock auto_lock(lock_);
+
+  is_complete_ = true;
+  source_status_ = status;
+  // If contents_ is non-empty, a callback has already been posted.
+  if (contents_.empty() && sink_task_runner_.get() != NULL)

[ahendrickson, 2012/04/16 15:14:27]

Nit:  Add braces.

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

Done.

+    sink_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&content::ByteStream::RunSinkCallback,
+                              this, sink_task_runner_));
+}
+
+bool ByteStream::IsFull() {
+  base::AutoLock auto_lock(lock_);
+
+  return (data_size_ >= buffer_size_);
+}
+
+void ByteStream::RegisterSourceCallback(
+    scoped_refptr<base::TaskRunner> source_task_runner,
+    ByteStreamCallback source_callback) {
+  base::AutoLock auto_lock(lock_);
+
+  source_task_runner_ = source_task_runner;
+  source_callback_ = source_callback;
+}
+
+ByteStream::StreamState ByteStream::GetData(scoped_refptr<net::IOBuffer>* data,
+                                            size_t* length) {
+  base::AutoLock auto_lock(lock_);
+
+  if (contents_.empty()) {
+    if (is_complete_)
+      return STREAM_COMPLETE;
+    return STREAM_EMPTY;
+  }
+
+  if (data_size_ >= buffer_size_ &&
+      data_size_ - contents_.front().second < buffer_size_ &&
+      source_task_runner_.get() != NULL)

[ahendrickson, 2012/04/16 15:14:27]

Nit:  Add braces.

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

Done.

+    // Nothing actually touches the data on this object can be

[ahendrickson, 2012/04/16 15:14:27]

*Nothing that

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

Done.

+    // executed until we drop the lock, so it's ok to
+    // dispatch this before we actually take the data.
+    source_task_runner_->PostTask(FROM_HERE, base::Bind(
+        &content::ByteStream::RunSourceCallback, this, source_task_runner_));
+
+  *data = contents_.front().first;
+  *length = contents_.front().second;
+  contents_.pop_front();
+  data_size_ -= *length;
+  return STREAM_NON_EMPTY;
+}
+
+DownloadInterruptReason ByteStream::GetSourceResult () {
+  base::AutoLock auto_lock(lock_);
+
+  return source_status_;
+}
+
+void ByteStream::RegisterSinkCallback(
+    scoped_refptr<base::TaskRunner> sink_task_runner,
+    ByteStreamCallback sink_callback) {
+  base::AutoLock auto_lock(lock_);
+
+  sink_task_runner_ = sink_task_runner;
+  sink_callback_ = sink_callback;
+}
+
+ByteStream::~ByteStream() {
+}
+
+void ByteStream::RunSourceCallback(
+    scoped_refptr<base::TaskRunner> target_runner) {
+  base::Closure callback;
+  {
+    base::AutoLock auto_lock(lock_);
+
+    // If the target_runner has been updated, that implies an access to
+    // the class, so we can drop this request on the floor.  (The
+    // alternative would be to forward it on to the new task runner.)
+    if (target_runner.get() != source_task_runner_.get())
+      return;
+
+    if (source_callback_.is_null())
+      return;
+
+    callback = source_callback_;
+  }
+
+  // Run unlocked to allow caller to call back into us.
+  callback.Run();
+}
+
+void ByteStream::RunSinkCallback(
+    scoped_refptr<base::TaskRunner> target_runner) {
+  base::Closure callback;
+  {
+    base::AutoLock auto_lock(lock_);
+
+    // If the target_runner has been updated, that implies an access to
+    // the class, so we can drop this request on the floor.  (The
+    // alternative would be to forward it on to the new task runner.)
+    if (target_runner.get() != sink_task_runner_.get())
+      return;
+
+    if (sink_callback_.is_null())
+      return;
+
+    callback = sink_callback_;
+  }
+
+  // Run unlocked to allow caller to call back into us.
+  callback.Run();
+}
+
+}  // namespace content