Make Response::body return v8-extra based stream behind flag

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/V8BindingMacros.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 "modules/fetch/Body.h"
 #include "modules/fetch/DataConsumerHandleUtil.h"
+#include "platform/RuntimeEnabledFeatures.h"
 #include "platform/blob/BlobData.h"
 #include "platform/network/EncodedFormData.h"
 
 namespace blink {
 
 class BodyStreamBuffer::LoaderClient final : public GarbageCollectedFinalized<LoaderClient>, public ActiveDOMObject, public FetchDataLoader::Client {
     WTF_MAKE_NONCOPYABLE(LoaderClient);
     USING_GARBAGE_COLLECTED_MIXIN(LoaderClient);
 public:
     LoaderClient(ExecutionContext* executionContext, BodyStreamBuffer* buffer, FetchDataLoader::Client* client)
@@ skipping 45 matching lines @@
 private:
     void stop() override
     {
         m_buffer->stopLoading();
     }
 
     Member<BodyStreamBuffer> m_buffer;
     Member<FetchDataLoader::Client> m_client;
 };
 
-BodyStreamBuffer::BodyStreamBuffer(PassOwnPtr<FetchDataConsumerHandle> handle)
-    : m_handle(handle)
+BodyStreamBuffer::BodyStreamBuffer(ScriptState* scriptState, PassOwnPtr<FetchDataConsumerHandle> handle)
+    : UnderlyingSourceBase(scriptState)
+    , m_scriptState(scriptState)
+    , m_handle(handle)
     , m_reader(m_handle->obtainReader(this))
-    , m_stream(new ReadableByteStream(this, new ReadableByteStream::StrictStrategy))
-    , m_streamNeedsMore(false)
 {
-    m_stream->didSourceStart();
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        ScriptState::Scope scope(scriptState);
+        v8::Local<v8::Value> bodyValue = toV8(this, scriptState);
+        ASSERT(!bodyValue.IsEmpty());
+        ASSERT(bodyValue->IsObject());
+        v8::Local<v8::Object> body = bodyValue.As<v8::Object>();
+
+        ScriptValue readableStream = ReadableStreamOperations::createReadableStream(
+            scriptState, this, ReadableStreamOperations::createCountQueuingStrategy(scriptState, 0));
+        V8HiddenValue::setHiddenValue(scriptState, body, V8HiddenValue::internalBodyStream(scriptState->isolate()), readableStream.v8Value());
+    } else {
+        m_stream = new ReadableByteStream(this, new ReadableByteStream::StrictStrategy);
+        m_stream->didSourceStart();
+    }
 }
 
-PassRefPtr<BlobDataHandle> BodyStreamBuffer::drainAsBlobDataHandle(ExecutionContext* executionContext, FetchDataConsumerHandle::Reader::BlobSizePolicy policy)
+ScriptValue BodyStreamBuffer::stream()
+{
+    ScriptState::Scope scope(m_scriptState.get());
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        v8::Local<v8::Value> bodyValue = toV8(this, m_scriptState.get());
+        ASSERT(!bodyValue.IsEmpty());
+        ASSERT(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())));
+    }
+    return ScriptValue(m_scriptState.get(), toV8(m_stream, m_scriptState.get()));
+}
+
+PassRefPtr<BlobDataHandle> BodyStreamBuffer::drainAsBlobDataHandle(FetchDataConsumerHandle::Reader::BlobSizePolicy policy)
 {
     ASSERT(!isStreamLocked());
     ASSERT(!isStreamDisturbed());
     if (isStreamClosed() || isStreamErrored())
         return nullptr;
 
     RefPtr<BlobDataHandle> blobDataHandle = m_reader->drainAsBlobDataHandle(policy);
     if (blobDataHandle) {
-        NonThrowableExceptionState exceptionState;
-        m_stream->getBytesReader(executionContext, exceptionState);
-        m_stream->setIsDisturbed();
+        lockAndDisturb();
         close();
         return blobDataHandle.release();
     }
     return nullptr;
 }
 
