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

net/url_request/url_request_job.cc

 // 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 "net/url_request/url_request_job.h"
 
 #include <utility>
 
 #include "base/bind.h"
+#include "base/callback_helpers.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
+#include "base/memory/ptr_util.h"
 #include "base/metrics/histogram_macros.h"
 #include "base/power_monitor/power_monitor.h"
 #include "base/profiler/scoped_tracker.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_split.h"
 #include "base/strings/string_util.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "base/values.h"
 #include "net/base/auth.h"
 #include "net/base/host_port_pair.h"
 #include "net/base/io_buffer.h"
 #include "net/base/load_flags.h"
 #include "net/base/load_states.h"
 #include "net/base/net_errors.h"
 #include "net/base/network_delegate.h"
-#include "net/filter/filter.h"
 #include "net/http/http_response_headers.h"
 #include "net/nqe/network_quality_estimator.h"
 #include "net/url_request/url_request_context.h"
 
 namespace net {
 
 namespace {
 
 // Callback for TYPE_URL_REQUEST_FILTERS_SET net-internals event.
-std::unique_ptr<base::Value> FiltersSetCallback(
-    Filter* filter,
+std::unique_ptr<base::Value> StreamSourceSetCallback(
+    StreamSource* stream_source,
     NetLogCaptureMode /* capture_mode */) {
   std::unique_ptr<base::DictionaryValue> event_params(
       new base::DictionaryValue());
-  event_params->SetString("filters", filter->OrderedFilterList());
+  event_params->SetString("filters", stream_source->OrderedTypeStringList());
   return std::move(event_params);
 }
 
 std::string ComputeMethodForRedirect(const std::string& method,
                                      int http_status_code) {
   // For 303 redirects, all request methods except HEAD are converted to GET,
   // as per the latest httpbis draft.  The draft also allows POST requests to
   // be converted to GETs when following 301/302 redirects, for historical
   // reasons. Most major browsers do this and so shall we.  Both RFC 2616 and
   // the httpbis draft say to prompt the user to confirm the generation of new
@@ skipping 54 matching lines @@
         base::CompareCaseInsensitiveASCII(token, "unsafe-url") == 0) {
       new_policy = URLRequest::NEVER_CLEAR_REFERRER;
       continue;
     }
   }
   return new_policy;
 }
 
 }  // namespace
 
+// StreamSources own the previous StreamSource in the chain, but the ultimate
+// source is URLRequestJob, which has other ownership semantics, so this class
+// is a proxy for URLRequestJob that is owned by the first filter (in dataflow
+// order).
+class URLRequestJob::URLRequestJobStreamSource : public StreamSource {
+ public:
+  explicit URLRequestJobStreamSource(URLRequestJob* job)
+      : StreamSource(StreamSource::TYPE_NONE), job_(job) {}

[mmenke, 2016/07/28 18:40:13]

DCHECK(job_);?  Since we never modify it, DCHECK seems more useful here, if we
want to include one.

[xunjieli, 2016/08/01 16:46:23]

Done.

+
+  ~URLRequestJobStreamSource() override {}
+
+  // StreamSource implementation:
+  int Read(IOBuffer* dest_buffer,
+           size_t buffer_size,
+           const CompletionCallback& callback) override {
+    DCHECK(job_);
+    return job_->ReadRawDataHelper(dest_buffer, buffer_size, callback);
+  }
+
+ private:
+  URLRequestJob* job_;

[mmenke, 2016/07/28 18:40:13]

Maybe URLRequestJob* const job_?  (Meaning the pointer is const, not the job_)

[xunjieli, 2016/08/01 16:46:23]

Done.

+
+  DISALLOW_COPY_AND_ASSIGN(URLRequestJobStreamSource);
+};
+
 URLRequestJob::URLRequestJob(URLRequest* request,
                              NetworkDelegate* network_delegate)
     : request_(request),
       done_(false),
-      prefilter_bytes_read_(0),
-      postfilter_bytes_read_(0),
-      filter_needs_more_output_space_(false),
-      filtered_read_buffer_len_(0),
       has_handled_response_(false),
       expected_content_size_(-1),
       network_delegate_(network_delegate),
       last_notified_total_received_bytes_(0),
       last_notified_total_sent_bytes_(0),
