Bootstrap Mojo IPC independent of Chrome IPC

mojo/edk/system/channel_win.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 <windows.h>
 
 #include <algorithm>
 #include <deque>
@@ skipping 59 matching lines @@
                    public base::MessageLoop::DestructionObserver,
                    public base::MessageLoopForIO::IOHandler {
  public:
   ChannelWin(Delegate* delegate,
              ScopedPlatformHandle handle,
              scoped_refptr<base::TaskRunner> io_task_runner)
       : Channel(delegate),
         self_(this),
         handle_(std::move(handle)),
         io_task_runner_(io_task_runner) {
+    memset(&connect_context_, 0, sizeof(connect_context_));
+    connect_context_.handler = this;
+
     memset(&read_context_, 0, sizeof(read_context_));
     read_context_.handler = this;
 
     memset(&write_context_, 0, sizeof(write_context_));
     write_context_.handler = this;
   }
 
   void Start() override {
     io_task_runner_->PostTask(
         FROM_HERE, base::Bind(&ChannelWin::StartOnIOThread, this));
@@ skipping 44 matching lines @@
 
  private:
   // May run on any thread.
   ~ChannelWin() override {}
 
   void StartOnIOThread() {
     base::MessageLoop::current()->AddDestructionObserver(this);
     base::MessageLoopForIO::current()->RegisterIOHandler(
         handle_.get().handle, this);
 
-    // Now that we have registered our IOHandler, we can start writing.
+    // We may be starting the channel with a disconnected server pipe, e.g.,
+    // when not using PlatformChannelPair. Make sure the pipe is connected
+    // before we attempt to do any reading or writing.
+    BOOL result = ConnectNamedPipe(handle_.get().handle,
+                                   &connect_context_.overlapped);
+    if (!result && (GetLastError() == ERROR_PIPE_CONNECTED ||
+                    GetLastError() == ERROR_INVALID_FUNCTION)) {
+      // If the pipe's already connected, we can start using it immediately.
+      // Note that ERROR_INVALID_FUNCTION is returned if we call this on a
+      // client pipe. If this is a client pipe we know it's already connected.
+      OnPipeConnected();
+    } else if (result || GetLastError() == ERROR_IO_PENDING) {
+      // Balanced in OnIOCompleted.
+      AddRef();
+    } else {
+      OnError();
+    }
+  }
+
+  void OnPipeConnected() {
+    // Now that we are connected we can start writing and reading.
     {
       base::AutoLock lock(write_lock_);
       if (delay_writes_) {
         delay_writes_ = false;
         WriteNextNoLock();
       }
     }
 
     // Keep this alive in case we synchronously run shutdown.
     scoped_refptr<ChannelWin> keep_alive(this);
@@ skipping 18 matching lines @@
   }
 
   // base::MessageLoop::IOHandler:
   void OnIOCompleted(base::MessageLoopForIO::IOContext* context,
                      DWORD bytes_transfered,
                      DWORD error) override {
     if (error != ERROR_SUCCESS) {
       OnError();
     } else if (context == &read_context_) {
       OnReadDone(static_cast<size_t>(bytes_transfered));
+    } else if (context == &write_context_) {
+      OnWriteDone(static_cast<size_t>(bytes_transfered));
     } else {
-      CHECK(context == &write_context_);
-      OnWriteDone(static_cast<size_t>(bytes_transfered));
+      CHECK(context == &connect_context_);
+      OnPipeConnected();
     }
     Release();  // Balancing reference taken after ReadFile / WriteFile.
   }
 
   void OnReadDone(size_t bytes_read) {
     if (bytes_read > 0) {
       size_t next_read_size = 0;
       if (OnReadComplete(bytes_read, &next_read_size)) {
         ReadMore(next_read_size);
       } else {
@@ skipping 73 matching lines @@
       return true;
     return WriteNoLock(outgoing_messages_.front());
   }
 
   // Keeps the Channel alive at least until explicit shutdown on the IO thread.
   scoped_refptr<Channel> self_;
 
   ScopedPlatformHandle handle_;
   scoped_refptr<base::TaskRunner> io_task_runner_;
 
+  base::MessageLoopForIO::IOContext connect_context_;
   base::MessageLoopForIO::IOContext read_context_;
   base::MessageLoopForIO::IOContext write_context_;
 
   // Protects |reject_writes_| and |outgoing_messages_|.
   base::Lock write_lock_;
 
   bool delay_writes_ = true;
 
   bool reject_writes_ = false;
   std::deque<MessageView> outgoing_messages_;
 
   DISALLOW_COPY_AND_ASSIGN(ChannelWin);
 };
 
 }  // namespace
 
 // static
 scoped_refptr<Channel> Channel::Create(
     Delegate* delegate,
     ScopedPlatformHandle platform_handle,
     scoped_refptr<base::TaskRunner> io_task_runner) {
   return new ChannelWin(delegate, std::move(platform_handle), io_task_runner);
 }
 
 }  // namespace edk
 }  // namespace mojo