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..df7f2d1c447cb68e61a989a84fbdbb8d35c4dbd1 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,18 +178,23 @@ 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)
+    AsyncPutBatch(PassRefPtrWillBeRawPtr<ScriptPromiseResolver> resolver, Cache* cache, Request* request, Response* response, BodyStreamBuffer2* buffer, FetchDataLoader* loader)
         : m_resolver(resolver)
         , m_cache(cache)
+        , m_buffer(buffer)
     {
         request->populateWebServiceWorkerRequest(m_webRequest);
         response->populateWebServiceWorkerResponse(m_webResponse);
+        m_buffer->registerLoader(loader);
     }
     ~AsyncPutBatch() override { }
-    void didCreateBlobHandle(PassRefPtr<BlobDataHandle> handle) override
+
+    void didFetchDataLoadedBlobHandle(PassRefPtr<BlobDataHandle> handle) override
     {
+        m_buffer->unregisterLoader();
         WebVector<WebServiceWorkerCache::BatchOperation> batchOperations(size_t(1));
         batchOperations[0].operationType = WebServiceWorkerCache::OperationTypePut;
         batchOperations[0].request = m_webRequest;
@@ -196,19 +203,22 @@ public:
         m_cache->webCache()->dispatchBatch(new CallbackPromiseAdapter<void, CacheStorageError>(m_resolver.get()), batchOperations);
         cleanup();
     }
-    void didFail(DOMException* exception) override
+
+    void didFetchDataLoadFailed() override
     {
+        m_buffer->unregisterLoader();
         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(), "FOXES ARE CUTE")); // FIXME error message

[yhirano, 2015/06/25 10:45:31]

Please use TODO.
I do like messages like "network error" as this is user-exposed.

[hiroshige, 2015/06/25 11:27:28]

Done.

         cleanup();
     }
 
     DEFINE_INLINE_VIRTUAL_TRACE()
     {
+        Client::trace(visitor);
         visitor->trace(m_resolver);
         visitor->trace(m_cache);
-        BlobHandleCreatorClient::trace(visitor);
+        visitor->trace(m_buffer);
     }
 
 private:
@@ -216,9 +226,11 @@ private:
     {
         m_resolver = nullptr;
         m_cache = nullptr;
+        m_buffer = nullptr;
     }
     RefPtrWillBeMember<ScriptPromiseResolver> m_resolver;
     Member<Cache> m_cache;
+    Member<BodyStreamBuffer2> m_buffer;
     WebServiceWorkerRequest m_webRequest;
     WebServiceWorkerResponse m_webResponse;
 };
@@ -400,12 +412,12 @@ ScriptPromise Cache::putImpl(ScriptState* scriptState, Request* request, Respons
 
     RefPtrWillBeRawPtr<ScriptPromiseResolver> resolver = ScriptPromiseResolver::create(scriptState);
     const ScriptPromise promise = resolver->promise();
-    if (BodyStreamBuffer* buffer = response->internalBuffer()) {
+    if (BodyStreamBuffer2* 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());
+        loader->start(buffer->handle(), new AsyncPutBatch(resolver, this, request, response, buffer, loader));
         return promise;
     }
     WebVector<WebServiceWorkerCache::BatchOperation> batchOperations(size_t(1));