+      prefilter_bytes_read_(0),
+      postfilter_bytes_read_(0),
       weak_factory_(this) {
   base::PowerMonitor* power_monitor = base::PowerMonitor::Get();
   if (power_monitor)
     power_monitor->AddObserver(this);
 }
 
 URLRequestJob::~URLRequestJob() {
   base::PowerMonitor* power_monitor = base::PowerMonitor::Get();
   if (power_monitor)
     power_monitor->RemoveObserver(this);
@@ skipping 15 matching lines @@
   // Kill().
   // TODO(mmenke):  The URLRequest is currently deleted before this method
   // invokes its async callback whenever this is called by the URLRequest.
   // Try to simplify how cancellation works.
   NotifyCanceled();
 }
 
 // This function calls ReadRawData to get stream data. If a filter exists, it
 // passes the data to the attached filter. It then returns the output from
 // filter back to the caller.
+// This method passes reads down the filter chain, where they eventually end up
+// at URLRequestJobStreamSource::Read, which calls back into
+// URLRequestJob::ReadRawData.
 bool URLRequestJob::Read(IOBuffer* buf, int buf_size, int *bytes_read) {
   DCHECK_LT(buf_size, 1000000);  // Sanity check.
   DCHECK(buf);
   DCHECK(bytes_read);
-  DCHECK(!filtered_read_buffer_);
-  DCHECK_EQ(0, filtered_read_buffer_len_);
-
-  Error error = OK;
   *bytes_read = 0;
 
-  // Skip Filter if not present.
-  if (!filter_) {
-    error = ReadRawDataHelper(buf, buf_size, bytes_read);
-  } else {
-    // Save the caller's buffers while we do IO
-    // in the filter's buffers.
-    filtered_read_buffer_ = buf;
-    filtered_read_buffer_len_ = buf_size;
+  pending_read_buffer_ = buf;
+  int result = source_->Read(buf, buf_size,
+                             base::Bind(&URLRequestJob::SourceReadComplete,
+                                        weak_factory_.GetWeakPtr(), false));
+  if (result > 0)
+    *bytes_read = result;
 
-    error = ReadFilteredData(bytes_read);
-
-    // Synchronous EOF from the filter.
-    if (error == OK && *bytes_read == 0)
-      DoneReading();
+  if (result == ERR_IO_PENDING) {
+    SetStatus(URLRequestStatus::FromError(ERR_IO_PENDING));
+    return false;
   }
 
-  if (error == OK) {
-    // If URLRequestJob read zero bytes, the job is at EOF.
-    if (*bytes_read == 0)
-      NotifyDone(URLRequestStatus());
-  } else if (error == ERR_IO_PENDING) {
-    SetStatus(URLRequestStatus::FromError(ERR_IO_PENDING));
+  SourceReadComplete(true, result);
+  return result >= OK;
+}
+
+void URLRequestJob::SourceReadComplete(bool synchronous, int result) {
+  DCHECK_NE(ERR_IO_PENDING, result);
+
+  if (result > 0) {
+    postfilter_bytes_read_ += result;
+    if (request()->net_log().IsCapturing()) {
+      request()->net_log().AddByteTransferEvent(
+          NetLog::TYPE_URL_REQUEST_JOB_FILTERED_BYTES_READ, result,
+          pending_read_buffer_->data());
+    }
+    pending_read_buffer_ = nullptr;

[mmenke, 2016/07/28 18:40:13]

Optional:  May be a little cleaner as:

if (result > 0) {
  // AddByteTransferEvent code goes here
}
pending_read_buffer_ = nullptr;

// All other logic goes here.
if (result > 0) ...

It's no less code, but it makes clearer that the pending_read_buffer_ is done on
all branches, and seems marginally less regression prone to me.

[xunjieli, 2016/08/01 16:46:23]

Done. great suggestion!

+    SetStatus(URLRequestStatus());
+    if (!synchronous)
+      request_->NotifyReadCompleted(result);

[mmenke, 2016/07/28 18:40:13]

I think early return is a little better in each of these cases - we really don't
want to share any code at the end of these branches, and these cases are very
mutually exclusive, so think it's the preferred pattern, in this case.

[xunjieli, 2016/08/01 16:46:23]

Done.

+  } else if (result == 0) {
+    pending_read_buffer_ = nullptr;
+    DoneReading();
+    NotifyDone(URLRequestStatus());
+    if (!synchronous)
+      request_->NotifyReadCompleted(result);
   } else {
-    NotifyDone(URLRequestStatus::FromError(error));
-    *bytes_read = -1;
+    pending_read_buffer_ = nullptr;
+    NotifyDone(URLRequestStatus::FromError(result));
   }
-  return error == OK;
 }
 
 void URLRequestJob::StopCaching() {
   // Nothing to do here.
 }
 
 bool URLRequestJob::GetFullRequestHeaders(HttpRequestHeaders* headers) const {
   // Most job types don't send request headers.
   return false;
 }
@@ skipping 26 matching lines @@
 }
 
 bool URLRequestJob::GetRemoteEndpoint(IPEndPoint* endpoint) const {
   return false;
 }
 
 void URLRequestJob::PopulateNetErrorDetails(NetErrorDetails* details) const {
   return;
 }
 
-std::unique_ptr<Filter> URLRequestJob::SetupFilter() const {
-  return nullptr;
-}
-
 bool URLRequestJob::IsRedirectResponse(GURL* location,
                                        int* http_status_code) {
   // For non-HTTP jobs, headers will be null.
   HttpResponseHeaders* headers = request_->response_headers();
   if (!headers)
     return false;
 
   std::string value;
   if (!headers->IsRedirect(&value))
     return false;
@@ skipping 138 matching lines @@
       return GURL();
     case URLRequest::MAX_REFERRER_POLICY:
       NOTREACHED();
       return GURL();
   }
 
   NOTREACHED();
   return GURL();
 }
 
