Make Stream return STREAM_ABORTED to tell the reader it's failed.

Using this, StreamURLRequestJob aborts itself.

For now, this value will be used to deal with memory limit exceed.

data_bytes_read_ is moved using this opportunity since it's a part of temporary buffer consists of data_ and data_length_.

BUG=169957

Review URL: https://chromiumcodereview.appspot.com/22942003

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@219170 0039d316-1c4b-4281-b951-d872f2087c98

4 files changed, 43 insertions, 10 deletions

diff --git a/content/browser/streams/stream.cc b/content/browser/streams/stream.cc
index a9d43332..5d20fe6 100644
--- a/content/browser/streams/stream.cc
+++ b/content/browser/streams/stream.cc
@@ -23,10 +23,10 @@ namespace content {
 Stream::Stream(StreamRegistry* registry,
                StreamWriteObserver* write_observer,
                const GURL& url)
-    : data_bytes_read_(0),
-      can_add_data_(true),
+    : can_add_data_(true),
       url_(url),
       data_length_(0),
+      data_bytes_read_(0),
       last_total_buffered_bytes_(0),
       registry_(registry),
       read_observer_(NULL),
@@ -73,6 +73,7 @@ void Stream::Abort() {
   // is used for both input and output operation.
   writer_.reset();
   reader_.reset();
+  ClearBuffer();
   can_add_data_ = false;
   registry_->UnregisterStream(url());
 }
@@ -97,6 +98,9 @@ void Stream::AddData(scoped_refptr<net::IOBuffer> buffer, size_t size) {
 }
 
 void Stream::AddData(const char* data, size_t size) {
+  if (!writer_.get())
+    return;
+
   scoped_refptr<net::IOBuffer> io_buffer(new net::IOBuffer(size));
   memcpy(io_buffer->data(), data, size);
   AddData(io_buffer, size);
@@ -123,13 +127,12 @@ Stream::StreamState Stream::ReadRawData(net::IOBuffer* buf,
 
   *bytes_read = 0;
   if (!data_.get()) {
-    // TODO(tyoshino): Add STREAM_ABORTED type to tell the reader that this
-    // stream is aborted.
+    DCHECK(!data_length_);
+    DCHECK(!data_bytes_read_);
+
     if (!reader_.get())
-      return STREAM_EMPTY;
+      return STREAM_ABORTED;
 
-    data_length_ = 0;
-    data_bytes_read_ = 0;
     ByteStreamReader::StreamState state = reader_->Read(&data_, &data_length_);
     switch (state) {
       case ByteStreamReader::STREAM_HAS_DATA:
@@ -149,7 +152,7 @@ Stream::StreamState Stream::ReadRawData(net::IOBuffer* buf,
   memcpy(buf->data(), data_->data() + data_bytes_read_, to_read);
   data_bytes_read_ += to_read;
   if (data_bytes_read_ >= data_length_)
-    data_ = NULL;
+    ClearBuffer();
 
   *bytes_read = to_read;
   return STREAM_HAS_DATA;
@@ -186,4 +189,10 @@ void Stream::OnDataAvailable() {
     read_observer_->OnDataAvailable(this);
 }
 
+void Stream::ClearBuffer() {
+  data_ = NULL;
+  data_length_ = 0;
+  data_bytes_read_ = 0;
+}
+
 }  // namespace content
diff --git a/content/browser/streams/stream.h b/content/browser/streams/stream.h
index 1878adf..a0599de 100644
--- a/content/browser/streams/stream.h
+++ b/content/browser/streams/stream.h
@@ -36,6 +36,7 @@ class CONTENT_EXPORT Stream : public base::RefCountedThreadSafe<Stream> {
     STREAM_HAS_DATA,
     STREAM_COMPLETE,
     STREAM_EMPTY,
+    STREAM_ABORTED,
   };
 
   // Creates a stream.
@@ -86,6 +87,10 @@ class CONTENT_EXPORT Stream : public base::RefCountedThreadSafe<Stream> {
     return last_total_buffered_bytes_;
   }
 
+ protected:
+  // Stops accepting new data and make ReadRawData() return STREAM_ABORTED.
+  void Abort();
+
  private:
   friend class base::RefCountedThreadSafe<Stream>;
 
@@ -94,15 +99,21 @@ class CONTENT_EXPORT Stream : public base::RefCountedThreadSafe<Stream> {
   void OnSpaceAvailable();
   void OnDataAvailable();
 
-  void Abort();
+  // Clears |data_| and related variables.
+  void ClearBuffer();
 
-  size_t data_bytes_read_;
   bool can_add_data_;
 
   GURL url_;
 
+  // Buffer for storing data read from |reader_| but not yet read out from this
+  // Stream by ReadRawData() method.
   scoped_refptr<net::IOBuffer> data_;
+  // Number of bytes read from |reader_| into |data_| including bytes already
+  // read out.
   size_t data_length_;
+  // Number of bytes in |data_| that are already read out.
+  size_t data_bytes_read_;
 
   // Last value returned by writer_->TotalBufferedBytes() in AddData(). Stored
   // in order to check memory usage.
diff --git a/content/browser/streams/stream_unittest.cc b/content/browser/streams/stream_unittest.cc
index 2e0386b..a2d9593 100644
--- a/content/browser/streams/stream_unittest.cc
+++ b/content/browser/streams/stream_unittest.cc
@@ -65,6 +65,9 @@ class TestStreamReader : public StreamReadObserver {
         case Stream::STREAM_EMPTY:
           EXPECT_FALSE(completed_);
           return;
+        case Stream::STREAM_ABORTED:
+          EXPECT_FALSE(completed_);
+          return;
       }
       size_t old_capacity = buffer_->capacity();
       buffer_->SetCapacity(old_capacity + bytes_read);
diff --git a/content/browser/streams/stream_url_request_job.cc b/content/browser/streams/stream_url_request_job.cc
index 0965178..09b9e6d 100644
--- a/content/browser/streams/stream_url_request_job.cc
+++ b/content/browser/streams/stream_url_request_job.cc
@@ -61,6 +61,11 @@ void StreamURLRequestJob::OnDataAvailable(Stream* stream) {
     case Stream::STREAM_EMPTY:
       NOTREACHED();
       break;
+    case Stream::STREAM_ABORTED:
+      // Handle this as connection reset.
+      NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
+                                       net::ERR_CONNECTION_RESET));
+      break;
   }
 
   // Clear the buffers before notifying the read is complete, so that it is
@@ -115,6 +120,11 @@ bool StreamURLRequestJob::ReadRawData(net::IOBuffer* buf,
       pending_buffer_size_ = to_read;
       SetStatus(net::URLRequestStatus(net::URLRequestStatus::IO_PENDING, 0));
       return false;
+    case Stream::STREAM_ABORTED:
+      // Handle this as connection reset.
+      NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
+                                       net::ERR_CONNECTION_RESET));
+      return false;
   }
   NOTREACHED();
   return false;