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..bffc895e1485f75a46b739e4c3e7163fa9c71624
--- /dev/null
+++ b/net/filter/filter_source_stream.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_source_stream.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
+
+FilterSourceStream::FilterSourceStream(SourceType type,
+                                       std::unique_ptr<SourceStream> upstream)
+    : SourceStream(type),
+      upstream_(std::move(upstream)),
+      next_state_(STATE_NONE),
+      output_buffer_size_(0),
+      upstream_end_reached_(false) {
+  DCHECK(upstream_);
+}
+
+FilterSourceStream::~FilterSourceStream() {}
+
+int FilterSourceStream::Read(IOBuffer* read_buffer,
+                             size_t read_buffer_size,
+                             const CompletionCallback& callback) {

[mmenke, 2016/08/17 19:01:55]

DCHECK_EQ(STATE_NONE, next_state_);

[mmenke, 2016/08/17 19:01:56]

Maybe also add:

DCHECK(read_buffer);
DCHECK_LT(0, read_buffer_size);

We could fail rather far down the line if they're nullptr/0.

[xunjieli, 2016/08/17 20:00:19]

Done.

[xunjieli, 2016/08/17 20:00:19]

Done.

+  // Allocate a BlockBuffer during first Read().
+  if (!input_buffer_)
+    input_buffer_ = new IOBufferWithSize(kBufferSize);

[mmenke, 2016/08/17 19:01:55]

Should we really be taking read_buffer_size as a size_t?  If
(static_cast<int>read_buffer) is <= 0, we're in big trouble.

[xunjieli, 2016/08/17 20:00:19]

Done. I will keep it as int then to conform to the other Read() patterns in
net/.

+
+  // Start with filtering data, which tells us whether it needs input data.

[mmenke, 2016/08/17 19:01:56]

it -> this filter?

[xunjieli, 2016/08/17 20:00:19]

Done.

+  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 rv;

[mmenke, 2016/08/17 19:01:55]

this could be simplified as:

if (rv == ERR_IO_PENDING)
  callback_ = callback;
return rv;

i.e. remove the rv > OK case.

[xunjieli, 2016/08/17 20:00:19]

Done.

+}
+
+std::string FilterSourceStream::OrderedTypeStringList() const {
+  std::string next_type_string = upstream_->OrderedTypeStringList();
+  if (next_type_string.empty())
+    return GetTypeAsString();
+  return next_type_string + "," + GetTypeAsString();
+}
+
+int FilterSourceStream::DoLoop(int result) {
+  DCHECK(this);

[mmenke, 2016/08/17 19:01:56]

I don't think this is needed?

[xunjieli, 2016/08/17 20:00:19]

Done. Sorry I put it there and forgot to remove it.

+  DCHECK_NE(STATE_NONE, next_state_);

[mmenke, 2016/08/17 19:01:56]

Suggest a blank line between initial-state DCHECKs and the method body.

[xunjieli, 2016/08/17 20:00:19]

Done.

+  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;
+      default:
+        NOTREACHED() << "bad state: " << state;
+        rv = ERR_UNEXPECTED;
+        break;
+    }
+  } while (rv != ERR_IO_PENDING && next_state_ != STATE_NONE);
+  return rv;
+}
+
+int FilterSourceStream::DoReadData() {
+  // Read more data means subclasses have consumed all input or this is the
+  // first read in which case the |drainable_input_buffer_| is not initialized.
+  DCHECK(drainable_input_buffer_ == nullptr ||
+         0 == drainable_input_buffer_->BytesRemaining());

[mmenke, 2016/08/17 19:01:55]

Suggest a blank line after the DCHECK.

[xunjieli, 2016/08/17 20:00:19]

Done.

+  // Use base::Unretained here is safe because |this| owns |upstream_|.
+  int rv = upstream_->Read(input_buffer_.get(), kBufferSize,
+                           base::Bind(&FilterSourceStream::OnNextReadCompleted,
+                                      base::Unretained(this)));
+
+  if (rv != ERR_IO_PENDING)
+    next_state_ = STATE_READ_DATA_COMPLETE;
+  return rv;
+}
+
+int FilterSourceStream::DoReadDataComplete(int result) {
+  DCHECK_NE(ERR_IO_PENDING, result);
+
+  // If EOF is read (result == OK), also pass that to the the filter.
+  if (result >= OK) {
+    drainable_input_buffer_ =
+        new DrainableIOBuffer(input_buffer_.get(), result);
+    next_state_ = STATE_FILTER_DATA;
+  }
+  if (result <= OK)
+    upstream_end_reached_ = true;
+  return result;
+}
+
+void FilterSourceStream::OnNextReadCompleted(int result) {
+  next_state_ = STATE_READ_DATA_COMPLETE;
+  int rv = DoLoop(result);
+  if (rv != ERR_IO_PENDING)
+    DoCallback(rv);
+}
+
+int FilterSourceStream::DoFilterData(int result) {
+  DCHECK(output_buffer_);
+  DCHECK_LE(0, result);

[mmenke, 2016/08/17 19:01:55]

This method doesn't actually use result.  Suggest getting rid of it as an
argument, and moving the DCHECK into the case statement (That's generally what
we do for unused results in DoLoops, I believe)

[xunjieli, 2016/08/17 20:00:19]

Done.

+
+  // This is first Read(), short circuit it and go straight to read data from
+  // |upstream_|.
+  if (drainable_input_buffer_ == nullptr) {

[mmenke, 2016/08/17 19:01:55]

Can we just get rid of this check?  We have a check if input_buffer_ if nullptr
in ::Read.  Can't we just set next_state_ to STATE_READ_DATA there, instead, and
remove this entirely?  Can then add a DCHECK(drainable_input_buffer_) here, too.

[xunjieli, 2016/08/17 20:00:19]

Done.

+    next_state_ = STATE_READ_DATA;
+    return OK;
+  }
+
+  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.
+  DCHECK_NE(ERR_IO_PENDING, bytes_output);
+
+  // If data can still be read from |upstream_| and filter did not return any
+  // data, it is likely that the filter needs more input.
+  if (bytes_output == OK && !upstream_end_reached_)
+    next_state_ = STATE_READ_DATA;
+  return bytes_output;
+}
+
+void FilterSourceStream::DoCallback(int result) {

[mmenke, 2016/08/17 19:01:55]

Should we just inline this?  We only call it in one place.

[xunjieli, 2016/08/17 20:00:19]

Done.

+  DCHECK_NE(ERR_IO_PENDING, result);
+  DCHECK(!callback_.is_null());
+
+  output_buffer_ = nullptr;
+  output_buffer_size_ = 0;
+
+  base::ResetAndReturn(&callback_).Run(result);
+}
+
+}  // namespace net