+int64_t URLRequestJob::prefilter_bytes_read() const {
+  return base::checked_cast<int64_t>(prefilter_bytes_read_);
+}
+
 void URLRequestJob::NotifyCertificateRequested(
     SSLCertRequestInfo* cert_request_info) {
   request_->NotifyCertificateRequested(cert_request_info);
 }
 
 void URLRequestJob::NotifySSLCertificateError(const SSLInfo& ssl_info,
                                               bool fatal) {
   request_->NotifySSLCertificateError(ssl_info, fatal);
 }
 
@@ skipping 68 matching lines @@
     // Need to check for a NULL auth_info because the server may have failed
     // to send a challenge with the 401 response.
     if (auth_info.get()) {
       request_->NotifyAuthRequired(auth_info.get());
       // Wait for SetAuth or CancelAuth to be called.
       return;
     }
   }
 
   has_handled_response_ = true;
-  if (request_->status().is_success())
-    filter_ = SetupFilter();
+  if (request_->status().is_success()) {
+    DCHECK(!source_);
+    source_ = SetupSource();
 
-  if (!filter_.get()) {
-    std::string content_length;
-    request_->GetResponseHeaderByName("content-length", &content_length);
-    if (!content_length.empty())
-      base::StringToInt64(content_length, &expected_content_size_);
-  } else {
-    request_->net_log().AddEvent(
-        NetLog::TYPE_URL_REQUEST_FILTERS_SET,
-        base::Bind(&FiltersSetCallback, base::Unretained(filter_.get())));
+    if (source_ == nullptr) {
+      NotifyDone(URLRequestStatus(URLRequestStatus::FAILED, ERR_FAILED));

[mmenke, 2016/07/28 18:40:13]

Could we add an error code for this case?  Don't need to add any error text, but
a clear way to know about this failure would be useful in identifying the rare
cases when it happens.

[xunjieli, 2016/08/01 16:46:23]

Done.

+      return;
+    }

[mmenke, 2016/07/28 18:40:13]

nit:  Suggest a blank line here.

[xunjieli, 2016/08/01 16:46:23]

Done.

+    if (source_->type() == StreamSource::TYPE_NONE) {
+      std::string content_length;
+      request_->GetResponseHeaderByName("content-length", &content_length);
+      if (!content_length.empty())
+        base::StringToInt64(content_length, &expected_content_size_);
+    } else {
+      request_->net_log().AddEvent(NetLog::TYPE_URL_REQUEST_FILTERS_SET,
+                                   base::Bind(&StreamSourceSetCallback,
+                                              base::Unretained(source_.get())));
+    }
   }
 
   request_->NotifyResponseStarted();
 
   // |this| may be destroyed at this point.
 }
 
 void URLRequestJob::ConvertResultToError(int result, Error* error, int* count) {
   if (result >= 0) {
     *error = OK;
     *count = result;
   } else {
     *error = static_cast<Error>(result);
     *count = 0;
   }
 }
 
 void URLRequestJob::ReadRawDataComplete(int result) {
   DCHECK(request_->status().is_io_pending());
+  DCHECK_NE(ERR_IO_PENDING, result);
 
   // TODO(cbentzel): Remove ScopedTracker below once crbug.com/475755 is fixed.
   tracked_objects::ScopedTracker tracking_profile(
       FROM_HERE_WITH_EXPLICIT_FUNCTION(
           "475755 URLRequestJob::RawReadCompleted"));
 
   // TODO(darin): Bug 1004233. Re-enable this test once all of the chrome
   // unit_tests have been fixed to not trip this.
 #if 0
   DCHECK(!request_->status().is_io_pending());
 #endif
   // The headers should be complete before reads complete
   DCHECK(has_handled_response_);
 
-  Error error;
-  int bytes_read;
-  ConvertResultToError(result, &error, &bytes_read);
+  GatherRawReadStats(result);
 
-  DCHECK_NE(ERR_IO_PENDING, error);
+  // Notify StreamSource.
+  DCHECK(!read_raw_callback_.is_null());
 
-  GatherRawReadStats(error, bytes_read);
-
-  if (filter_.get() && error == OK) {
-    // |bytes_read| being 0 indicates an EOF was received.  ReadFilteredData
-    // can incorrectly return ERR_IO_PENDING when 0 bytes are passed to it, so
-    // just don't call into the filter in that case.
-    int filter_bytes_read = 0;
-    if (bytes_read > 0) {
-      // Tell the filter that it has more data.
-      PushInputToFilter(bytes_read);
-
-      // Filter the data.
-      error = ReadFilteredData(&filter_bytes_read);
-    }
-
-    if (error == OK && !filter_bytes_read)
-      DoneReading();
-
-    DVLOG(1) << __func__ << "() \"" << request_->url().spec() << "\""
-             << " pre bytes read = " << bytes_read
-             << " pre total = " << prefilter_bytes_read_
-             << " post total = " << postfilter_bytes_read_;
-    bytes_read = filter_bytes_read;
-  } else {
-    DVLOG(1) << __func__ << "() \"" << request_->url().spec() << "\""
-             << " pre bytes read = " << bytes_read
-             << " pre total = " << prefilter_bytes_read_
-             << " post total = " << postfilter_bytes_read_;
-  }
-
-  // Synchronize the URLRequest state machine with the URLRequestJob state
-  // machine. If this read succeeded, either the request is at EOF and the
-  // URLRequest state machine goes to 'finished', or it is not and the
-  // URLRequest state machine goes to 'success'. If the read failed, the
-  // URLRequest state machine goes directly to 'finished'.  If filtered data is
-  // pending, then there's nothing to do, since the status of the request is
-  // already pending.
-  //
-  // Update the URLRequest's status first, so that NotifyReadCompleted has an
-  // accurate view of the request.
-  if (error == OK && bytes_read > 0) {
-    SetStatus(URLRequestStatus());
-  } else if (error != ERR_IO_PENDING) {
-    NotifyDone(URLRequestStatus::FromError(error));
-  }
-
-  // NotifyReadCompleted should be called after SetStatus or NotifyDone updates
-  // the status.
-  if (error == OK)
-    request_->NotifyReadCompleted(bytes_read);
-
+  base::ResetAndReturn(&read_raw_callback_).Run(result);
   // |this| may be destroyed at this point.
 }
 
 void URLRequestJob::NotifyStartError(const URLRequestStatus &status) {
   DCHECK(!has_handled_response_);
   DCHECK(request_->status().is_io_pending());
 
   has_handled_response_ = true;
   // There may be relevant information in the response info even in the
   // error case.
@@ skipping 29 matching lines @@
     request_->set_status(status);
   }
 
   // If the request succeeded (And wasn't cancelled) and the response code was
   // 4xx or 5xx, record whether or not the main frame was blank.  This is
   // intended to be a short-lived histogram, used to figure out how important
   // fixing http://crbug.com/331745 is.
   if (request_->status().is_success()) {
     int response_code = GetResponseCode();
     if (400 <= response_code && response_code <= 599) {
-      bool page_has_content = (postfilter_bytes_read_ != 0);
+      bool page_has_content = (postfilter_bytes_read() != 0);
       if (request_->load_flags() & net::LOAD_MAIN_FRAME) {
         UMA_HISTOGRAM_BOOLEAN("Net.ErrorResponseHasContentMainFrame",
                               page_has_content);
       } else {
         UMA_HISTOGRAM_BOOLEAN("Net.ErrorResponseHasContentNonMainFrame",
                               page_has_content);
       }
     }
   }
 
