ServiceWorker: Use mojo's data pipe for respondWith(stream)

third_party/WebKit/Source/modules/serviceworkers/FetchRespondWithObserver.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 "modules/serviceworkers/FetchRespondWithObserver.h"
 
 #include <v8.h>
 #include "bindings/core/v8/ScriptValue.h"
 #include "bindings/core/v8/V8Binding.h"
 #include "bindings/modules/v8/V8Response.h"
@@ skipping 92 matching lines @@
   return frameType != WebURLRequest::FrameTypeNone;
 }
 
 bool isClientRequest(WebURLRequest::FrameType frameType,
                      WebURLRequest::RequestContext requestContext) {
   return isNavigationRequest(frameType) ||
          requestContext == WebURLRequest::RequestContextSharedWorker ||
          requestContext == WebURLRequest::RequestContextWorker;
 }
 
-class NoopLoaderClient final
-    : public GarbageCollectedFinalized<NoopLoaderClient>,
+class FetchLoaderClient final
+    : public GarbageCollectedFinalized<FetchLoaderClient>,
       public FetchDataLoader::Client {
-  WTF_MAKE_NONCOPYABLE(NoopLoaderClient);
-  USING_GARBAGE_COLLECTED_MIXIN(NoopLoaderClient);
+  WTF_MAKE_NONCOPYABLE(FetchLoaderClient);
+  USING_GARBAGE_COLLECTED_MIXIN(FetchLoaderClient);
 
  public:
-  NoopLoaderClient() = default;
-  void didFetchDataLoadedStream() override {}
-  void didFetchDataLoadFailed() override {}
+  FetchLoaderClient(std::unique_ptr<WebServiceWorkerStreamHandle> handle)
+      : m_handle(std::move(handle)) {}
+
+  void didFetchDataLoadedDataPipe() override { m_handle->completed(); }
+  void didFetchDataLoadFailed() override { m_handle->aborted(); }
   DEFINE_INLINE_TRACE() { FetchDataLoader::Client::trace(visitor); }
+
+ private:
+  std::unique_ptr<WebServiceWorkerStreamHandle> m_handle;
 };
 
 }  // namespace
 
 FetchRespondWithObserver* FetchRespondWithObserver::create(
     ExecutionContext* context,
     int fetchEventID,
     const KURL& requestURL,
     WebURLRequest::FetchRequestMode requestMode,
     WebURLRequest::FetchRedirectMode redirectMode,
@@ skipping 10 matching lines @@
   DCHECK(getExecutionContext());
   getExecutionContext()->addConsoleMessage(
       ConsoleMessage::create(JSMessageSource, WarningMessageLevel,
                              getMessageForResponseError(error, m_requestURL)));
 
   // The default value of WebServiceWorkerResponse's status is 0, which maps
   // to a network error.
   WebServiceWorkerResponse webResponse;
   webResponse.setError(error);
   ServiceWorkerGlobalScopeClient::from(getExecutionContext())
-      ->respondToFetchEvent(m_eventID, webResponse, m_eventDispatchTime);
+      ->respondToFetchEventWithResponse(m_eventID, webResponse,
+                                        m_eventDispatchTime);
 }
 
 void FetchRespondWithObserver::onResponseFulfilled(const ScriptValue& value) {
   DCHECK(getExecutionContext());
   if (!V8Response::hasInstance(value.v8Value(),
                                toIsolate(getExecutionContext()))) {
     onResponseRejected(WebServiceWorkerResponseErrorNoV8Instance);
     return;
   }
   Response* response = V8Response::toImplWithTypeCheck(
@@ skipping 45 matching lines @@
     return;
   }
 
   WebServiceWorkerResponse webResponse;
   response->populateWebServiceWorkerResponse(webResponse);
   BodyStreamBuffer* buffer = response->internalBodyBuffer();
   if (buffer) {
     RefPtr<BlobDataHandle> blobDataHandle = buffer->drainAsBlobDataHandle(
         BytesConsumer::BlobSizePolicy::AllowBlobWithInvalidSize);
     if (blobDataHandle) {
+      // Handle the blob response.
       webResponse.setBlobDataHandle(blobDataHandle);
+      ServiceWorkerGlobalScopeClient::from(getExecutionContext())
+          ->respondToFetchEventWithResponse(m_eventID, webResponse,
+                                            m_eventDispatchTime);
+      return;
     } else {
-      Stream* outStream = Stream::create(getExecutionContext(), "");
-      webResponse.setStreamURL(outStream->url());
-      buffer->startLoading(FetchDataLoader::createLoaderAsStream(outStream),
-                           new NoopLoaderClient);
+      // Handle the stream response.
+      mojo::DataPipe dataPipe;
+      DCHECK(dataPipe.producer_handle.is_valid());
+      DCHECK(dataPipe.consumer_handle.is_valid());
+
+      std::unique_ptr<WebServiceWorkerStreamHandle> streamHandle =
+          WTF::makeUnique<WebServiceWorkerStreamHandle>(
+              std::move(dataPipe.consumer_handle));
+      WebServiceWorkerStreamHandle* streamHandleRawptr = streamHandle.get();
+
+      buffer->startLoading(FetchDataLoader::createLoaderAsDataPipe(
+                               std::move(dataPipe.producer_handle)),
+                           new FetchLoaderClient(std::move(streamHandle)));
+      ServiceWorkerGlobalScopeClient::from(getExecutionContext())
+          ->respondToFetchEventWithResponseStream(
+              m_eventID, webResponse, streamHandleRawptr, m_eventDispatchTime);
+      return;
     }
   }
   ServiceWorkerGlobalScopeClient::from(getExecutionContext())
-      ->respondToFetchEvent(m_eventID, webResponse, m_eventDispatchTime);
+      ->respondToFetchEventWithResponse(m_eventID, webResponse,
+                                        m_eventDispatchTime);
 }
 
 void FetchRespondWithObserver::onNoResponse() {
   ServiceWorkerGlobalScopeClient::from(getExecutionContext())
       ->respondToFetchEvent(m_eventID, m_eventDispatchTime);
 }
 
 FetchRespondWithObserver::FetchRespondWithObserver(
     ExecutionContext* context,
     int fetchEventID,
     const KURL& requestURL,
     WebURLRequest::FetchRequestMode requestMode,
     WebURLRequest::FetchRedirectMode redirectMode,
     WebURLRequest::FrameType frameType,
     WebURLRequest::RequestContext requestContext,
     WaitUntilObserver* observer)
     : RespondWithObserver(context, fetchEventID, observer),
       m_requestURL(requestURL),
       m_requestMode(requestMode),
       m_redirectMode(redirectMode),
       m_frameType(frameType),
       m_requestContext(requestContext) {}
 
 DEFINE_TRACE(FetchRespondWithObserver) {
   RespondWithObserver::trace(visitor);
 }
 
 }  // namespace blink