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 bbe9c75e7693d29c83e60a08481562561573c82e..bb3b5961ae1cd36b6ee4c623fb197150147ff543 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"
@@ -176,7 +178,8 @@ private:
     Member<Request> m_request;
 };
 
-class Cache::AsyncPutBatch final : public BodyStreamBuffer::BlobHandleCreatorClient {
+class Cache::AsyncPutBatch final : public GarbageCollectedFinalized<AsyncPutBatch>, public FetchDataLoader::Client {
+    USING_GARBAGE_COLLECTED_MIXIN(AsyncPutBatch);
 public:
     AsyncPutBatch(PassRefPtrWillBeRawPtr<ScriptPromiseResolver> resolver, Cache* cache, Request* request, Response* response)
         : m_resolver(resolver)
@@ -186,7 +189,8 @@ public:
         response->populateWebServiceWorkerResponse(m_webResponse);
     }
     ~AsyncPutBatch() override { }
-    void didCreateBlobHandle(PassRefPtr<BlobDataHandle> handle) override
+
+    void didFetchDataLoadedBlobHandle(PassRefPtr<BlobDataHandle> handle) override
     {
         WebVector<WebServiceWorkerCache::BatchOperation> batchOperations(size_t(1));
         batchOperations[0].operationType = WebServiceWorkerCache::OperationTypePut;
@@ -196,19 +200,20 @@ public:
         m_cache->webCache()->dispatchBatch(new CallbackPromiseAdapter<void, CacheStorageError>(m_resolver.get()), batchOperations);
         cleanup();
     }
-    void didFail(DOMException* exception) override
+
+    void didFetchDataLoadFailed() override
     {
         ScriptState* state = m_resolver->scriptState();
         ScriptState::Scope scope(state);
-        m_resolver->reject(V8ThrowException::createTypeError(state->isolate(), exception->toString()));
+        m_resolver->reject(V8ThrowException::createTypeError(state->isolate(), "network error"));
         cleanup();
     }
 
     DEFINE_INLINE_VIRTUAL_TRACE()
     {
+        Client::trace(visitor);
         visitor->trace(m_resolver);
         visitor->trace(m_cache);
-        BlobHandleCreatorClient::trace(visitor);
     }
 
 private:
@@ -403,9 +408,9 @@ ScriptPromise Cache::putImpl(ScriptState* scriptState, Request* request, Respons
     if (BodyStreamBuffer* buffer = response->internalBuffer()) {
         if (buffer == response->buffer() && response->isBodyConsumed())
             buffer = response->createDrainingStream();
-        // If the response body type is stream, read the all data and create the
-        // blob handle and dispatch the put batch asynchronously.
-        buffer->readAllAndCreateBlobHandle(response->internalMIMEType(), new AsyncPutBatch(resolver, this, request, response));
+        // Create a blob handle and dispatch the put batch asynchronously.
+        FetchDataLoader* loader = FetchDataLoader::createLoaderAsBlobHandle(response->internalMIMEType());
+        buffer->startLoading(loader, new AsyncPutBatch(resolver, this, request, response));
         return promise;
     }
     WebVector<WebServiceWorkerCache::BatchOperation> batchOperations(size_t(1));