Remove WebDataConsumerHandle usage from DataPipeAndDataBytesConsumer.

As a step towards removing WebDataConsumerHandle stop using the type in
DataPipeAndDataBytesConsumer.  Instead this code can instead use
DataPipeBytesConsumer to read the mojo::DataPipe.

Bug: 894819
Change-Id: I8b8cac639441a22b4498bc064f1b7bfd929b6acf
Reviewed-on: https://chromium-review.googlesource.com/c/1323789Reviewed-by: Yutaka Hirano <yhirano@chromium.org>
Commit-Queue: Ben Kelly <wanderview@chromium.org>
Cr-Commit-Position: refs/heads/master@{#606712}

third_party/blink/renderer/core/fetch/data_pipe_bytes_consumer.cc

@@ -143,6 +143,7 @@ void DataPipeBytesConsumer::ClearClient() {
 
 void DataPipeBytesConsumer::Cancel() {
   DCHECK(!is_in_two_phase_read_);
+  ClearClient();
   ClearDataPipe();
   SignalComplete();
 }

third_party/blink/renderer/core/fetch/form_data_bytes_consumer.cc

@@ -7,7 +7,7 @@
 #include "third_party/blink/public/platform/platform.h"
 #include "third_party/blink/renderer/core/execution_context/execution_context.h"
 #include "third_party/blink/renderer/core/fetch/blob_bytes_consumer.h"
-#include "third_party/blink/renderer/core/fetch/bytes_consumer_for_data_consumer_handle.h"
+#include "third_party/blink/renderer/core/fetch/data_pipe_bytes_consumer.h"
 #include "third_party/blink/renderer/core/typed_arrays/dom_array_buffer.h"
 #include "third_party/blink/renderer/core/typed_arrays/dom_array_buffer_view.h"
 #include "third_party/blink/renderer/platform/blob/blob_data.h"
@@ -199,11 +199,12 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
                 std::move(pipe_producer_handle),
                 WTF::Bind(&DataPipeAndDataBytesConsumer::DataPipeGetterCallback,
                           WrapWeakPersistent(this)));
-        std::unique_ptr<WebDataConsumerHandle> consumer_handle =
-            Platform::Current()->CreateDataConsumerHandle(
-                std::move(pipe_consumer_handle));
-        data_pipe_consumer_ = new BytesConsumerForDataConsumerHandle(
-            execution_context_, std::move(consumer_handle));
+        DataPipeBytesConsumer::CompletionNotifier* completion_notifier =
+            nullptr;
+        data_pipe_consumer_ = new DataPipeBytesConsumer(
+            execution_context_, std::move(pipe_consumer_handle),
+            &completion_notifier);
+        completion_notifier_ = completion_notifier;
         if (client_)
           data_pipe_consumer_->SetClient(client_);
       }
@@ -216,14 +217,9 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
       }
 
       if (result == Result::kDone) {
-        // We've finished reading from the data pipe, but we're not done
-        // until we get the DataPipeGetterCallback signalling success.
-        if (!was_data_pipe_getter_callback_called_)
-          return Result::kShouldWait;
-
         // We're done. Move on to the next element.
-        was_data_pipe_getter_callback_called_ = false;
         data_pipe_consumer_ = nullptr;
+        completion_notifier_ = nullptr;
         ++iter_;
         return BeginRead(buffer, available);
       }
@@ -321,6 +317,7 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
     visitor->Trace(client_);
     visitor->Trace(simple_consumer_);
     visitor->Trace(data_pipe_consumer_);
+    visitor->Trace(completion_notifier_);
     BytesConsumer::Trace(visitor);
   }
 
@@ -328,22 +325,26 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
   void DataPipeGetterCallback(int32_t status, uint64_t size) {
     switch (state_) {
       case PublicState::kErrored:
+        // The error should have already been propagated to the notifier.
+        DCHECK(!completion_notifier_);
+        DCHECK(!data_pipe_consumer_);
         return;
       case PublicState::kClosed:
+        // The data_pipe_consumer_ should already be cleaned up.
+        DCHECK(!completion_notifier_);
+        DCHECK(!data_pipe_consumer_);
         return;
       case PublicState::kReadableOrWaiting:
         break;
     }
 
-    DCHECK(data_pipe_consumer_);
-    auto client = client_;
-    if (status != 0)  // 0 is net::OK.
-      SetError();
-    else
-      was_data_pipe_getter_callback_called_ = true;
-
-    if (client)
-      client->OnStateChange();
+    DCHECK(completion_notifier_);
+    if (status != 0) {
+      // 0 is net::OK.
+      completion_notifier_->SignalError(Error("error"));
+    } else {
+      completion_notifier_->SignalComplete();
+    }
   }
 
   void Close() {
@@ -353,7 +354,11 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
     state_ = PublicState::kClosed;
     ClearClient();
     simple_consumer_ = nullptr;
+    if (data_pipe_consumer_) {
+      data_pipe_consumer_->Cancel();
       data_pipe_consumer_ = nullptr;
+      completion_notifier_ = nullptr;
+    }
   }
 
   void SetError() {
@@ -364,8 +369,12 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
     error_ = Error("error");
     ClearClient();
     simple_consumer_ = nullptr;
+    if (completion_notifier_) {
+      completion_notifier_->SignalError(error_);
+      completion_notifier_ = nullptr;
       data_pipe_consumer_ = nullptr;
     }
+  }
 
   Member<ExecutionContext> execution_context_;
   PublicState state_ = PublicState::kReadableOrWaiting;
@@ -374,8 +383,8 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
   Error error_;
   Member<BytesConsumer::Client> client_;
   Member<SimpleFormDataBytesConsumer> simple_consumer_;
-  Member<BytesConsumerForDataConsumerHandle> data_pipe_consumer_;
-  bool was_data_pipe_getter_callback_called_ = false;
+  Member<DataPipeBytesConsumer> data_pipe_consumer_;
+  Member<DataPipeBytesConsumer::CompletionNotifier> completion_notifier_;
 };
 
 class ComplexFormDataBytesConsumer final : public BytesConsumer {