Re-land: [mojo] Use a pipe path to initialise Mojo in elevated utility processes.

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 <stdint.h>
 #include <windows.h>
 
 #include <algorithm>
@@ skipping 62 matching lines @@
                    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) {
     CHECK(handle_.is_valid());
+
+    wait_for_connect_ = handle_.get().needs_connection;
   }
 
   void Start() override {
     io_task_runner_->PostTask(
         FROM_HERE, base::Bind(&ChannelWin::StartOnIOThread, this));
   }
 
   void ShutDownImpl() override {
     // Always shut down asynchronously when called through the public interface.
     io_task_runner_->PostTask(
@@ skipping 39 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);
 
+    if (wait_for_connect_) {
+      BOOL ok = ConnectNamedPipe(handle_.get().handle,
+                                 &connect_context_.overlapped);
+      if (ok) {
+        PLOG(ERROR) << "Unexpected success while waiting for pipe connection";
+        OnError();
+        return;
+      }
+
+      const DWORD err = GetLastError();
+      switch (err) {
+        case ERROR_PIPE_CONNECTED:
+          wait_for_connect_ = false;
+          break;
+        case ERROR_IO_PENDING:
+          AddRef();
+          return;
+        case ERROR_NO_DATA:
+          OnError();
+          return;
+      }
+    }
+
     // Now that we have registered our IOHandler, we can start writing.
     {
       base::AutoLock lock(write_lock_);
       if (delay_writes_) {
         delay_writes_ = false;
         WriteNextNoLock();
       }
     }
 
     // Keep this alive in case we synchronously run shutdown.
@@ skipping 20 matching lines @@
     if (self_)
       ShutDownOnIOThread();
   }
 
   // base::MessageLoop::IOHandler:
   void OnIOCompleted(base::MessageLoopForIO::IOContext* context,
                      DWORD bytes_transfered,
                      DWORD error) override {
     if (error != ERROR_SUCCESS) {
       OnError();
+    } else if (context == &connect_context_) {
+      DCHECK(wait_for_connect_);
+      wait_for_connect_ = false;
+      ReadMore(0);
+
+      base::AutoLock lock(write_lock_);
+      if (delay_writes_) {
+        delay_writes_ = false;
+        WriteNextNoLock();
+      }
     } else if (context == &read_context_) {
       OnReadDone(static_cast<size_t>(bytes_transfered));
     } else {
       CHECK(context == &write_context_);
       OnWriteDone(static_cast<size_t>(bytes_transfered));
     }
     Release();  // Balancing reference taken after ReadFile / WriteFile.
   }
 
   void OnReadDone(size_t bytes_read) {
@@ skipping 78 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_;
 
+  bool wait_for_connect_;
+
   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