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

Source/modules/fetch/FetchResponseData.cpp

Index: Source/modules/fetch/FetchResponseData.cpp
diff --git a/Source/modules/fetch/FetchResponseData.cpp b/Source/modules/fetch/FetchResponseData.cpp
index 1e331e3d4878f45531236e374a4e42342740aa9b..a5054184f1e2e22928bd57fb541f1434ddc80f4b 100644
--- a/Source/modules/fetch/FetchResponseData.cpp
+++ b/Source/modules/fetch/FetchResponseData.cpp
@@ -8,6 +8,8 @@
 #include "core/dom/DOMArrayBuffer.h"
 #include "core/fetch/CrossOriginAccessControl.h"
 #include "modules/fetch/BodyStreamBuffer.h"
+#include "modules/fetch/DataConsumerHandleUtil.h"
+#include "modules/fetch/DataConsumerTee.h"
 #include "modules/fetch/FetchHeaderList.h"
 #include "public/platform/WebServiceWorkerResponse.h"
 
@@ -15,68 +17,6 @@ namespace blink {
 
 namespace {
 
-class BranchCanceller : public BodyStreamBuffer::Canceller {
-public:
-    static void create(BodyStreamBuffer* buffer, BranchCanceller** canceller1, BranchCanceller** canceller2)
-    {
-        auto context = new Context(buffer);
-        *canceller1 = new BranchCanceller(context, First);
-        *canceller2 = new BranchCanceller(context, Second);
-    }
-
-    void setBuffer(BodyStreamBuffer* buffer) { m_buffer = buffer; }
-
-    void cancel() override
-    {
-        if (m_tag == First) {
-            m_context->isFirstCancelled = true;
-        } else {
-            ASSERT(m_tag == Second);
-            m_context->isSecondCancelled = true;
-        }
-        ASSERT(m_buffer);
-        ASSERT(!m_buffer->isClosed());
-        ASSERT(!m_buffer->hasError());
-        m_buffer->close();
-        if (m_context->isFirstCancelled && m_context->isSecondCancelled)
-            m_context->buffer->cancel();
-    }
-
-    DEFINE_INLINE_VIRTUAL_TRACE()
-    {
-        visitor->trace(m_context);
-        visitor->trace(m_buffer);
-        BodyStreamBuffer::Canceller::trace(visitor);
-    }
-
-private:
-    enum Tag {
-        First,
-        Second,
-    };
-    class Context : public GarbageCollected<Context> {
-    public:
-        explicit Context(BodyStreamBuffer* buffer)
-            : buffer(buffer)
-            , isFirstCancelled(false)
-            , isSecondCancelled(false) { }
-
-        DEFINE_INLINE_VIRTUAL_TRACE()
-        {
-            visitor->trace(buffer);
-        }
-
-        Member<BodyStreamBuffer> buffer;
-        bool isFirstCancelled;
-        bool isSecondCancelled;
-    };
-
-    BranchCanceller(Context* context, Tag tag) : m_context(context), m_tag(tag) { }
-    Member<Context> m_context;
-    Member<BodyStreamBuffer> m_buffer;
-    Tag m_tag;
-};
-
 WebServiceWorkerResponseType fetchTypeToWebType(FetchResponseData::Type fetchType)
 {
     WebServiceWorkerResponseType webType = WebServiceWorkerResponseTypeDefault;
@@ -137,7 +77,6 @@ FetchResponseData* FetchResponseData::createBasicFilteredResponse()
             continue;
         response->m_headerList->append(header->first, header->second);
     }
-    response->m_blobDataHandle = m_blobDataHandle;
     response->m_buffer = m_buffer;
     response->m_mimeType = m_mimeType;
     response->m_internalResponse = this;
@@ -165,7 +104,6 @@ FetchResponseData* FetchResponseData::createCORSFilteredResponse()
             continue;
         response->m_headerList->append(header->first, header->second);
     }
-    response->m_blobDataHandle = m_blobDataHandle;
     response->m_buffer = m_buffer;
     response->m_mimeType = m_mimeType;
     response->m_internalResponse = this;
@@ -187,14 +125,6 @@ String FetchResponseData::mimeType() const
     return m_mimeType;
 }
 
