Make it possible to write a message pipe endpoint's peer into it.

mojo/edk/system/data_pipe_consumer_dispatcher.cc

 // Copyright 2013 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 "mojo/edk/system/data_pipe_consumer_dispatcher.h"
 
 #include <utility>
 
 #include "base/logging.h"
 #include "mojo/edk/system/data_pipe.h"
@@ skipping 54 matching lines @@
   data_pipe_->ConsumerCancelAllAwakables();
 }
 
 void DataPipeConsumerDispatcher::CloseImplNoLock() {
   mutex().AssertHeld();
   data_pipe_->ConsumerClose();
   data_pipe_ = nullptr;
 }
 
 RefPtr<Dispatcher>
-DataPipeConsumerDispatcher::CreateEquivalentDispatcherAndCloseImplNoLock() {
+DataPipeConsumerDispatcher::CreateEquivalentDispatcherAndCloseImplNoLock(
+    MessagePipe* /*message_pipe*/,
+    unsigned /*port*/) {
   mutex().AssertHeld();
 
+  CancelAllAwakablesNoLock();
+
   auto dispatcher = DataPipeConsumerDispatcher::Create();
   dispatcher->Init(std::move(data_pipe_));
   return dispatcher;
 }
 
 MojoResult DataPipeConsumerDispatcher::SetDataPipeConsumerOptionsImplNoLock(
     UserPointer<const MojoDataPipeConsumerOptions> options) {
   mutex().AssertHeld();
 
   // The default of 0 means 1 element (however big that is).
@@ skipping 134 matching lines @@
   return rv;
 }
 
 bool DataPipeConsumerDispatcher::IsBusyNoLock() const {
   mutex().AssertHeld();
   return data_pipe_->ConsumerIsBusy();
 }
 
 }  // namespace system
 }  // namespace mojo