Add net::SourceStream and net::FilterSourceStream

net/filter/filter_source_stream.cc

Index: net/filter/filter_source_stream.cc
diff --git a/net/filter/filter_source_stream.cc b/net/filter/filter_source_stream.cc
new file mode 100644
index 0000000000000000000000000000000000000000..484316662dae7a2ac6fa640e2be732257c177a3f
--- /dev/null
+++ b/net/filter/filter_source_stream.cc
@@ -0,0 +1,90 @@
+// Copyright 2016 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 "net/filter/filter_source_stream.h"
+
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/logging.h"
+#include "base/metrics/histogram_macros.h"
+#include "base/numerics/safe_conversions.h"
+#include "base/strings/string_util.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+
+namespace net {
+
+namespace {
+
+const size_t kBufferSize = 32 * 1024;
+
+}  // namespace
+
+FilterSourceStream::FilterSourceStream(SourceType type,
+                                       std::unique_ptr<SourceStream> upstream)
+    : SourceStream(type),
+      upstream_(std::move(upstream)),
+      upstream_end_reached_(false) {
+  DCHECK(upstream_);
+}
+
+FilterSourceStream::~FilterSourceStream() {}
+
+int FilterSourceStream::Read(IOBuffer* read_buffer,
+                             int read_buffer_size,
+                             const CompletionCallback& callback) {
+  DCHECK(read_buffer);
+  DCHECK_LT(0, read_buffer_size);
+
+  return ReadInternal(read_buffer, read_buffer_size, callback);
+}
+
+std::string FilterSourceStream::Description() const {
+  std::string next_type_string = upstream_->Description();
+  if (next_type_string.empty())
+    return GetTypeAsString();
+  return next_type_string + "," + GetTypeAsString();
+}
+
+int FilterSourceStream::Pull(const CompletionCallback& callback) {
+  DCHECK(nullptr == drainable_input_buffer_ ||
+         0 == drainable_input_buffer_->BytesRemaining());
+  DCHECK(pull_callback_.is_null());
+
+  // Allocate a BlockBuffer during first Read().
+  if (!input_buffer_)
+    input_buffer_ = new IOBufferWithSize(kBufferSize);
+
+  // Use base::Unretained here is safe because |this| owns |upstream_|.
+  int rv = upstream_->Read(
+      input_buffer_.get(), kBufferSize,
+      base::Bind(&FilterSourceStream::OnIOComplete, base::Unretained(this)));
+
+  if (rv == ERR_IO_PENDING)
+    pull_callback_ = callback;
+
+  MaybeFillDrainableInputBuffer(rv);
+  return rv;
+}
+
+void FilterSourceStream::OnIOComplete(int result) {
+  DCHECK(!pull_callback_.is_null());
+  DCHECK_NE(ERR_IO_PENDING, result);
+
+  MaybeFillDrainableInputBuffer(result);
+  base::ResetAndReturn(&pull_callback_).Run(result);
+}
+
+void FilterSourceStream::MaybeFillDrainableInputBuffer(int bytes_read) {
+  if (bytes_read == ERR_IO_PENDING)
+    return;
+  if (bytes_read <= 0)
+    upstream_end_reached_ = true;
+  if (bytes_read < 0)
+    return;
+  drainable_input_buffer_ =
+      new DrainableIOBuffer(input_buffer_.get(), bytes_read);
+}
+
+}  // namespace net