[Fetch] Support various operations after reading data partially.

Source/modules/fetch/Body.cpp

 // Copyright 2014 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 "config.h"
 #include "modules/fetch/Body.h"
 
 #include "bindings/core/v8/ExceptionState.h"
 #include "bindings/core/v8/ScriptPromiseResolver.h"
 #include "bindings/core/v8/ScriptState.h"
@@ skipping 41 matching lines @@
     USING_GARBAGE_COLLECTED_MIXIN(ReadableStreamSource);
 public:
     enum State {
         Initial,
         Streaming,
         ReadingBlob,
         Closed,
         Errored,
         BodyUsed,
     };
-    ReadableStreamSource(Body* body) : m_body(body), m_state(Initial), m_queueCount(0)
+    ReadableStreamSource(Body* body) : m_body(body), m_state(Initial)
     {
         if (m_body->buffer()) {
             m_bodyStreamBuffer = m_body->buffer();
         } else {
             m_blobDataHandle = m_body->blobDataHandle();
             if (!m_blobDataHandle)
                 m_blobDataHandle = BlobDataHandle::create(BlobData::create(), 0);
         }
     }
     ~ReadableStreamSource() override { }
@@ skipping 20 matching lines @@
                 return onClose();
         } else {
             ASSERT(m_blobDataHandle);
             m_state = ReadingBlob;
             FileReaderLoader::ReadType readType = FileReaderLoader::ReadAsArrayBuffer;
             m_loader = adoptPtr(new FileReaderLoader(readType, this));
             m_loader->start(m_body->executionContext(), m_blobDataHandle);
         }
     }
     // Creates a new BodyStreamBuffer to drain the data.
-    BodyStreamBuffer* createDrainingStream(bool* dataLost)
+    BodyStreamBuffer* createDrainingStream()
     {
         ASSERT(!m_drainingStreamBuffer);
         ASSERT(m_state != Initial);
         ASSERT(m_state != BodyUsed);
         ASSERT(m_stream);
-        ASSERT(dataLost);
         m_drainingStreamBuffer = new BodyStreamBuffer();
         if (m_state == Errored) {
             m_drainingStreamBuffer->error(exception());
             return m_drainingStreamBuffer;
         }
         // Take back the data in |m_stream|.
         Deque<std::pair<RefPtr<DOMArrayBuffer>, size_t>> tmp_queue;
         if (m_stream->stateInternal() == ReadableStream::Readable)
             m_stream->readInternal(tmp_queue);
-        *dataLost = m_queueCount != tmp_queue.size();
         while (!tmp_queue.isEmpty()) {
             std::pair<RefPtr<DOMArrayBuffer>, size_t> data = tmp_queue.takeFirst();
             m_drainingStreamBuffer->write(data.first);
         }
         if (m_state == Closed)
             m_drainingStreamBuffer->close();
         else
             m_stream->close();
         return m_drainingStreamBuffer;
     }
@@ skipping 46 matching lines @@
         close();
     }
     void didFail(FileError::ErrorCode) override
     {
         ASSERT(m_state == ReadingBlob);
         error();
     }
 
     void write(PassRefPtr<DOMArrayBuffer> buf)
     {
         if (m_drainingStreamBuffer) {

[horo, 2015/03/20 02:28:54]

nit: we can remove brackets.

[yhirano, 2015/03/20 03:24:49]

Done.

             m_drainingStreamBuffer->write(buf);
         } else {
-            ++m_queueCount;
             m_stream->enqueue(buf);
         }
     }
     void close()
     {
         m_state = Closed;
         if (m_drainingStreamBuffer)
             m_drainingStreamBuffer->close();
         else
             m_stream->close();
@@ skipping 21 matching lines @@
     // Set when the data container of the Body is a BodyStreamBuffer.
     Member<BodyStreamBuffer> m_bodyStreamBuffer;
     // Set when the data container of the Body is a BlobDataHandle.
     RefPtr<BlobDataHandle> m_blobDataHandle;
     // Used to read the data from BlobDataHandle.
     OwnPtr<FileReaderLoader> m_loader;
     // Created when createDrainingStream is called to drain the data.
     Member<BodyStreamBuffer> m_drainingStreamBuffer;
     Member<ReadableStreamImpl<ReadableStreamChunkTypeTraits<DOMArrayBuffer>>> m_stream;
     State m_state;
-    // The count of the chunks which were enqueued to the ReadableStream.
-    size_t m_queueCount;
 };
 
 ScriptPromise Body::readAsync(ScriptState* scriptState, ResponseType type)
 {
     if (bodyUsed())
         return ScriptPromise::reject(scriptState, V8ThrowException::createTypeError(scriptState->isolate(), "Already read"));
 
     // When the main thread sends a V8::TerminateExecution() signal to a worker
     // thread, any V8 API on the worker thread starts returning an empty
     // handle. This can happen in Body::readAsync. To avoid the situation, we
     // first check the ExecutionContext and return immediately if it's already
     // gone (which means that the V8::TerminateExecution() signal has been sent
     // to this worker thread).
     ExecutionContext* executionContext = scriptState->executionContext();
     if (!executionContext)
         return ScriptPromise();
 
     setBodyUsed();
     m_responseType = type;
 
     ASSERT(!m_resolver);
     m_resolver = ScriptPromiseResolver::create(scriptState);
     ScriptPromise promise = m_resolver->promise();
 
     if (m_stream) {
         ASSERT(m_streamSource);
-        bool dataLost;
-        m_streamSource->createDrainingStream(&dataLost)->readAllAndCreateBlobHandle(contentTypeForBuffer(), new BlobHandleReceiver(this));
+        m_streamSource->createDrainingStream()->readAllAndCreateBlobHandle(contentTypeForBuffer(), new BlobHandleReceiver(this));
     } else if (buffer()) {
         buffer()->readAllAndCreateBlobHandle(contentTypeForBuffer(), new BlobHandleReceiver(this));
     } else {
         readAsyncFromBlob(blobDataHandle());
     }
     return promise;
 }
 
 void Body::readAsyncFromBlob(PassRefPtr<BlobDataHandle> handle)
 {
@@ skipping 91 matching lines @@
         ASSERT(!exceptionState.hadException());
     }
     m_bodyUsed = true;
 }
 
 bool Body::streamAccessed() const
 {
     return m_stream;
 }
 
-BodyStreamBuffer* Body::createDrainingStream(bool* dataLost)
+BodyStreamBuffer* Body::createDrainingStream()
 {
     ASSERT(m_stream);
-    BodyStreamBuffer* newBuffer = m_streamSource->createDrainingStream(dataLost);
+    BodyStreamBuffer* newBuffer = m_streamSource->createDrainingStream();
     m_stream = nullptr;
     m_streamSource = nullptr;
     return newBuffer;
 }
 
 void Body::stop()
 {
     // Canceling the load will call didFail which will remove the resolver.
     if (m_loader)
         m_loader->cancel();
@@ skipping 101 matching lines @@
 
 void Body::didBlobHandleReceiveError(PassRefPtrWillBeRawPtr<DOMException> exception)
 {
     if (!m_resolver)
         return;
     m_resolver->reject(exception);
     m_resolver.clear();
 }
 
 } // namespace blink