Support creating mojo peer connections from named pipes.

mojo/edk/system/channel_posix.cc

 // Copyright 2016 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/channel.h"
 
 #include <errno.h>
 #include <sys/socket.h>
 
 #include <algorithm>
 #include <deque>
 #include <limits>
 #include <memory>
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
 #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"
 
 #if !defined(OS_NACL)
 #include <sys/uio.h>
 #endif
 
 namespace mojo {
 namespace edk {
 
@@ skipping 172 matching lines @@
     DCHECK(!read_watcher_);
     DCHECK(!write_watcher_);
     for (auto handle : incoming_platform_handles_)
       handle.CloseIfNecessary();
   }
 
   void StartOnIOThread() {
     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() {
     base::AutoLock lock(write_lock_);
     WaitForWriteOnIOThreadNoLock();
   }
 
   void WaitForWriteOnIOThreadNoLock() {
     if (pending_write_)
       return;
@@ skipping 29 matching lines @@
   // base::MessageLoop::DestructionObserver:
   void WillDestroyCurrentMessageLoop() override {
     DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
     if (self_)
       ShutDownOnIOThread();
   }
 
   // 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;
     size_t buffer_capacity = 0;
     size_t total_bytes_read = 0;
     size_t bytes_read = 0;
     do {
       buffer_capacity = next_read_size;
       char* buffer = GetReadBuffer(&buffer_capacity);
       DCHECK_GT(buffer_capacity, 0u);
@@ skipping 36 matching lines @@
         reject_writes_ = write_error = true;
     }
     if (write_error)
       OnError();
   }
 
   // Attempts to write a message directly to the channel. If the full message
   // 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);
 
       ssize_t result;
       ScopedPlatformHandleVectorPtr handles = message_view.TakeHandles();
       if (handles && handles->size()) {
         iovec iov = {
           const_cast<void*>(message_view.data()),
           message_view.data_num_bytes()
@@ skipping 177 matching lines @@
 // static
 scoped_refptr<Channel> Channel::Create(
     Delegate* delegate,
     ScopedPlatformHandle platform_handle,
     scoped_refptr<base::TaskRunner> io_task_runner) {
   return new ChannelPosix(delegate, std::move(platform_handle), io_task_runner);
 }
 
 }  // namespace edk
 }  // namespace mojo