[ON HOLD] Implement pull-based design for content decoding

net/filter/filter_stream_source.cc

Index: net/filter/filter_stream_source.cc
diff --git a/net/filter/filter_stream_source.cc b/net/filter/filter_stream_source.cc
new file mode 100644
index 0000000000000000000000000000000000000000..a2fd2a0c4a3ef5ea2069f328525ca27e4bf75b66
--- /dev/null
+++ b/net/filter/filter_stream_source.cc
@@ -0,0 +1,160 @@
+// 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_stream_source.h"
+
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/metrics/histogram_macros.h"
+#include "base/numerics/safe_conversions.h"
+#include "base/strings/string_util.h"
+
+namespace net {
+
+namespace {
+
+const size_t kBufferSize = 32 * 1024;
+
+}  // namespace
+
+FilterStreamSource::FilterStreamSource(SourceType type,
+                                       std::unique_ptr<StreamSource> previous)
+    : StreamSource(type),
+      previous_(std::move(previous)),
+      next_state_(STATE_NONE),
+      output_buffer_size_(0),
+      previous_eof_reached_(false) {
+  DCHECK(previous_);
+}
+
+FilterStreamSource::~FilterStreamSource() {}
+
+int FilterStreamSource::Read(IOBuffer* read_buffer,
+                             size_t read_buffer_size,
+                             const CompletionCallback& callback) {
+  // Allocate a BlockBuffer during first Read().
+  if (!input_buffer_) {
+    input_buffer_ = new IOBufferWithSize(kBufferSize);
+    drainable_input_buffer_ = new DrainableIOBuffer(input_buffer_.get(), 0);
+  }
+  // Start with filtering data, which tells us whether it needs input data.
+  next_state_ = STATE_FILTER_DATA;
+
+  output_buffer_ = read_buffer;
+  output_buffer_size_ = read_buffer_size;
+  int rv = DoLoop(OK);
+  if (rv > OK) {
+    return rv;
+  } else if (rv == ERR_IO_PENDING) {
+    callback_ = callback;
+  }
+  return static_cast<Error>(rv);
+}
+
+std::string FilterStreamSource::OrderedTypeStringList() const {
+  return previous_->OrderedTypeStringList() + "," + GetTypeAsString();
+}
+
+bool FilterStreamSource::Init() {
+  return true;
+}
+
+int FilterStreamSource::DoLoop(int result) {
+  CHECK(this);
+  DCHECK_NE(STATE_NONE, next_state_);
+  int rv = result;
+  do {
+    State state = next_state_;
+    next_state_ = STATE_NONE;
+    switch (state) {
+      case STATE_READ_DATA:
+        rv = DoReadData();
+        break;
+      case STATE_READ_DATA_COMPLETE:
+        rv = DoReadDataComplete(rv);
+        break;
+      case STATE_FILTER_DATA:
+        rv = DoFilterData(rv);
+        break;
+      case STATE_FILTER_DATA_COMPLETE:
+        rv = DoFilterDataComplete(rv);
+        break;
+      default:
+        NOTREACHED() << "bad state: " << state;
+        rv = ERR_UNEXPECTED;
+        break;
+    }
+  } while (rv != ERR_IO_PENDING && next_state_ != STATE_NONE);
+  return rv;
+}
+
+int FilterStreamSource::DoReadData() {
+  DCHECK_EQ(0, drainable_input_buffer_->BytesRemaining());
+  // Use base::Unretained here is safe because |this| owns |previous_|.
+  int rv =
+      previous_->Read(input_buffer_.get(), kBufferSize,
+                      base::Bind(&FilterStreamSource::OnPreviousReadCompleted,
+                                 base::Unretained(this)));
+
+  if (rv != ERR_IO_PENDING)
+    next_state_ = STATE_READ_DATA_COMPLETE;
+  return rv;
+}
+
+int FilterStreamSource::DoReadDataComplete(int result) {
+  CHECK_NE(ERR_IO_PENDING, result);
+
+  if (result > OK) {
+    drainable_input_buffer_ = new DrainableIOBuffer(
+        input_buffer_.get(), base::checked_cast<size_t>(result));
+    next_state_ = STATE_FILTER_DATA;
+  } else {
+    previous_eof_reached_ = true;
+  }
+  return result;
+}
+
+void FilterStreamSource::OnPreviousReadCompleted(int result) {
+  next_state_ = STATE_READ_DATA_COMPLETE;
+  int rv = DoLoop(result);
+  if (rv != ERR_IO_PENDING)
+    DoCallback(rv);
+}
+
+int FilterStreamSource::DoFilterData(int result) {
+  DCHECK(output_buffer_);
+  CHECK_LE(0, result);
+
+  int bytes_output = FilterData(output_buffer_.get(), output_buffer_size_,
+                                drainable_input_buffer_.get());
+  if (bytes_output == ERR_CONTENT_DECODING_FAILED) {
+    UMA_HISTOGRAM_ENUMERATION("Net.ContentDecodingFailed.FilterType", type(),
+                              TYPE_MAX);
+  }
+  // FilterData() is not allowed to return ERR_IO_PENDING.
+  CHECK_NE(ERR_IO_PENDING, bytes_output);
+
+  next_state_ = STATE_FILTER_DATA_COMPLETE;
+  return bytes_output;
+}
+
+int FilterStreamSource::DoFilterDataComplete(int result) {
+  DCHECK_NE(ERR_IO_PENDING, result);
+
+  if (result == OK && !previous_eof_reached_)
+    next_state_ = STATE_READ_DATA;
+  return result;
+}
+
+void FilterStreamSource::DoCallback(int result) {
+  CHECK_NE(ERR_IO_PENDING, result);
+
+  output_buffer_ = nullptr;
+  output_buffer_size_ = 0;
+
+  DCHECK(!callback_.is_null());
+  base::ResetAndReturn(&callback_).Run(result);
+}
+
+}  // namespace net