Add net::SourceStream and net::FilterSourceStream

net/filter/filter_source_stream.cc

+// 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)),
+      next_state_(STATE_NONE),
+      output_buffer_size_(0),
+      upstream_end_reached_(false) {
+  DCHECK(upstream_);
+}
+
+FilterSourceStream::~FilterSourceStream() {}
+
+int FilterSourceStream::Read(IOBuffer* read_buffer,
+                             int read_buffer_size,
+                             const CompletionCallback& callback) {
+  DCHECK_EQ(STATE_NONE, next_state_);
+  DCHECK(read_buffer);
+  DCHECK_LT(0, read_buffer_size);
+
+  // Allocate a BlockBuffer during first Read().
+  if (!input_buffer_) {
+    input_buffer_ = new IOBufferWithSize(kBufferSize);
+    // This is first Read(), start with reading data from |upstream_|.
+    next_state_ = STATE_READ_DATA;
+  } else {
+    // Otherwise start with filtering data, which will tell us whether this
+    // stream needs input data.
+    next_state_ = STATE_FILTER_DATA;
+  }
+
+  output_buffer_ = read_buffer;
+  output_buffer_size_ = read_buffer_size;
+  int rv = DoLoop(OK);
+
+  if (rv == ERR_IO_PENDING)
+    callback_ = callback;
+  return rv;
+}
+
+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::DoLoop(int result) {
+  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:
+        DCHECK_LE(0, rv);
+        rv = DoFilterData();
+        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());
+
+  next_state_ = STATE_READ_DATA_COMPLETE;
+  // 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)));
+
+  return rv;
+}
+
+int FilterSourceStream::DoReadDataComplete(int result) {
+  DCHECK_NE(ERR_IO_PENDING, result);
+
+  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;
+}
+
+int FilterSourceStream::DoFilterData() {
+  DCHECK(output_buffer_);
+  DCHECK(drainable_input_buffer_);
+
+  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);
+
+  // Received data or encountered an error.
+  if (bytes_output != 0)
+    return bytes_output;
+  // If no data is returned, continue reading if |this| needs more input.
+  if (NeedMoreData())
+    next_state_ = STATE_READ_DATA;

[Randy Smith (Not in Mondays), 2016/09/08 18:57:41]

nit, suggestion: I know it's redundant with the DCHECK in DoReadData(), but I'd
sorta like to have a DCHECK here confirming that all data's been drained from
drainable_input_buffer_, just so a failure will be close to the cause of the
failure and linked to it?  But up to you.

[xunjieli, 2016/09/08 19:11:43]

Done. 

+  // TODO(xunjieli): if NeedMoreData() but eof is reached, emit an error.

[Randy Smith (Not in Mondays), 2016/09/08 18:57:41]

I don't think this is accurate, though I totally see where the confusion comes
from and maybe that means we should update the documentation.

The use case I see "NeedsMoreData()" as targeting is for the rare filter that
has output all its going to output no matter what the upstream provides.  SDCH's
meta-refresh is an (ugly) example; once SDCH decides that it can't decode, it
outputs meta-refresh and has no need to read anything else, so will start
returning false from NeedsMoreData().

Most filters will just default "NeedsMoreData()" to true, meaning "If upstream
has more data for me, I can send more data to downstream".  The filters that
want to say "Upstream has no more data for me?  That's an error" will be
returning errors from FilterData when EOF is signalled.

I think that this is reasonably clear from the header documentation, but if you
disagree, please feel free to update it (or if you disagree with this analysis,
let me know, though that generally goes without saying :-}).

[xunjieli, 2016/09/08 19:11:43]

Done.

+  return bytes_output;
+}
+
+void FilterSourceStream::OnIOComplete(int result) {
+  DCHECK_EQ(STATE_READ_DATA_COMPLETE, next_state_);
+
+  int rv = DoLoop(result);
+  if (rv == ERR_IO_PENDING)
+    return;
+
+  output_buffer_ = nullptr;
+  output_buffer_size_ = 0;
+
+  base::ResetAndReturn(&callback_).Run(rv);
+}
+
+bool FilterSourceStream::NeedMoreData() const {
+  return !upstream_end_reached_;
+}
+
+}  // namespace net