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

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