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/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 142 matching lines @@
         : ScriptFunction(scriptState)
         , m_cache(cache)
         , m_request(request)
     {
     }
 
     Member<Cache> m_cache;
     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;
         batchOperations[0].response = m_webResponse;
         batchOperations[0].response.setBlobDataHandle(handle);
         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:
     void cleanup()
     {
         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;
 };
 
 Cache* Cache::create(WeakPtr<GlobalFetch::ScopedFetcher> fetcher, WebServiceWorkerCache* webCache)
 {
     return new Cache(fetcher, webCache);
 }
 
 ScriptPromise Cache::match(ScriptState* scriptState, const RequestInfo& request, const CacheQueryOptions& options, ExceptionState& exceptionState)
@@ skipping 161 matching lines @@
     if (response->hasBody() && response->bodyUsed())
         return ScriptPromise::reject(scriptState, V8ThrowException::createTypeError(scriptState->isolate(), "Response body is already used"));
 
     if (request->hasBody())
         request->lockBody(Body::PassBody);
     if (response->hasBody())
         response->lockBody(Body::PassBody);
 
     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));
     batchOperations[0].operationType = WebServiceWorkerCache::OperationTypePut;
     request->populateWebServiceWorkerRequest(batchOperations[0].request);
     response->populateWebServiceWorkerResponse(batchOperations[0].response);
 
     m_webCache->dispatchBatch(new CallbackPromiseAdapter<void, CacheStorageError>(resolver), batchOperations);
     return promise;
 }
@@ skipping 16 matching lines @@
     m_webCache->dispatchKeys(new CacheWithRequestsCallbacks(resolver), 0, toWebQueryParams(options));
     return promise;
 }
 
 WebServiceWorkerCache* Cache::webCache() const
 {
     return m_webCache.get();
 }
 
 } // namespace blink