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

Source/modules/cachestorage/Cache.cpp

Index: Source/modules/cachestorage/Cache.cpp
diff --git a/Source/modules/cachestorage/Cache.cpp b/Source/modules/cachestorage/Cache.cpp
index d2c9e5fb112e1e5d0877272dd31c682680976102..4191869fd1d44a0e0e20c0bee4b4ef3a5cb048c3 100644
--- a/Source/modules/cachestorage/Cache.cpp
+++ b/Source/modules/cachestorage/Cache.cpp
@@ -12,8 +12,10 @@
 #include "bindings/core/v8/V8ThrowException.h"
 #include "bindings/modules/v8/V8Response.h"
 #include "core/dom/DOMException.h"
+#include "core/dom/ExceptionCode.h"
 #include "modules/cachestorage/CacheStorageError.h"
 #include "modules/fetch/BodyStreamBuffer.h"
+#include "modules/fetch/FetchDataLoader.h"
 #include "modules/fetch/GlobalFetch.h"
 #include "modules/fetch/Request.h"
 #include "modules/fetch/Response.h"
@@ -198,20 +200,14 @@ public:
         m_cache->webCache()->dispatchBatch(new CallbackPromiseAdapter<void, CacheStorageError>(m_resolver), m_batchOperations);
     }
 
-    void onError(DOMException* exception)
+    void onError(const String& errorMessage)
     {
         if (m_completed)
             return;
         m_completed = true;
-        m_resolver->reject(exception);
-    }
-
-    void onError(v8::Local<v8::Value> exception)
-    {
-        if (m_completed)
-            return;
-        m_completed = true;
-        m_resolver->reject(exception);
+        ScriptState* state = m_resolver->scriptState();
+        ScriptState::Scope scope(state);
+        m_resolver->reject(V8ThrowException::createTypeError(state->isolate(), errorMessage));
     }
 
     DEFINE_INLINE_VIRTUAL_TRACE()
@@ -228,7 +224,8 @@ private:
     Vector<WebServiceWorkerCache::BatchOperation> m_batchOperations;
 };
 
-class Cache::BlobHandleCallbackForPut final : public BodyStreamBuffer::BlobHandleCreatorClient {
+class Cache::BlobHandleCallbackForPut final : public GarbageCollectedFinalized<BlobHandleCallbackForPut>, public FetchDataLoader::Client {
+    USING_GARBAGE_COLLECTED_MIXIN(BlobHandleCallbackForPut);
 public:
     BlobHandleCallbackForPut(size_t index, BarrierCallbackForPut* barrierCallback, Request* request, Response* response)
         : m_index(index)
@@ -239,7 +236,7 @@ public:
     }
     ~BlobHandleCallbackForPut() override { }
 
-    void didCreateBlobHandle(PassRefPtr<BlobDataHandle> handle) override
+    void didFetchDataLoadedBlobHandle(PassRefPtr<BlobDataHandle> handle) override
     {
         WebServiceWorkerCache::BatchOperation batchOperation;
         batchOperation.operationType = WebServiceWorkerCache::OperationTypePut;
@@ -249,15 +246,15 @@ public:
         m_barrierCallback->onSuccess(m_index, batchOperation);
     }
 
-    void didFail(DOMException* exception) override
+    void didFetchDataLoadFailed() override
     {
-        m_barrierCallback->onError(exception);
+        m_barrierCallback->onError("network error");
     }
 
     DEFINE_INLINE_VIRTUAL_TRACE()
     {
         visitor->trace(m_barrierCallback);
-        BlobHandleCreatorClient::trace(visitor);
+        FetchDataLoader::Client::trace(visitor);
     }
 
 private:
@@ -435,19 +432,19 @@ ScriptPromise Cache::putImpl(ScriptState* scriptState, const HeapVector<Member<R
     for (size_t i = 0; i < requests.size(); ++i) {
         KURL url(KURL(), requests[i]->url());
         if (!url.protocolIsInHTTPFamily()) {
-            barrierCallback->onError(V8ThrowException::createTypeError(scriptState->isolate(), "Request scheme '" + url.protocol() + "' is unsupported"));
+            barrierCallback->onError("Request scheme '" + url.protocol() + "' is unsupported");
             return promise;
         }
         if (requests[i]->method() != "GET") {
-            barrierCallback->onError(V8ThrowException::createTypeError(scriptState->isolate(), "Request method '" + requests[i]->method() + "' is unsupported"));
+            barrierCallback->onError("Request method '" + requests[i]->method() + "' is unsupported");
             return promise;
         }
         if (requests[i]->hasBody() && requests[i]->bodyUsed()) {
-            barrierCallback->onError(V8ThrowException::createTypeError(scriptState->isolate(), "Request body is already used"));
+            barrierCallback->onError("Request body is already used");
             return promise;
         }
         if (responses[i]->hasBody() && responses[i]->bodyUsed()) {
-            barrierCallback->onError(V8ThrowException::createTypeError(scriptState->isolate(), "Response body is already used"));
+            barrierCallback->onError("Response body is already used");
             return promise;
         }
 
@@ -456,12 +453,11 @@ ScriptPromise Cache::putImpl(ScriptState* scriptState, const HeapVector<Member<R
         if (responses[i]->hasBody())
             responses[i]->lockBody(Body::PassBody);
 
-        if (BodyStreamBuffer* buffer = responses[i]->internalBuffer()) {
-            if (buffer == responses[i]->buffer() && responses[i]->isBodyConsumed())
-                buffer = responses[i]->createDrainingStream();
-            // If the response body type is stream, read the all data and create
+        if (OwnPtr<DrainingBodyStreamBuffer> buffer = responses[i]->createInternalDrainingStream()) {
+            // If the response has body, read the all data and create
             // the blob handle and dispatch the put batch asynchronously.
-            buffer->readAllAndCreateBlobHandle(responses[i]->internalMIMEType(), new BlobHandleCallbackForPut(i, barrierCallback, requests[i], responses[i]));
+            FetchDataLoader* loader = FetchDataLoader::createLoaderAsBlobHandle(responses[i]->internalMIMEType());
+            buffer->startLoading(loader, new BlobHandleCallbackForPut(i, barrierCallback, requests[i], responses[i]));
             continue;
         }