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 95 matching lines @@
   return frame_type != WebURLRequest::kFrameTypeNone;
 }
 
 bool IsClientRequest(WebURLRequest::FrameType frame_type,
                      WebURLRequest::RequestContext request_context) {
   return IsNavigationRequest(frame_type) ||
          request_context == WebURLRequest::kRequestContextSharedWorker ||
          request_context == WebURLRequest::kRequestContextWorker;
 }
 
-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)
+      : handle_(std::move(handle)) {}
+
+  void DidFetchDataLoadedDataPipe() override { handle_->Completed(); }
+  void DidFetchDataLoadFailed() override { handle_->Aborted(); }
   DEFINE_INLINE_TRACE() { FetchDataLoader::Client::Trace(visitor); }
+
+ private:
+  std::unique_ptr<WebServiceWorkerStreamHandle> handle_;
 };
 
 }  // namespace
 
 FetchRespondWithObserver* FetchRespondWithObserver::Create(
     ExecutionContext* context,
     int fetch_event_id,
     const KURL& request_url,
     WebURLRequest::FetchRequestMode request_mode,
     WebURLRequest::FetchRedirectMode redirect_mode,
     WebURLRequest::FrameType frame_type,
     WebURLRequest::RequestContext request_context,
     WaitUntilObserver* observer) {
   return new FetchRespondWithObserver(context, fetch_event_id, request_url,
                                       request_mode, redirect_mode, frame_type,
                                       request_context, observer);
 }
 
 void FetchRespondWithObserver::OnResponseRejected(
     WebServiceWorkerResponseError error) {
   DCHECK(GetExecutionContext());
   GetExecutionContext()->AddConsoleMessage(
       ConsoleMessage::Create(kJSMessageSource, kWarningMessageLevel,
                              GetMessageForResponseError(error, request_url_)));
 
   // The default value of WebServiceWorkerResponse's status is 0, which maps
   // to a network error.
-  WebServiceWorkerResponse web_response;
-  web_response.SetError(error);
+  WebServiceWorkerResponse webResponse;
+  webResponse.SetError(error);
   ServiceWorkerGlobalScopeClient::From(GetExecutionContext())
-      ->RespondToFetchEvent(event_id_, web_response, event_dispatch_time_);
+      ->RespondToFetchEventWithResponse(event_id_, webResponse,
+                                        event_dispatch_time_);
 }
 
 void FetchRespondWithObserver::OnResponseFulfilled(const ScriptValue& value) {
   DCHECK(GetExecutionContext());
   if (!V8Response::hasInstance(value.V8Value(),
                                ToIsolate(GetExecutionContext()))) {
     OnResponseRejected(kWebServiceWorkerResponseErrorNoV8Instance);
     return;
   }
   Response* response = V8Response::toImplWithTypeCheck(
@@ skipping 47 matching lines @@
     return;
   }
 
   WebServiceWorkerResponse web_response;
   response->PopulateWebServiceWorkerResponse(web_response);
   BodyStreamBuffer* buffer = response->InternalBodyBuffer();
   if (buffer) {
     RefPtr<BlobDataHandle> blob_data_handle = buffer->DrainAsBlobDataHandle(
         BytesConsumer::BlobSizePolicy::kAllowBlobWithInvalidSize);
     if (blob_data_handle) {
+      // Handle the blob response.
       web_response.SetBlobDataHandle(blob_data_handle);
+      ServiceWorkerGlobalScopeClient::From(GetExecutionContext())
+          ->RespondToFetchEventWithResponse(event_id_, web_response,
+                                            event_dispatch_time_);
+      return;
     } else {
-      Stream* out_stream = Stream::Create(GetExecutionContext(), "");
-      web_response.SetStreamURL(out_stream->Url());
-      buffer->StartLoading(FetchDataLoader::CreateLoaderAsStream(out_stream),
-                           new NoopLoaderClient);
+      // Handle the stream response.
+      mojo::DataPipe data_pipe;
+      DCHECK(data_pipe.producer_handle.is_valid());
+      DCHECK(data_pipe.consumer_handle.is_valid());
+
+      std::unique_ptr<WebServiceWorkerStreamHandle> stream_handle =
+          WTF::MakeUnique<WebServiceWorkerStreamHandle>(
+              std::move(data_pipe.consumer_handle));
+      WebServiceWorkerStreamHandle* stream_handle_rawptr = stream_handle.get();
+
+      buffer->StartLoading(FetchDataLoader::CreateLoaderAsDataPipe(
+                               std::move(data_pipe.producer_handle)),
+                           new FetchLoaderClient(std::move(stream_handle)));
+      ServiceWorkerGlobalScopeClient::From(GetExecutionContext())
+          ->RespondToFetchEventWithResponseStream(event_id_, web_response,

[horo, 2017/04/10 09:24:38]

It is safe to call StartLoading() before
RespondToFetchEventWithResponseStream(), because
FetchDataLoaderAsDataPipe::Start() starts loading asynchronously.

But I think calling RespondToFetchEventWithResponseStream() before
StartLoading() looks more clear.
WDYT?

[shimazu, 2017/04/11 01:32:58]

It makes sense. Changed.

+                                                  stream_handle_rawptr,
+                                                  event_dispatch_time_);
+      return;
     }
   }
   ServiceWorkerGlobalScopeClient::From(GetExecutionContext())
-      ->RespondToFetchEvent(event_id_, web_response, event_dispatch_time_);
+      ->RespondToFetchEventWithResponse(event_id_, web_response,
+                                        event_dispatch_time_);
 }
 
 void FetchRespondWithObserver::OnNoResponse() {
   ServiceWorkerGlobalScopeClient::From(GetExecutionContext())
       ->RespondToFetchEvent(event_id_, event_dispatch_time_);
 }
 
 FetchRespondWithObserver::FetchRespondWithObserver(
     ExecutionContext* context,
     int fetch_event_id,
     const KURL& request_url,
     WebURLRequest::FetchRequestMode request_mode,
     WebURLRequest::FetchRedirectMode redirect_mode,
     WebURLRequest::FrameType frame_type,
     WebURLRequest::RequestContext request_context,
     WaitUntilObserver* observer)
     : RespondWithObserver(context, fetch_event_id, observer),
       request_url_(request_url),
       request_mode_(request_mode),
       redirect_mode_(redirect_mode),
       frame_type_(frame_type),
       request_context_(request_context) {}
 
 DEFINE_TRACE(FetchRespondWithObserver) {
   RespondWithObserver::Trace(visitor);
 }
 
 }  // namespace blink