@@ skipping 45 matching lines @@
   return 0;
 }
 
 void URLRequestJob::DoneReading() {
   // Do nothing.
 }
 
 void URLRequestJob::DoneReadingRedirectResponse() {
 }
 
-void URLRequestJob::PushInputToFilter(int bytes_read) {
-  DCHECK(filter_);
-  filter_->FlushStreamBuffer(bytes_read);
-}
-
-Error URLRequestJob::ReadFilteredData(int* bytes_read) {
-  DCHECK(filter_);
-  DCHECK(filtered_read_buffer_.get());
-  DCHECK_GT(filtered_read_buffer_len_, 0);
-  DCHECK_LT(filtered_read_buffer_len_, 1000000);  // Sanity check.
-  DCHECK(!raw_read_buffer_);
-
-  *bytes_read = 0;
-  Error error = ERR_FAILED;
-
-  for (;;) {
-    if (is_done())
-      return OK;
-
-    if (!filter_needs_more_output_space_ && !filter_->stream_data_len()) {
-      // We don't have any raw data to work with, so read from the transaction.
-      int filtered_data_read;
-      error = ReadRawDataForFilter(&filtered_data_read);
-      // If ReadRawDataForFilter returned some data, fall through to the case
-      // below; otherwise, return early.
-      if (error != OK || filtered_data_read == 0)
-        return error;
-      filter_->FlushStreamBuffer(filtered_data_read);
-    }
-
-    if ((filter_->stream_data_len() || filter_needs_more_output_space_) &&
-        !is_done()) {
-      // Get filtered data.
-      int filtered_data_len = filtered_read_buffer_len_;
-      int output_buffer_size = filtered_data_len;
-      Filter::FilterStatus status =
-          filter_->ReadData(filtered_read_buffer_->data(), &filtered_data_len);
-
-      if (filter_needs_more_output_space_ && !filtered_data_len) {
-        // filter_needs_more_output_space_ was mistaken... there are no more
-        // bytes and we should have at least tried to fill up the filter's input
-        // buffer. Correct the state, and try again.
-        filter_needs_more_output_space_ = false;
-        continue;
-      }
-      filter_needs_more_output_space_ =
-          (filtered_data_len == output_buffer_size);
-
-      switch (status) {
-        case Filter::FILTER_DONE: {
-          filter_needs_more_output_space_ = false;
-          *bytes_read = filtered_data_len;
-          postfilter_bytes_read_ += filtered_data_len;
-          error = OK;
-          break;
-        }
-        case Filter::FILTER_NEED_MORE_DATA: {
-          // We have finished filtering all data currently in the buffer.
-          // There might be some space left in the output buffer. One can
-          // consider reading more data from the stream to feed the filter
-          // and filling up the output buffer. This leads to more complicated
-          // buffer management and data notification mechanisms.
-          // We can revisit this issue if there is a real perf need.
-          if (filtered_data_len > 0) {
-            *bytes_read = filtered_data_len;
-            postfilter_bytes_read_ += filtered_data_len;
-            error = OK;
-          } else {
-            // Read again since we haven't received enough data yet (e.g., we
-            // may not have a complete gzip header yet).
-            continue;
-          }
-          break;
-        }
-        case Filter::FILTER_OK: {
-          *bytes_read = filtered_data_len;
-          postfilter_bytes_read_ += filtered_data_len;
-          error = OK;
-          break;
-        }
-        case Filter::FILTER_ERROR: {
-          DVLOG(1) << __func__ << "() \"" << request_->url().spec() << "\""
-                   << " Filter Error";
-          filter_needs_more_output_space_ = false;
-          error = ERR_CONTENT_DECODING_FAILED;
-          UMA_HISTOGRAM_ENUMERATION("Net.ContentDecodingFailed.FilterType",
-                                    filter_->type(), Filter::FILTER_TYPE_MAX);
-          break;
-        }
-        default: {
-          NOTREACHED();
-          filter_needs_more_output_space_ = false;
-          error = ERR_FAILED;
-          break;
-        }
-      }
-
-      // If logging all bytes is enabled, log the filtered bytes read.
-      if (error == OK && filtered_data_len > 0 &&
-          request()->net_log().IsCapturing()) {
-        request()->net_log().AddByteTransferEvent(
-            NetLog::TYPE_URL_REQUEST_JOB_FILTERED_BYTES_READ, filtered_data_len,
-            filtered_read_buffer_->data());
-      }
-    } else {
-      // we are done, or there is no data left.
-      error = OK;
-    }
-    break;
-  }
-
-  if (error == OK) {
-    // When we successfully finished a read, we no longer need to save the
-    // caller's buffers. Release our reference.
-    filtered_read_buffer_ = NULL;
-    filtered_read_buffer_len_ = 0;
-  }
-  return error;
-}
-
-void URLRequestJob::DestroyFilters() {
-  filter_.reset();
+std::unique_ptr<StreamSource> URLRequestJob::SetupSource() {
+  return base::MakeUnique<URLRequestJobStreamSource>(this);
 }
 
 const URLRequestStatus URLRequestJob::GetStatus() {
   return request_->status();
 }
 
 void URLRequestJob::SetStatus(const URLRequestStatus &status) {
   // An error status should never be replaced by a non-error status by a
   // URLRequestJob.  URLRequest has some retry paths, but it resets the status
   // itself, if needed.
   DCHECK(request_->status().is_io_pending() ||
          request_->status().is_success() ||
          (!status.is_success() && !status.is_io_pending()));
   request_->set_status(status);
 }
 
 void URLRequestJob::SetProxyServer(const HostPortPair& proxy_server) {
   request_->proxy_server_ = proxy_server;
 }
 
-Error URLRequestJob::ReadRawDataForFilter(int* bytes_read) {
-  Error error = ERR_FAILED;
-  DCHECK(bytes_read);
-  DCHECK(filter_.get());
-
-  *bytes_read = 0;
-
-  // Get more pre-filtered data if needed.
-  // TODO(mbelshe): is it possible that the filter needs *MORE* data
-  //    when there is some data already in the buffer?
-  if (!filter_->stream_data_len() && !is_done()) {
-    IOBuffer* stream_buffer = filter_->stream_buffer();
-    int stream_buffer_size = filter_->stream_buffer_size();
-    error = ReadRawDataHelper(stream_buffer, stream_buffer_size, bytes_read);
-  }
-  return error;
+int64_t URLRequestJob::postfilter_bytes_read() const {
+  return base::checked_cast<int64_t>(postfilter_bytes_read_);
 }
 
-Error URLRequestJob::ReadRawDataHelper(IOBuffer* buf,
-                                       int buf_size,
-                                       int* bytes_read) {
+int URLRequestJob::ReadRawDataHelper(IOBuffer* buf,
+                                     int buf_size,
+                                     const CompletionCallback& callback) {
   DCHECK(!raw_read_buffer_);
 
-  // Keep a pointer to the read buffer, so we have access to it in
-  // GatherRawReadStats() in the event that the read completes asynchronously.
+  // Keep a pointer to the read buffer, so URLRequestJob::GatherRawReadStats()
+  // has access to it to log stats.
   raw_read_buffer_ = buf;
-  Error error;
-  ConvertResultToError(ReadRawData(buf, buf_size), &error, bytes_read);
+  int result = ReadRawData(buf, buf_size);
 
-  if (error != ERR_IO_PENDING) {
+  if (result != ERR_IO_PENDING) {
     // If the read completes synchronously, either success or failure, invoke
     // GatherRawReadStats so we can account for the completed read.
-    GatherRawReadStats(error, *bytes_read);
+    GatherRawReadStats(result);
+  } else {
+    read_raw_callback_ = callback;

[mmenke, 2016/07/28 18:40:13]

Should we use a callback, or just keep a pointer to the
URLRequestJobStreamSource?  Can't use ResetAndReturn or DCHECKs with the latter
(Speaking of those DCHECKs...this method could use one that it's not set), but
it makes the code clearer.

[xunjieli, 2016/08/01 16:46:23]

Acknowledged. Talked offline and we decided to leave this as it is. This is
because URLRequestJob indirectly owns URLRequestJobSteamSource, the callback
here is the callback passed in by the SourceStream above
URLRequestJobStreamSource. 

   }
-  return error;
+  return result;
 }
 
 void URLRequestJob::FollowRedirect(const RedirectInfo& redirect_info) {
   int rv = request_->Redirect(redirect_info);
   if (rv != OK)
     NotifyDone(URLRequestStatus(URLRequestStatus::FAILED, rv));
 }
 
-void URLRequestJob::GatherRawReadStats(Error error, int bytes_read) {
+void URLRequestJob::GatherRawReadStats(int bytes_read) {
   DCHECK(raw_read_buffer_ || bytes_read == 0);
-  DCHECK_NE(ERR_IO_PENDING, error);
-
-  if (error != OK) {
-    raw_read_buffer_ = nullptr;
-    return;
-  }
-  // If |filter_| is non-NULL, bytes will be logged after it is applied
-  // instead.
-  if (!filter_.get() && bytes_read > 0 && request()->net_log().IsCapturing()) {
-    request()->net_log().AddByteTransferEvent(
-        NetLog::TYPE_URL_REQUEST_JOB_BYTES_READ, bytes_read,
-        raw_read_buffer_->data());
-  }
+  DCHECK_NE(ERR_IO_PENDING, bytes_read);
 
   if (bytes_read > 0) {
+    if (request()->net_log().IsCapturing()) {
+      request()->net_log().AddByteTransferEvent(
+          NetLog::TYPE_URL_REQUEST_JOB_BYTES_READ, bytes_read,
+          raw_read_buffer_->data());
+    }
     RecordBytesRead(bytes_read);
   }
   raw_read_buffer_ = nullptr;
 }
 
 void URLRequestJob::RecordBytesRead(int bytes_read) {
   DCHECK_GT(bytes_read, 0);
-  prefilter_bytes_read_ += bytes_read;
+  prefilter_bytes_read_ += base::checked_cast<size_t>(bytes_read);
 
   // On first read, notify NetworkQualityEstimator that response headers have
   // been received.
   // TODO(tbansal): Move this to url_request_http_job.cc. This may catch
   // Service Worker jobs twice.
   // If prefilter_bytes_read_ is equal to bytes_read, it indicates this is the
   // first raw read of the response body. This is used as the signal that
   // response headers have been received.
   if (request_->context()->network_quality_estimator() &&
-      prefilter_bytes_read_ == bytes_read) {
+      prefilter_bytes_read() == bytes_read) {
     request_->context()->network_quality_estimator()->NotifyHeadersReceived(
         *request_);
   }
 
-  if (!filter_.get())
-    postfilter_bytes_read_ += bytes_read;
-  DVLOG(2) << __func__ << "() \"" << request_->url().spec() << "\""
+  DVLOG(2) << __FUNCTION__ << "() "
+           << "\"" << request_->url().spec() << "\""
            << " pre bytes read = " << bytes_read
-           << " pre total = " << prefilter_bytes_read_
-           << " post total = " << postfilter_bytes_read_;
+           << " pre total = " << prefilter_bytes_read()
+           << " post total = " << postfilter_bytes_read();
   UpdatePacketReadTimes();  // Facilitate stats recording if it is active.
 
   // Notify observers if any additional network usage has occurred. Note that
   // the number of received bytes over the network sent by this notification
   // could be vastly different from |bytes_read|, such as when a large chunk of
   // network bytes is received before multiple smaller raw reads are performed
   // on it.
   MaybeNotifyNetworkBytes();
 }
 
-bool URLRequestJob::FilterHasData() {
-    return filter_.get() && filter_->stream_data_len();
-}
-
 void URLRequestJob::UpdatePacketReadTimes() {
 }
 
 RedirectInfo URLRequestJob::ComputeRedirectInfo(const GURL& location,
                                                 int http_status_code) {
   const GURL& url = request_->url();
 
   RedirectInfo redirect_info;
 
   redirect_info.status_code = http_status_code;
@@ skipping 66 matching lines @@
   int64_t total_sent_bytes = GetTotalSentBytes();
   DCHECK_GE(total_sent_bytes, last_notified_total_sent_bytes_);
   if (total_sent_bytes > last_notified_total_sent_bytes_) {
     network_delegate_->NotifyNetworkBytesSent(
         request_, total_sent_bytes - last_notified_total_sent_bytes_);
   }
   last_notified_total_sent_bytes_ = total_sent_bytes;
 }
 
 }  // namespace net