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

Source/modules/cachestorage/Cache.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/cachestorage/Cache.h"
 
 #include "bindings/core/v8/CallbackPromiseAdapter.h"
 #include "bindings/core/v8/ExceptionState.h"
 #include "bindings/core/v8/ScriptPromiseResolver.h"
 #include "bindings/core/v8/ScriptState.h"
 #include "bindings/core/v8/V8Binding.h"
 #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"
 #include "public/platform/WebServiceWorkerCache.h"
 
 namespace blink {
 
 namespace {
 
 // FIXME: Consider using CallbackPromiseAdapter.
@@ skipping 259 matching lines @@
     {
         ASSERT(index < m_batchOperations.size());
         if (m_completed)
             return;
         m_batchOperations[index] = batchOperation;
         if (--m_numberOfRemainingOperations != 0)
             return;
         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()
     {
         visitor->trace(m_cache);
         visitor->trace(m_resolver);
     }
 
 private:
     bool m_completed = false;
     int m_numberOfRemainingOperations;
     Member<Cache> m_cache;
     RefPtrWillBeMember<ScriptPromiseResolver> m_resolver;
     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)
         , m_barrierCallback(barrierCallback)
     {
         request->populateWebServiceWorkerRequest(m_webRequest);
         response->populateWebServiceWorkerResponse(m_webResponse);
     }
     ~BlobHandleCallbackForPut() override { }
 
-    void didCreateBlobHandle(PassRefPtr<BlobDataHandle> handle) override
+    void didFetchDataLoadedBlobHandle(PassRefPtr<BlobDataHandle> handle) override
     {
         WebServiceWorkerCache::BatchOperation batchOperation;
         batchOperation.operationType = WebServiceWorkerCache::OperationTypePut;
         batchOperation.request = m_webRequest;
         batchOperation.response = m_webResponse;
         batchOperation.response.setBlobDataHandle(handle);
         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:
     const size_t m_index;
     Member<BarrierCallbackForPut> m_barrierCallback;
 
     WebServiceWorkerRequest m_webRequest;
     WebServiceWorkerResponse m_webResponse;
 };
 
@@ skipping 166 matching lines @@
 
 ScriptPromise Cache::putImpl(ScriptState* scriptState, const HeapVector<Member<Request>>& requests, const HeapVector<Member<Response>>& responses)
 {
     RefPtrWillBeRawPtr<ScriptPromiseResolver> resolver = ScriptPromiseResolver::create(scriptState);
     const ScriptPromise promise = resolver->promise();
     BarrierCallbackForPut* barrierCallback = new BarrierCallbackForPut(requests.size(), this, resolver.get());
 
     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;
         }
 
         if (requests[i]->hasBody())
             requests[i]->lockBody(Body::PassBody);
         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;
         }
 
         WebServiceWorkerCache::BatchOperation batchOperation;
         batchOperation.operationType = WebServiceWorkerCache::OperationTypePut;
         requests[i]->populateWebServiceWorkerRequest(batchOperation.request);
         responses[i]->populateWebServiceWorkerResponse(batchOperation.response);
         barrierCallback->onSuccess(i, batchOperation);
     }
 
@@ skipping 18 matching lines @@
     m_webCache->dispatchKeys(new CacheWithRequestsCallbacks(resolver), 0, toWebQueryParams(options));
     return promise;
 }
 
 WebServiceWorkerCache* Cache::webCache() const
 {
     return m_webCache.get();
 }
 
 } // namespace blink