Use BytesConsumer in BodyStreamBuffer

third_party/WebKit/Source/modules/fetch/BodyStreamBuffer.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 "modules/fetch/BodyStreamBuffer.h"
 
 #include "bindings/core/v8/ScriptState.h"
 #include "bindings/core/v8/V8HiddenValue.h"
 #include "bindings/core/v8/WorkerOrWorkletScriptController.h"
 #include "core/dom/DOMArrayBuffer.h"
 #include "core/dom/DOMTypedArray.h"
 #include "core/dom/ExceptionCode.h"
 #include "core/streams/ReadableStreamController.h"
 #include "core/streams/ReadableStreamOperations.h"
 #include "core/workers/WorkerGlobalScope.h"
 #include "modules/fetch/Body.h"
 #include "modules/fetch/BytesConsumerForDataConsumerHandle.h"
 #include "modules/fetch/DataConsumerHandleUtil.h"
-#include "modules/fetch/DataConsumerTee.h"
 #include "modules/fetch/ReadableStreamDataConsumerHandle.h"
 #include "platform/blob/BlobData.h"
 #include "platform/network/EncodedFormData.h"
 #include <memory>
 
 namespace blink {
 
 namespace {
 
 bool isTerminating(ScriptState* scriptState)
@@ skipping 62 matching lines @@
     void stop() override
     {
         m_buffer->stopLoading();
     }
 
     Member<BodyStreamBuffer> m_buffer;
     Member<FetchDataLoader::Client> m_client;
 };
 
 BodyStreamBuffer::BodyStreamBuffer(ScriptState* scriptState, std::unique_ptr<FetchDataConsumerHandle> handle)
+    : BodyStreamBuffer(scriptState, new BytesConsumerForDataConsumerHandle(std::move(handle)))
+{
+}
+
+BodyStreamBuffer::BodyStreamBuffer(ScriptState* scriptState, BytesConsumer* consumer)
     : UnderlyingSourceBase(scriptState)
     , m_scriptState(scriptState)
-    , m_handle(std::move(handle))
-    , m_reader(m_handle->obtainFetchDataReader(this))
+    , m_consumer(consumer)
     , m_madeFromReadableStream(false)
 {
     if (isTerminating(scriptState)) {
-        m_reader = nullptr;
-        m_handle = nullptr;
+        m_consumer->cancel();
         return;
     }
     v8::Local<v8::Value> bodyValue = toV8(this, scriptState);
     if (bodyValue.IsEmpty()) {
         DCHECK(isTerminating(scriptState));
-        m_reader = nullptr;
-        m_handle = nullptr;
+        m_consumer->cancel();
         return;
     }
     DCHECK(bodyValue->IsObject());
     v8::Local<v8::Object> body = bodyValue.As<v8::Object>();
 
     ScriptValue readableStream = ReadableStreamOperations::createReadableStream(
         scriptState, this, ReadableStreamOperations::createCountQueuingStrategy(scriptState, 0));
     if (isTerminating(scriptState)) {
-        m_reader = nullptr;
-        m_handle = nullptr;
+        m_consumer->cancel();
         return;
     }
     V8HiddenValue::setHiddenValue(scriptState, body, V8HiddenValue::internalBodyStream(scriptState->isolate()), readableStream.v8Value());
+    m_consumer->setClient(this);
 }
 
 BodyStreamBuffer::BodyStreamBuffer(ScriptState* scriptState, ScriptValue stream)
     : UnderlyingSourceBase(scriptState)
     , m_scriptState(scriptState)
     , m_madeFromReadableStream(true)
 {
     DCHECK(ReadableStreamOperations::isReadableStream(scriptState, stream));
     if (isTerminating(scriptState))
         return;
@@ skipping 16 matching lines @@
     v8::Local<v8::Value> bodyValue = toV8(this, m_scriptState.get());
     if (bodyValue.IsEmpty()) {
         DCHECK(isTerminating(m_scriptState.get()));
         return ScriptValue();
     }
     DCHECK(bodyValue->IsObject());
     v8::Local<v8::Object> body = bodyValue.As<v8::Object>();
     return ScriptValue(m_scriptState.get(), V8HiddenValue::getHiddenValue(m_scriptState.get(), body, V8HiddenValue::internalBodyStream(m_scriptState->isolate())));
 }
 
-PassRefPtr<BlobDataHandle> BodyStreamBuffer::drainAsBlobDataHandle(FetchDataConsumerHandle::Reader::BlobSizePolicy policy)
+PassRefPtr<BlobDataHandle> BodyStreamBuffer::drainAsBlobDataHandle(BytesConsumer::BlobSizePolicy policy)
 {
     ASSERT(!isStreamLocked());
     ASSERT(!isStreamDisturbed());
     if (isStreamClosed() || isStreamErrored())
         return nullptr;
 
     if (m_madeFromReadableStream)
         return nullptr;
 
-    RefPtr<BlobDataHandle> blobDataHandle = m_reader->drainAsBlobDataHandle(policy);
+    RefPtr<BlobDataHandle> blobDataHandle = m_consumer->drainAsBlobDataHandle(policy);
     if (blobDataHandle) {
         closeAndLockAndDisturb();
         return blobDataHandle.release();
     }
     return nullptr;
 }
 
 PassRefPtr<EncodedFormData> BodyStreamBuffer::drainAsFormData()
 {
     ASSERT(!isStreamLocked());
     ASSERT(!isStreamDisturbed());
     if (isStreamClosed() || isStreamErrored())
         return nullptr;
 
     if (m_madeFromReadableStream)
         return nullptr;
 
-    RefPtr<EncodedFormData> formData = m_reader->drainAsFormData();
+    RefPtr<EncodedFormData> formData = m_consumer->drainAsFormData();
     if (formData) {
         closeAndLockAndDisturb();
         return formData.release();
     }
     return nullptr;
 }
 
 void BodyStreamBuffer::startLoading(FetchDataLoader* loader, FetchDataLoader::Client* client)
 {
     ASSERT(!m_loader);
     ASSERT(m_scriptState->contextIsValid());
-    std::unique_ptr<FetchDataConsumerHandle> handle = releaseHandle();
     m_loader = loader;
-    loader->start(new BytesConsumerForDataConsumerHandle(std::move(handle)), new LoaderClient(m_scriptState->getExecutionContext(), this, client));
+    loader->start(releaseHandle(), new LoaderClient(m_scriptState->getExecutionContext(), this, client));
 }
 
 void BodyStreamBuffer::tee(BodyStreamBuffer** branch1, BodyStreamBuffer** branch2)
 {
     DCHECK(!isStreamLocked());
     DCHECK(!isStreamDisturbed());
     *branch1 = nullptr;
     *branch2 = nullptr;
 
     if (m_madeFromReadableStream) {
         ScriptValue stream1, stream2;
         ReadableStreamOperations::tee(m_scriptState.get(), stream(), &stream1, &stream2);
         *branch1 = new BodyStreamBuffer(m_scriptState.get(), stream1);
         *branch2 = new BodyStreamBuffer(m_scriptState.get(), stream2);
         return;
     }
-    std::unique_ptr<FetchDataConsumerHandle> handle = releaseHandle();
-    std::unique_ptr<FetchDataConsumerHandle> handle1, handle2;
-    DataConsumerTee::create(m_scriptState->getExecutionContext(), std::move(handle), &handle1, &handle2);
-    *branch1 = new BodyStreamBuffer(m_scriptState.get(), std::move(handle1));
-    *branch2 = new BodyStreamBuffer(m_scriptState.get(), std::move(handle2));
+    BytesConsumer* dest1 = nullptr;
+    BytesConsumer* dest2 = nullptr;
+    BytesConsumer::tee(m_scriptState->getExecutionContext(), releaseHandle(), &dest1, &dest2);
+    *branch1 = new BodyStreamBuffer(m_scriptState.get(), dest1);
+    *branch2 = new BodyStreamBuffer(m_scriptState.get(), dest2);
 }
 
 ScriptPromise BodyStreamBuffer::pull(ScriptState* scriptState)
 {
     ASSERT(scriptState == m_scriptState.get());
     if (m_streamNeedsMore)
         return ScriptPromise::castUndefined(scriptState);
     m_streamNeedsMore = true;
     processData();
     return ScriptPromise::castUndefined(scriptState);
 }
 
 ScriptPromise BodyStreamBuffer::cancel(ScriptState* scriptState, ScriptValue reason)
 {
     ASSERT(scriptState == m_scriptState.get());
     close();
     return ScriptPromise::castUndefined(scriptState);
 }
 
-void BodyStreamBuffer::didGetReadable()
+void BodyStreamBuffer::onStateChange()
 {
-    if (!m_reader || !getExecutionContext() || getExecutionContext()->activeDOMObjectsAreStopped())
+    if (!m_consumer || !getExecutionContext() || getExecutionContext()->activeDOMObjectsAreStopped())
         return;
 
-    if (!m_streamNeedsMore) {
-        // Perform zero-length read to call close()/error() early.
-        size_t readSize;
-        WebDataConsumerHandle::Result result = m_reader->read(nullptr, 0, WebDataConsumerHandle::FlagNone, &readSize);
-        switch (result) {
-        case WebDataConsumerHandle::Ok:
-        case WebDataConsumerHandle::ShouldWait:
-            return;
-        case WebDataConsumerHandle::Done:
-            close();
-            return;
-        case WebDataConsumerHandle::Busy:
-        case WebDataConsumerHandle::ResourceExhausted:
-        case WebDataConsumerHandle::UnexpectedError:
-            error();
-            return;
-        }
+    switch (m_consumer->getPublicState()) {
+    case BytesConsumer::PublicState::ReadableOrWaiting:
+        break;
+    case BytesConsumer::PublicState::Closed:
+        close();
+        return;
+    case BytesConsumer::PublicState::Errored:
+        error();
         return;
     }
     processData();
 }
 
 bool BodyStreamBuffer::hasPendingActivity() const
 {
     if (m_loader)
         return true;
     return UnderlyingSourceBase::hasPendingActivity();
 }
 
 void BodyStreamBuffer::stop()
 {
-    m_reader = nullptr;
-    m_handle = nullptr;
+    if (m_consumer)
+        m_consumer->cancel();
+    m_consumer = nullptr;
     UnderlyingSourceBase::stop();
 }
 
 bool BodyStreamBuffer::isStreamReadable()
 {
     ScriptState::Scope scope(m_scriptState.get());
     return ReadableStreamOperations::isReadable(m_scriptState.get(), stream());
 }
 
 bool BodyStreamBuffer::isStreamClosed()
@@ skipping 30 matching lines @@
 
     ScriptState::Scope scope(m_scriptState.get());
     NonThrowableExceptionState exceptionState;
     ScriptValue reader = ReadableStreamOperations::getReader(m_scriptState.get(), stream(), exceptionState);
     ReadableStreamOperations::defaultReaderRead(m_scriptState.get(), reader);
 }
 
 void BodyStreamBuffer::close()
 {
     controller()->close();
-    m_reader = nullptr;
-    m_handle = nullptr;
+    if (m_consumer) {
+        m_consumer->cancel();
+        m_consumer = nullptr;
+    }
 }
 
 void BodyStreamBuffer::error()
 {
     controller()->error(DOMException::create(NetworkError, "network error"));
-    m_reader = nullptr;
-    m_handle = nullptr;
+    if (m_consumer) {
+        m_consumer->cancel();
+        m_consumer = nullptr;
+    }
 }
 
 void BodyStreamBuffer::processData()
 {
-    ASSERT(m_reader);
+    DCHECK(m_consumer);
     while (m_streamNeedsMore) {
-        const void* buffer;
-        size_t available;
-        WebDataConsumerHandle::Result result = m_reader->beginRead(&buffer, WebDataConsumerHandle::FlagNone, &available);
-        switch (result) {
-        case WebDataConsumerHandle::Ok: {
-            DOMUint8Array* array = DOMUint8Array::create(static_cast<const unsigned char*>(buffer), available);
-            m_reader->endRead(available);
+        const char* buffer = nullptr;
+        size_t available = 0;
+
+        switch (m_consumer->beginRead(&buffer, &available)) {
+        case BytesConsumer::Result::Ok: {
+            DOMUint8Array* array = DOMUint8Array::create(reinterpret_cast<const unsigned char*>(buffer), available);
+            m_consumer->endRead(available);

[hiroshige, 2016/09/07 09:25:54]

We have to check the return value of endRead().

[yhirano, 2016/09/08 01:41:05]

Done.

             // Clear m_streamNeedsMore in order to detect a pull call.
             m_streamNeedsMore = false;
             controller()->enqueue(array);
             // If m_streamNeedsMore is true, it means that pull is called and
             // the stream needs more data even if the desired size is not
             // positive.
             if (!m_streamNeedsMore)
                 m_streamNeedsMore = controller()->desiredSize() > 0;
             break;
         }
-        case WebDataConsumerHandle::Done:
+        case BytesConsumer::Result::ShouldWait:
+            return;
+        case BytesConsumer::Result::Done:
             close();
             return;
-
-        case WebDataConsumerHandle::ShouldWait:
-            return;
-
-        case WebDataConsumerHandle::Busy:
-        case WebDataConsumerHandle::ResourceExhausted:
-        case WebDataConsumerHandle::UnexpectedError:
+        case BytesConsumer::Result::Error:
             error();
             return;
         }
     }
 }
 
 void BodyStreamBuffer::endLoading()
 {
     ASSERT(m_loader);
     m_loader = nullptr;
 }
 
 void BodyStreamBuffer::stopLoading()
 {
     if (!m_loader)
         return;
     m_loader->cancel();
     m_loader = nullptr;
 }
 
-std::unique_ptr<FetchDataConsumerHandle> BodyStreamBuffer::releaseHandle()
+BytesConsumer* BodyStreamBuffer::releaseHandle()
 {
     DCHECK(!isStreamLocked());
     DCHECK(!isStreamDisturbed());
 
     if (m_madeFromReadableStream) {
         ScriptState::Scope scope(m_scriptState.get());
         // We need to have |reader| alive by some means (as written in
         // ReadableStreamDataConsumerHandle). Based on the following facts
         //  - This function is used only from tee and startLoading.
         //  - This branch cannot be taken when called from tee.
         //  - startLoading makes hasPendingActivity return true while loading.
         // , we don't need to keep the reader explicitly.
         NonThrowableExceptionState exceptionState;
         ScriptValue reader = ReadableStreamOperations::getReader(m_scriptState.get(), stream(), exceptionState);
-        return ReadableStreamDataConsumerHandle::create(m_scriptState.get(), reader);
+        return new BytesConsumerForDataConsumerHandle(ReadableStreamDataConsumerHandle::create(m_scriptState.get(), reader));
     }
     // We need to call these before calling closeAndLockAndDisturb.
     const bool isClosed = isStreamClosed();
     const bool isErrored = isStreamErrored();
-    std::unique_ptr<FetchDataConsumerHandle> handle = std::move(m_handle);
+    BytesConsumer* consumer = m_consumer.release();
 
     closeAndLockAndDisturb();
 
     if (isClosed) {
         // Note that the stream cannot be "draining", because it doesn't have
         // the internal buffer.
-        return createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle());
+        return new BytesConsumerForDataConsumerHandle(createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle()));
     }
     if (isErrored)
-        return createFetchDataConsumerHandleFromWebHandle(createUnexpectedErrorDataConsumerHandle());
+        return new BytesConsumerForDataConsumerHandle(createFetchDataConsumerHandleFromWebHandle(createUnexpectedErrorDataConsumerHandle()));
 
-    DCHECK(handle);
-    return handle;
+    DCHECK(consumer);
+    consumer->clearClient();
+    return consumer;
 }
 
 } // namespace blink