-PassRefPtr<EncodedFormData> BodyStreamBuffer::drainAsFormData(ExecutionContext* executionContext)
+PassRefPtr<EncodedFormData> BodyStreamBuffer::drainAsFormData()
 {
     ASSERT(!isStreamLocked());
     ASSERT(!isStreamDisturbed());
     if (isStreamClosed() || isStreamErrored())
         return nullptr;
 
     RefPtr<EncodedFormData> formData = m_reader->drainAsFormData();
     if (formData) {
-        NonThrowableExceptionState exceptionState;
-        m_stream->getBytesReader(executionContext, exceptionState);
-        m_stream->setIsDisturbed();
+        lockAndDisturb();
         close();
         return formData.release();
     }
     return nullptr;
 }
 
-PassOwnPtr<FetchDataConsumerHandle> BodyStreamBuffer::releaseHandle(ExecutionContext* executionContext)
+PassOwnPtr<FetchDataConsumerHandle> BodyStreamBuffer::releaseHandle()
 {
     ASSERT(!isStreamLocked());
     ASSERT(!isStreamDisturbed());
-    m_reader = nullptr;
-    m_stream->setIsDisturbed();
-    NonThrowableExceptionState exceptionState;
-    m_stream->getBytesReader(executionContext, exceptionState);
+    lockAndDisturb();
 
     if (isStreamClosed())
         return createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle());
     if (isStreamErrored())
         return createFetchDataConsumerHandleFromWebHandle(createUnexpectedErrorDataConsumerHandle());
 
     ASSERT(m_handle);
     OwnPtr<FetchDataConsumerHandle> handle = m_handle.release();
     close();
     return handle.release();
 }
 
-void BodyStreamBuffer::startLoading(ExecutionContext* executionContext, FetchDataLoader* loader, FetchDataLoader::Client* client)
+void BodyStreamBuffer::startLoading(FetchDataLoader* loader, FetchDataLoader::Client* client)
 {
     ASSERT(!m_loader);
-    OwnPtr<FetchDataConsumerHandle> handle = releaseHandle(executionContext);
+    OwnPtr<FetchDataConsumerHandle> handle = releaseHandle();
     m_loader = loader;
-    loader->start(handle.get(), new LoaderClient(executionContext, this, client));
+    loader->start(handle.get(), new LoaderClient(m_scriptState->getExecutionContext(), this, client));

[haraken, 2016/04/21 04:36:28]

Don't you need:

  if (!m_scriptState->contextIsValid())
    return;

?

[yhirano, 2016/04/21 08:14:22]

I'd like to assert that the context is valid here.

 }
 
 bool BodyStreamBuffer::hasPendingActivity() const
 {
-    return m_loader || (isStreamLocked() && isStreamReadable());
+    if (m_loader)
+        return true;
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled())
+        return UnderlyingSourceBase::hasPendingActivity();
+
+    return m_stream->stateInternal() == ReadableStream::Readable && m_stream->isLocked();
 }
 
 void BodyStreamBuffer::stop()
 {
     m_reader = nullptr;
     m_handle = nullptr;
 }
 
 void BodyStreamBuffer::pullSource()
 {
     ASSERT(!m_streamNeedsMore);
     m_streamNeedsMore = true;
     processData();
 }
 
 ScriptPromise BodyStreamBuffer::cancelSource(ScriptState* scriptState, ScriptValue)
 {
+    ASSERT(scriptState == m_scriptState.get());
     close();
     return ScriptPromise::castUndefined(scriptState);
 }
+
+ScriptPromise BodyStreamBuffer::pull(ScriptState* scriptState)
+{
+    ASSERT(!m_streamNeedsMore);
+    ASSERT(scriptState == m_scriptState.get());
+    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()
 {
     if (!m_reader)
         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;
         }
         return;
     }
     processData();
 }
 
-bool BodyStreamBuffer::isStreamReadable() const
+bool BodyStreamBuffer::isStreamReadable()
 {
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        ScriptState::Scope scope(m_scriptState.get());

[haraken, 2016/04/21 04:36:28]

Ditto. I guess you need to check m_scriptState->contextIsValid() before using
the ScriptState.

[yhirano, 2016/04/21 08:14:22]

I'd like to assert that the context is valid here, and that assertion is already
put in ScriptState::Scope.

+        return ReadableStreamOperations::isReadable(m_scriptState.get(), stream());
+    }
     return m_stream->stateInternal() == ReadableStream::Readable;
 }
 
