Revert "Remove WebDataConsumerHandle usage from DataPipeAndDataBytesConsumer."

This reverts commit 06b39212.

Reason for revert: to fix the test external/wpt/workers/semantics/structured-clone/shared.html

first started to fail in build: https://ci.chromium.org/p/chromium/builders/luci.chromium.ci/Linux%20Tests%20%28dbg%29%281%29/75530

Original change's description:
> 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: Ie9cad46dfa9d67998d23bf0efd6300870428f3cb
> Reviewed-on: https://chromium-review.googlesource.com/c/1313389
> Reviewed-by: Yutaka Hirano <yhirano@chromium.org>
> Commit-Queue: Ben Kelly <wanderview@chromium.org>
> Cr-Commit-Position: refs/heads/master@{#605689}

TBR=yhirano@chromium.org,wanderview@chromium.org

Change-Id: I6a792df67c062020085362929f402c433d3fa60e
No-Presubmit: true
No-Tree-Checks: true
No-Try: true
Bug: 894819
Reviewed-on: https://chromium-review.googlesource.com/c/1321856Reviewed-by: Dmitry Titov <dimich@chromium.org>
Commit-Queue: Dmitry Titov <dimich@chromium.org>
Cr-Commit-Position: refs/heads/master@{#605909}

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

@@ -143,7 +143,6 @@ 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/data_pipe_bytes_consumer.h"
+#include "third_party/blink/renderer/core/fetch/bytes_consumer_for_data_consumer_handle.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,12 +199,11 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
                 std::move(pipe_producer_handle),
                 WTF::Bind(&DataPipeAndDataBytesConsumer::DataPipeGetterCallback,
                           WrapWeakPersistent(this)));
-        DataPipeBytesConsumer::CompletionNotifier* completion_notifier =
-            nullptr;
-        data_pipe_consumer_ = new DataPipeBytesConsumer(
-            execution_context_, std::move(pipe_consumer_handle),
-            &completion_notifier);
-        completion_notifier_ = completion_notifier;
+        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));
         if (client_)
           data_pipe_consumer_->SetClient(client_);
       }
@@ -225,7 +224,6 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
         // 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);
       }
@@ -323,7 +321,6 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
     visitor->Trace(client_);
     visitor->Trace(simple_consumer_);
     visitor->Trace(data_pipe_consumer_);
-    visitor->Trace(completion_notifier_);
     BytesConsumer::Trace(visitor);
   }
 
@@ -331,28 +328,19 @@ 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(completion_notifier_);
+    DCHECK(data_pipe_consumer_);
     auto client = client_;
-    if (status != 0) {
-      // 0 is net::OK.
+    if (status != 0)  // 0 is net::OK.
       SetError();
-    } else {
+    else
       was_data_pipe_getter_callback_called_ = true;
-      completion_notifier_->SignalComplete();
-    }
 
     if (client)
       client->OnStateChange();
@@ -365,11 +353,7 @@ 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;
-    }
+    data_pipe_consumer_ = nullptr;
   }
 
   void SetError() {
@@ -380,11 +364,7 @@ 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;
-    }
+    data_pipe_consumer_ = nullptr;
   }
 
   Member<ExecutionContext> execution_context_;
@@ -394,8 +374,7 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
   Error error_;
   Member<BytesConsumer::Client> client_;
   Member<SimpleFormDataBytesConsumer> simple_consumer_;
-  Member<DataPipeBytesConsumer> data_pipe_consumer_;
-  Member<DataPipeBytesConsumer::CompletionNotifier> completion_notifier_;
+  Member<BytesConsumerForDataConsumerHandle> data_pipe_consumer_;
   bool was_data_pipe_getter_callback_called_ = false;
 };