Change BodyStreamBuffer to be FetchDataConsumerHandle-based and enable backpressure in Fetch API

Source/modules/fetch/Response.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/Response.h"
 
 #include "bindings/core/v8/Dictionary.h"
 #include "bindings/core/v8/ExceptionState.h"
 #include "core/dom/DOMArrayBuffer.h"
 #include "core/dom/DOMArrayBufferView.h"
 #include "core/fileapi/Blob.h"
 #include "core/html/DOMFormData.h"
+#include "modules/fetch/BodyStreamBuffer.h"
+#include "modules/fetch/FetchBlobDataConsumerHandle.h"
 #include "modules/fetch/ResponseInit.h"
 #include "platform/network/FormData.h"
 #include "platform/network/HTTPHeaderMap.h"
 #include "public/platform/WebServiceWorkerResponse.h"
 #include "wtf/RefPtr.h"
 
 namespace blink {
 
 namespace {
 
-FetchResponseData* createFetchResponseDataFromWebResponse(const WebServiceWorkerResponse& webResponse)
+FetchResponseData* createFetchResponseDataFromWebResponse(ExecutionContext* executionContext, const WebServiceWorkerResponse& webResponse)
 {
     FetchResponseData* response = 0;
     if (200 <= webResponse.status() && webResponse.status() < 300)
         response = FetchResponseData::create();
     else
         response = FetchResponseData::createNetworkErrorResponse();
 
     response->setURL(webResponse.url());
     response->setStatus(webResponse.status());
     response->setStatusMessage(webResponse.statusText());
 
     for (HTTPHeaderMap::const_iterator i = webResponse.headers().begin(), end = webResponse.headers().end(); i != end; ++i) {
         response->headerList()->append(i->key, i->value);
     }
 
-    response->setBlobDataHandle(webResponse.blobDataHandle());
+    response->replaceBodyStreamBuffer(BodyStreamBuffer::create(FetchBlobDataConsumerHandle::create(executionContext, webResponse.blobDataHandle())));
 
     // Filter the response according to |webResponse|'s ResponseType.
     switch (webResponse.responseType()) {
     case WebServiceWorkerResponseTypeBasic:
         response = response->createBasicFilteredResponse();
         break;
     case WebServiceWorkerResponseTypeCORS:
         response = response->createCORSFilteredResponse();
         break;
     case WebServiceWorkerResponseTypeOpaque:
@@ skipping 147 matching lines @@
     if (body) {
         // "1. Let |stream| and |Content-Type| be the result of extracting body."
         // "2. Set |r|'s response's body to |stream|."
         // "3. If |Content-Type| is non-null and |r|'s response's header list
         // contains no header named `Content-Type`, append `Content-Type`/
         // |Content-Type| to |r|'s response's header list."
         // https://fetch.spec.whatwg.org/#concept-bodyinit-extract
         // Step 3, Blob:
         // "If object's type attribute is not the empty byte sequence, set
         // Content-Type to its value."
-        r->m_response->setBlobDataHandle(body->blobDataHandle());
-        r->setBody(body->blobDataHandle());
+        r->m_response->replaceBodyStreamBuffer(BodyStreamBuffer::create(FetchBlobDataConsumerHandle::create(context, body->blobDataHandle())));
+        r->refreshBody();
         if (!body->type().isEmpty() && !r->m_response->headerList()->has("Content-Type"))
             r->m_response->headerList()->append("Content-Type", body->type());
     }
 
     // "8. Set |r|'s MIME type to the result of extracting a MIME type
     // from |r|'s response's header list."
     r->m_response->setMIMEType(r->m_response->headerList()->extractMIMEType());
 
     // "9. Return |r|."
     return r;
 }
 
 Response* Response::create(ExecutionContext* context, FetchResponseData* response)
 {
     Response* r = new Response(context, response);
     r->suspendIfNeeded();
     return r;
 }
 
 Response* Response::create(ExecutionContext* context, const WebServiceWorkerResponse& webResponse)
 {
-    FetchResponseData* responseData = createFetchResponseDataFromWebResponse(webResponse);
+    FetchResponseData* responseData = createFetchResponseDataFromWebResponse(context, webResponse);
     Response* r = new Response(context, responseData);
     r->suspendIfNeeded();
     return r;
 }
 
 Response* Response::error(ExecutionContext* context)
 {
     FetchResponseData* responseData = FetchResponseData::createNetworkErrorResponse();
     Response* r = new Response(context, responseData);
     r->m_headers->setGuard(Headers::ImmutableGuard);
@@ skipping 78 matching lines @@
     // "The headers attribute's getter must return the associated Headers object."
     return m_headers;
 }
 
 Response* Response::clone(ExceptionState& exceptionState)
 {
     if (bodyUsed()) {
         exceptionState.throwTypeError("Response body is already used");
         return nullptr;
     }
-    if (isBodyConsumed()) {
-        BodyStreamBuffer* drainingStream = createDrainingStream();
-        m_response->replaceBodyStreamBuffer(drainingStream);
-    }
 
-    FetchResponseData* response = m_response->clone();
+    if (OwnPtr<DrainingBodyStreamBuffer> buffer = createDrainingStream())
+        m_response->replaceBodyStreamBuffer(buffer->leakBuffer());
+
+    FetchResponseData* response = m_response->clone(executionContext());
     Headers* headers = Headers::create(response->headerList());
     headers->setGuard(m_headers->guard());
     Response* r = new Response(executionContext(), response, headers);
     r->suspendIfNeeded();
 
     // Lock the old body and set |body| property to the new one.
     lockBody();
     refreshBody();
 
     return r;
@@ skipping 23 matching lines @@
 }
 
 Response::Response(ExecutionContext* context, FetchResponseData* response, Headers* headers)
     : Body(context) , m_response(response) , m_headers(headers)
 {
     refreshBody();
 }
 
 bool Response::hasBody() const
 {
-    return internalBlobDataHandle() || internalBuffer();
+    return m_response->internalBuffer();
 }
 
-PassRefPtr<BlobDataHandle> Response::blobDataHandle() const
-{
-    return m_response->blobDataHandle();
-}
-
-BodyStreamBuffer* Response::buffer() const
+void* Response::bufferForTest() const
 {
     return m_response->buffer();
 }
 
 String Response::mimeType() const
 {
     return m_response->mimeType();
 }
 
-PassRefPtr<BlobDataHandle> Response::internalBlobDataHandle() const
-{
-    return m_response->internalBlobDataHandle();
-}
-
-BodyStreamBuffer* Response::internalBuffer() const
+void* Response::internalBufferForTest() const
 {
     return m_response->internalBuffer();
 }
 
 String Response::internalMIMEType() const
 {
     return m_response->internalMIMEType();
 }
 
+PassOwnPtr<DrainingBodyStreamBuffer> Response::createInternalDrainingStream()
+{
+    if (BodyStreamBuffer* buffer = m_response->internalBuffer()) {
+        if (buffer == m_response->buffer())
+            return createDrainingStream();
+        return DrainingBodyStreamBuffer::create(buffer, nullptr);
+    }
+    return nullptr;
+}
+
 void Response::refreshBody()
 {
-    if (m_response->buffer())
-        setBody(m_response->buffer());
-    else
-        setBody(m_response->blobDataHandle());
+    setBody(m_response->buffer());
 }
 
 DEFINE_TRACE(Response)
 {
     Body::trace(visitor);
     visitor->trace(m_response);
     visitor->trace(m_headers);
 }
 
 } // namespace blink