-bool BodyStreamBuffer::isStreamClosed() const
+bool BodyStreamBuffer::isStreamClosed()
 {
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        ScriptState::Scope scope(m_scriptState.get());
+        return ReadableStreamOperations::isClosed(m_scriptState.get(), stream());
+    }
     return m_stream->stateInternal() == ReadableStream::Closed;
 }
 
-bool BodyStreamBuffer::isStreamErrored() const
+bool BodyStreamBuffer::isStreamErrored()
 {
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        ScriptState::Scope scope(m_scriptState.get());
+        return ReadableStreamOperations::isErrored(m_scriptState.get(), stream());
+    }
     return m_stream->stateInternal() == ReadableStream::Errored;
 }
 
-bool BodyStreamBuffer::isStreamLocked() const
+bool BodyStreamBuffer::isStreamLocked()
 {
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        ScriptState::Scope scope(m_scriptState.get());
+        return ReadableStreamOperations::isLocked(m_scriptState.get(), stream());
+    }
     return m_stream->isLocked();
 }
 
-bool BodyStreamBuffer::isStreamDisturbed() const
+bool BodyStreamBuffer::isStreamDisturbed()
 {
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        ScriptState::Scope scope(m_scriptState.get());
+        return ReadableStreamOperations::isDisturbed(m_scriptState.get(), stream());
+    }
     return m_stream->isDisturbed();
 }
 
+void BodyStreamBuffer::setDisturbed()
+{
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        ScriptState::Scope scope(m_scriptState.get());
+        ReadableStreamOperations::setDisturbed(m_scriptState.get(), stream());
+    } else {
+        m_stream->setIsDisturbed();
+    }
+}
+
+void BodyStreamBuffer::lockAndDisturb()
+{
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+        ScriptState::Scope scope(m_scriptState.get());
+        NonThrowableExceptionState exceptionState;
+        ReadableStreamOperations::getReader(m_scriptState.get(), stream(), exceptionState);
+        ReadableStreamOperations::setDisturbed(m_scriptState.get(), stream());
+    } else {
+        NonThrowableExceptionState exceptionState;
+        m_stream->getBytesReader(m_scriptState->getExecutionContext(), exceptionState);
+        m_stream->setIsDisturbed();
+    }
+}
+
 void BodyStreamBuffer::close()
 {
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled())
+        controller()->close();
+    else
+        m_stream->close();
     m_reader = nullptr;
-    m_stream->close();
-    m_handle.clear();
+    m_handle = nullptr;
 }
 
 void BodyStreamBuffer::error()
 {
+    if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled())
+        controller()->error(DOMException::create(NetworkError, "network error"));
+    else
+        m_stream->error(DOMException::create(NetworkError, "network error"));
     m_reader = nullptr;
-    m_stream->error(DOMException::create(NetworkError, "network error"));
-    m_handle.clear();
+    m_handle = nullptr;
 }
 
 void BodyStreamBuffer::processData()
 {
     ASSERT(m_reader);
     while (m_streamNeedsMore) {
         const void* buffer;
         size_t available;
         WebDataConsumerHandle::Result result = m_reader->beginRead(&buffer, WebDataConsumerHandle::FlagNone, &available);
         switch (result) {
         case WebDataConsumerHandle::Ok:
-            m_streamNeedsMore = m_stream->enqueue(DOMUint8Array::create(static_cast<const unsigned char*>(buffer), available));
+            if (RuntimeEnabledFeatures::responseBodyWithV8ExtraStreamEnabled()) {
+                controller()->enqueue(DOMUint8Array::create(static_cast<const unsigned char*>(buffer), available));
+                m_streamNeedsMore = controller()->desiredSize() > 0;
+            } else {
+                m_streamNeedsMore = m_stream->enqueue(DOMUint8Array::create(static_cast<const unsigned char*>(buffer), available));
+            }
             m_reader->endRead(available);
             break;
 
         case WebDataConsumerHandle::Done:
             close();
             return;
 
         case WebDataConsumerHandle::ShouldWait:
             return;
 
@@ skipping 14 matching lines @@
 
 void BodyStreamBuffer::stopLoading()
 {
     if (!m_loader)
         return;
     m_loader->cancel();
     m_loader = nullptr;
 }
 
 } // namespace blink