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

Source/modules/serviceworkers/RespondWithObserver.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/serviceworkers/RespondWithObserver.h"
 
 #include "bindings/core/v8/ScriptFunction.h"
 #include "bindings/core/v8/ScriptPromise.h"
 #include "bindings/core/v8/ScriptValue.h"
 #include "bindings/core/v8/V8Binding.h"
 #include "bindings/modules/v8/V8Response.h"
 #include "core/dom/ExceptionCode.h"
 #include "core/dom/ExecutionContext.h"
 #include "core/inspector/ConsoleMessage.h"
 #include "core/streams/Stream.h"
 #include "modules/fetch/BodyStreamBuffer.h"
 #include "modules/serviceworkers/ServiceWorkerGlobalScopeClient.h"
 #include "platform/RuntimeEnabledFeatures.h"
 #include "public/platform/WebServiceWorkerResponse.h"
 #include "wtf/Assertions.h"
 #include "wtf/RefPtr.h"
 #include <v8.h>
 
 namespace blink {
-namespace {
-
-class StreamUploader : public BodyStreamBuffer::Observer {
-public:
-    StreamUploader(BodyStreamBuffer* buffer, Stream* outStream)
-        : m_buffer(buffer), m_outStream(outStream)
-    {
-    }
-    ~StreamUploader() override { }
-    void onWrite() override
-    {
-        bool needToFlush = false;
-        while (RefPtr<DOMArrayBuffer> buf = m_buffer->read()) {
-            needToFlush = true;
-            m_outStream->addData(static_cast<const char*>(buf->data()), buf->byteLength());
-        }
-        if (needToFlush)
-            m_outStream->flush();
-    }
-    void onClose() override
-    {
-        m_outStream->finalize();
-        cleanup();
-    }
-    void onError() override
-    {
-        // If the stream is aborted soon after the stream is registered to the
-        // StreamRegistry, ServiceWorkerURLRequestJob may not notice the error
-        // and continue waiting forever.
-        // FIXME: Add new message to report the error to the browser process.
-        m_outStream->abort();
-        cleanup();
-    }
-    DEFINE_INLINE_TRACE()
-    {
-        visitor->trace(m_buffer);
-        visitor->trace(m_outStream);
-        BodyStreamBuffer::Observer::trace(visitor);
-    }
-    void start()
-    {
-        m_buffer->registerObserver(this);
-        onWrite();
-        if (m_buffer->hasError())
-            return onError();
-        if (m_buffer->isClosed())
-            return onClose();
-    }
-
-private:
-    void cleanup()
-    {
-        m_buffer->unregisterObserver();
-        m_buffer.clear();
-        m_outStream.clear();
-    }
-    Member<BodyStreamBuffer> m_buffer;
-    Member<Stream> m_outStream;
-};
-
-} // namespace
 
 class RespondWithObserver::ThenFunction final : public ScriptFunction {
 public:
     enum ResolveType {
         Fulfilled,
         Rejected,
     };
 
     static v8::Local<v8::Function> createFunction(ScriptState* scriptState, RespondWithObserver* observer, ResolveType type)
     {
@@ skipping 100 matching lines @@
         || (m_requestMode != WebURLRequest::FetchRequestModeNoCORS && responseType == FetchResponseData::OpaqueType)
         || (m_frameType != WebURLRequest::FrameTypeNone && responseType != FetchResponseData::BasicType && responseType != FetchResponseData::DefaultType)) {
         responseWasRejected();
         return;
     }
     if (response->bodyUsed()) {
         responseWasRejected();
         return;
     }
     response->lockBody(Body::PassBody);
-    if (BodyStreamBuffer* buffer = response->internalBuffer()) {
-        if (buffer == response->buffer() && response->isBodyConsumed())
-            buffer = response->createDrainingStream();
+    if (OwnPtr<DrainingBodyStreamBuffer> buffer = response->createInternalDrainingStream()) {
         WebServiceWorkerResponse webResponse;
         response->populateWebServiceWorkerResponse(webResponse);
+        if (RefPtr<BlobDataHandle> blobDataHandle = buffer->drainAsBlobDataHandle(FetchDataConsumerHandle::Reader::AllowBlobWithInvalidSize)) {
+            webResponse.setBlobDataHandle(blobDataHandle);
+            ServiceWorkerGlobalScopeClient::from(executionContext())->didHandleFetchEvent(m_eventID, webResponse);
+            m_state = Done;
+            return;
+        }
         Stream* outStream = Stream::create(executionContext(), "");
         webResponse.setStreamURL(outStream->url());
         ServiceWorkerGlobalScopeClient::from(executionContext())->didHandleFetchEvent(m_eventID, webResponse);
-        StreamUploader* uploader = new StreamUploader(buffer, outStream);
-        uploader->start();
+        FetchDataLoader* loader = FetchDataLoader::createLoaderAsStream(outStream);
+        buffer->startLoading(loader, nullptr);
         m_state = Done;
         return;
     }
     WebServiceWorkerResponse webResponse;
     response->populateWebServiceWorkerResponse(webResponse);
     ServiceWorkerGlobalScopeClient::from(executionContext())->didHandleFetchEvent(m_eventID, webResponse);
     m_state = Done;
 }
 
 RespondWithObserver::RespondWithObserver(ExecutionContext* context, int eventID, WebURLRequest::FetchRequestMode requestMode, WebURLRequest::FrameType frameType)
     : ContextLifecycleObserver(context)
     , m_eventID(eventID)
     , m_requestMode(requestMode)
     , m_frameType(frameType)
     , m_state(Initial)
 {
 }
 
 DEFINE_TRACE(RespondWithObserver)
 {
     ContextLifecycleObserver::trace(visitor);
 }
 
 } // namespace blink