Support creating mojo peer connections from named pipes.

mojo/edk/system/channel_posix.cc

Index: mojo/edk/system/channel_posix.cc
diff --git a/mojo/edk/system/channel_posix.cc b/mojo/edk/system/channel_posix.cc
index 16a9304462f9c5d4e48d9632174cdc1923663d13..12154d0bc1d0a3ab51c2d610f9a30eb53b2ee577 100644
--- a/mojo/edk/system/channel_posix.cc
+++ b/mojo/edk/system/channel_posix.cc
@@ -19,6 +19,7 @@
 #include "base/message_loop/message_loop.h"
 #include "base/synchronization/lock.h"
 #include "base/task_runner.h"
+#include "ipc/unix_domain_socket_util.h"
 #include "mojo/edk/embedder/platform_channel_utils_posix.h"
 #include "mojo/edk/embedder/platform_handle_vector.h"
 
@@ -211,11 +212,19 @@ class ChannelPosix : public Channel,
     DCHECK(!read_watcher_);
     DCHECK(!write_watcher_);
     read_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher);
-    write_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher);
-    base::MessageLoopForIO::current()->WatchFileDescriptor(
-        handle_.get().handle, true /* persistent */,
-        base::MessageLoopForIO::WATCH_READ, read_watcher_.get(), this);
     base::MessageLoop::current()->AddDestructionObserver(this);
+    if (handle_.get().needs_connection) {
+      base::MessageLoopForIO::current()->WatchFileDescriptor(
+          handle_.get().handle, false /* persistent */,
+          base::MessageLoopForIO::WATCH_READ, read_watcher_.get(), this);
+    } else {
+      write_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher);
+      base::MessageLoopForIO::current()->WatchFileDescriptor(
+          handle_.get().handle, true /* persistent */,
+          base::MessageLoopForIO::WATCH_READ, read_watcher_.get(), this);
+      base::AutoLock lock(write_lock_);
+      FlushOutgoingMessagesNoLock();
+    }
   }
 
   void WaitForWriteOnIOThread() {
@@ -265,6 +274,24 @@ class ChannelPosix : public Channel,
   // base::MessageLoopForIO::Watcher:
   void OnFileCanReadWithoutBlocking(int fd) override {
     CHECK_EQ(fd, handle_.get().handle);
+    if (handle_.get().needs_connection) {
+#if !defined(OS_NACL)
+      read_watcher_.reset();
+      base::MessageLoop::current()->RemoveDestructionObserver(this);
+
+      int accept_fd = -1;
+      if (!IPC::ServerOnConnect(handle_.get().handle, &accept_fd)) {
+        OnError();
+        return;
+      }
+
+      handle_.reset(PlatformHandle(accept_fd));
+      StartOnIOThread();
+#else
+      NOTREACHED();
+#endif
+      return;
+    }
 
     bool read_error = false;
     size_t next_read_size = 0;
@@ -321,6 +348,11 @@ class ChannelPosix : public Channel,
   // cannot be written, it's queued and a wait is initiated to write the message
   // ASAP on the I/O thread.
   bool WriteNoLock(MessageView message_view) {
+    if (handle_.get().needs_connection) {
+      outgoing_messages_.emplace_front(std::move(message_view));
+      return true;
+    }
+
     size_t bytes_written = 0;
     do {
       message_view.advance_data_offset(bytes_written);