-PassRefPtr<BlobDataHandle> FetchResponseData::internalBlobDataHandle() const
-{
-    if (m_internalResponse) {
-        return m_internalResponse->m_blobDataHandle;
-    }
-    return m_blobDataHandle;
-}
-
 BodyStreamBuffer* FetchResponseData::internalBuffer() const
 {
     if (m_internalResponse) {
@@ -211,7 +141,7 @@ String FetchResponseData::internalMIMEType() const
     return m_mimeType;
 }
 
-FetchResponseData* FetchResponseData::clone()
+FetchResponseData* FetchResponseData::clone(ExecutionContext* executionContext)
 {
     FetchResponseData* newResponse = create();
     newResponse->m_type = m_type;
@@ -223,17 +153,15 @@ FetchResponseData* FetchResponseData::clone()
     newResponse->m_status = m_status;
     newResponse->m_statusMessage = m_statusMessage;
     newResponse->m_headerList = m_headerList->clone();
-    newResponse->m_blobDataHandle = m_blobDataHandle;
     newResponse->m_mimeType = m_mimeType;
 
     switch (m_type) {
     case BasicType:
     case CORSType:
         ASSERT(m_internalResponse);
-        ASSERT(m_blobDataHandle == m_internalResponse->m_blobDataHandle);
         ASSERT(m_buffer == m_internalResponse->m_buffer);
         ASSERT(m_internalResponse->m_type == DefaultType);
-        newResponse->m_internalResponse = m_internalResponse->clone();
+        newResponse->m_internalResponse = m_internalResponse->clone(executionContext);
         m_buffer = m_internalResponse->m_buffer;
         newResponse->m_buffer = newResponse->m_internalResponse->m_buffer;
         break;
@@ -241,28 +169,23 @@ FetchResponseData* FetchResponseData::clone()
         ASSERT(!m_internalResponse);
         if (!m_buffer)
             return newResponse;
-        BodyStreamBuffer* original = m_buffer;
-        BranchCanceller* canceller1 = nullptr;
-        BranchCanceller* canceller2 = nullptr;
-        BranchCanceller::create(original, &canceller1, &canceller2);
-        m_buffer = new BodyStreamBuffer(canceller1);
-        newResponse->m_buffer = new BodyStreamBuffer(canceller2);
-        canceller1->setBuffer(m_buffer);
-        canceller2->setBuffer(newResponse->m_buffer);
-        original->startTee(m_buffer, newResponse->m_buffer);
+
+        OwnPtr<WebDataConsumerHandle> handle1;
+        OwnPtr<WebDataConsumerHandle> handle2;
+        DataConsumerTee::create(executionContext, m_buffer->releaseHandle(), &handle1, &handle2);
+        m_buffer = BodyStreamBuffer::create(createFetchDataConsumerHandleFromWebHandle(handle1.release()));
+        newResponse->m_buffer = BodyStreamBuffer::create(createFetchDataConsumerHandleFromWebHandle(handle2.release()));
         break;
     }
     case ErrorType:
         ASSERT(!m_internalResponse);
-        ASSERT(!m_blobDataHandle);
         ASSERT(!m_buffer);
         break;
     case OpaqueType:
         ASSERT(m_internalResponse);
-        ASSERT(!m_blobDataHandle);
         ASSERT(!m_buffer);
         ASSERT(m_internalResponse->m_type == DefaultType);
-        newResponse->m_internalResponse = m_internalResponse->clone();
+        newResponse->m_internalResponse = m_internalResponse->clone(executionContext);
         break;
     }
     return newResponse;
@@ -284,7 +207,6 @@ void FetchResponseData::populateWebServiceWorkerResponse(WebServiceWorkerRespons
         const FetchHeaderList::Header* header = headerList()->list()[i].get();
         response.appendHeader(header->first, header->second);
     }
-    response.setBlobDataHandle(m_blobDataHandle);
 }
 
 FetchResponseData::FetchResponseData(Type type, unsigned short status, AtomicString statusMessage)
@@ -295,23 +217,14 @@ FetchResponseData::FetchResponseData(Type type, unsigned short status, AtomicStr
 {
 }
 
-void FetchResponseData::setBlobDataHandle(PassRefPtr<BlobDataHandle> blobDataHandle)
-{
-    ASSERT(!m_buffer);
-    m_blobDataHandle = blobDataHandle;
-}
-
 void FetchResponseData::replaceBodyStreamBuffer(BodyStreamBuffer* buffer)
 {
     if (m_type == BasicType || m_type == CORSType) {
         ASSERT(m_internalResponse);
-        m_internalResponse->m_blobDataHandle = nullptr;
         m_internalResponse->m_buffer = buffer;
-        m_blobDataHandle = nullptr;
         m_buffer = buffer;
     } else if (m_type == DefaultType) {
         ASSERT(!m_internalResponse);
-        m_blobDataHandle = nullptr;
         m_buffer = buffer;
     }
 }