text: // 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.
  
  // TODO(vtl): Enable this on non-POSIX once we have a non-POSIX implementation.
  #include "build/build_config.h"
  #if defined(OS_POSIX)
  
  #include <stdint.h>
  
  #include <string>
  
  #include "base/basictypes.h"
  #include "base/bind.h"
  #include "base/callback.h"
  #include "base/location.h"
  #include "base/logging.h"
  #include "base/message_loop/message_loop.h"
  #include "base/threading/thread.h"
  #include "mojo/common/test/multiprocess_test_base.h"
  #include "mojo/system/channel.h"
  #include "mojo/system/local_message_pipe_endpoint.h"
  #include "mojo/system/message_pipe.h"
  #include "mojo/system/platform_channel.h"
  #include "mojo/system/proxy_message_pipe_endpoint.h"
  #include "mojo/system/test_utils.h"
  #include "mojo/system/waiter.h"
  
  namespace mojo {
  namespace system {
  namespace {
  
  class IOThreadWrapper {
   public:
    IOThreadWrapper() : io_thread_("io_thread") {}
    ~IOThreadWrapper() {
      CHECK(!channel_.get());
      CHECK(!io_thread_.IsRunning());
    }
  
    void PostTask(const tracked_objects::Location& from_here,
                  const base::Closure& task) {
      task_runner()->PostTask(from_here, task);
    }
  
    void PostTaskAndWait(const tracked_objects::Location& from_here,
                         const base::Closure& task) {
      test::PostTaskAndWait(task_runner(), from_here, task);
    }
  
    void Init(PlatformChannel* platform_channel, scoped_refptr<MessagePipe> mp) {
      io_thread_.StartWithOptions(
          base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
      PostTask(FROM_HERE,
               base::Bind(&IOThreadWrapper::InitOnIOThread,
                          base::Unretained(this),
                          platform_channel, mp));
    }
  
    void Shutdown() {
      PostTaskAndWait(FROM_HERE,
                      base::Bind(&IOThreadWrapper::ShutdownOnIOThread,
                                 base::Unretained(this)));
      io_thread_.Stop();
    }
  
    bool is_initialized() const { return !!channel_.get(); }
  
    base::MessageLoop* message_loop() {
      return io_thread_.message_loop();
    }
  
    scoped_refptr<base::TaskRunner> task_runner() {
      return message_loop()->message_loop_proxy();
    }
  
   private:
    void InitOnIOThread(PlatformChannel* platform_channel,
                        scoped_refptr<MessagePipe> mp) {
      CHECK_EQ(base::MessageLoop::current(), message_loop());
      CHECK(platform_channel);
      CHECK(platform_channel->is_valid());
  
      // Create and initialize |Channel|.
      channel_ = new Channel();
      CHECK(channel_->Init(platform_channel->PassHandle()));
  
      // Attach the message pipe endpoint.
      // Note: On the "server" (parent process) side, we need not attach the
      // message pipe endpoint immediately. However, on the "client" (child
      // process) side, this *must* be done here -- otherwise, the |Channel| may
      // receive/process messages (which it can do as soon as it's hooked up to
      // the IO thread message loop, and that message loop runs) before the
      // message pipe endpoint is attached.
      CHECK_EQ(channel_->AttachMessagePipeEndpoint(mp, 1),
               Channel::kBootstrapEndpointId);
      channel_->RunMessagePipeEndpoint(Channel::kBootstrapEndpointId,
                                       Channel::kBootstrapEndpointId);
    }
  
    void ShutdownOnIOThread() {
      CHECK(channel_.get());
      channel_->Shutdown();
      channel_ = NULL;
    }
  
    base::Thread io_thread_;
    scoped_refptr<Channel> channel_;
  
    DISALLOW_COPY_AND_ASSIGN(IOThreadWrapper);
  };
  
  class MultiprocessMessagePipeTest : public mojo::test::MultiprocessTestBase {
   public:
    MultiprocessMessagePipeTest() {}
    virtual ~MultiprocessMessagePipeTest() {}
  
    virtual void TearDown() OVERRIDE {
      if (io_thread_wrapper_.is_initialized())
        io_thread_wrapper_.Shutdown();
      mojo::test::MultiprocessTestBase::TearDown();
    }
  
    void Init(scoped_refptr<MessagePipe> mp) {
      io_thread_wrapper_.Init(platform_server_channel.get(), mp);
    }
  
   private:
    IOThreadWrapper io_thread_wrapper_;
  
    DISALLOW_COPY_AND_ASSIGN(MultiprocessMessagePipeTest);
  };
  
  MojoResult WaitIfNecessary(scoped_refptr<MessagePipe> mp, MojoWaitFlags flags) {
    Waiter waiter;
    waiter.Init();
  
    MojoResult add_result = mp->AddWaiter(0, &waiter, flags, MOJO_RESULT_OK);
    if (add_result != MOJO_RESULT_OK) {
      return (add_result == MOJO_RESULT_ALREADY_EXISTS) ? MOJO_RESULT_OK :
                                                          add_result;
    }
  
    MojoResult wait_result = waiter.Wait(MOJO_DEADLINE_INDEFINITE);
    mp->RemoveWaiter(0, &waiter);
    return wait_result;
  }
  
  // For each message received, sends a reply message with the same contents
  // repeated twice, until the other end is closed or it receives "quitquitquit"
  // (which it doesn't reply to). It'll return the number of messages received,
  // not including any "quitquitquit" message, modulo 100.
  MOJO_MULTIPROCESS_TEST_CHILD_MAIN(EchoEcho) {
    IOThreadWrapper io_thread_wrapper;
    PlatformClientChannel* const platform_client_channel =
        MultiprocessMessagePipeTest::platform_client_channel.get();
    CHECK(platform_client_channel);
    CHECK(platform_client_channel->is_valid());
    scoped_refptr<MessagePipe> mp(new MessagePipe(
        scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
        scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
    io_thread_wrapper.Init(platform_client_channel, mp);
  
    const std::string quitquitquit("quitquitquit");
    int rv = 0;
    for (;; rv = (rv + 1) % 100) {
      // Wait for our end of the message pipe to be readable.
      MojoResult result = WaitIfNecessary(mp, MOJO_WAIT_FLAG_READABLE);
      if (result != MOJO_RESULT_OK) {
        // It was closed, probably.
        CHECK_EQ(result, MOJO_RESULT_FAILED_PRECONDITION);
        break;
      }
  
      std::string read_buffer(1000, '\0');
      uint32_t read_buffer_size = static_cast<uint32_t>(read_buffer.size());
      CHECK_EQ(mp->ReadMessage(0,
                               &read_buffer[0], &read_buffer_size,
                               NULL, NULL,
                               MOJO_READ_MESSAGE_FLAG_NONE),
               MOJO_RESULT_OK);
      read_buffer.resize(read_buffer_size);
      VLOG(2) << "Child got: " << read_buffer;
  
      if (read_buffer == quitquitquit) {
        VLOG(2) << "Child quitting.";
        break;
      }
  
      std::string write_buffer = read_buffer + read_buffer;
      CHECK_EQ(mp->WriteMessage(0,
                                write_buffer.data(),
                                static_cast<uint32_t>(write_buffer.size()),
                                NULL,
                                MOJO_WRITE_MESSAGE_FLAG_NONE),
               MOJO_RESULT_OK);
    }
  
  
    mp->Close(0);
    io_thread_wrapper.Shutdown();
    return rv;
  }
  
  // Sends "hello" to child, and expects "hellohello" back.
  TEST_F(MultiprocessMessagePipeTest, Basic) {
    StartChild("EchoEcho");
  
    scoped_refptr<MessagePipe> mp(new MessagePipe(
        scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
        scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
    Init(mp);
  
    std::string hello("hello");
    EXPECT_EQ(MOJO_RESULT_OK,
              mp->WriteMessage(0,
                               hello.data(), static_cast<uint32_t>(hello.size()),
                               NULL,
                               MOJO_WRITE_MESSAGE_FLAG_NONE));
  
    EXPECT_EQ(MOJO_RESULT_OK, WaitIfNecessary(mp, MOJO_WAIT_FLAG_READABLE));
  
    std::string read_buffer(1000, '\0');
    uint32_t read_buffer_size = static_cast<uint32_t>(read_buffer.size());
    CHECK_EQ(mp->ReadMessage(0,
                             &read_buffer[0], &read_buffer_size,
                             NULL, NULL,
                             MOJO_READ_MESSAGE_FLAG_NONE),
             MOJO_RESULT_OK);
    read_buffer.resize(read_buffer_size);
    VLOG(2) << "Parent got: " << read_buffer;
    EXPECT_EQ(hello + hello, read_buffer);
  
    mp->Close(0);
  
    // We sent one message.
    EXPECT_EQ(1 % 100, WaitForChildShutdown());
  }
  
  // Sends a bunch of messages to the child. Expects them "repeated" back. Waits
  // for the child to close its end before quitting.
  TEST_F(MultiprocessMessagePipeTest, QueueMessages) {
    StartChild("EchoEcho");
  
    scoped_refptr<MessagePipe> mp(new MessagePipe(
        scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
        scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
    Init(mp);
  
    static const size_t kNumMessages = 1001;
    for (size_t i = 0; i < kNumMessages; i++) {
      std::string write_buffer(i, 'A' + (i % 26));
      EXPECT_EQ(MOJO_RESULT_OK,
                mp->WriteMessage(0,
                                 write_buffer.data(),
                                 static_cast<uint32_t>(write_buffer.size()),
                                 NULL,
                                 MOJO_WRITE_MESSAGE_FLAG_NONE));
    }
  
    const std::string quitquitquit("quitquitquit");
    EXPECT_EQ(MOJO_RESULT_OK,
              mp->WriteMessage(0,
                               quitquitquit.data(),
                               static_cast<uint32_t>(quitquitquit.size()),
                               NULL,
                               MOJO_WRITE_MESSAGE_FLAG_NONE));
  
    for (size_t i = 0; i < kNumMessages; i++) {
      EXPECT_EQ(MOJO_RESULT_OK, WaitIfNecessary(mp, MOJO_WAIT_FLAG_READABLE));
  
      std::string read_buffer(kNumMessages * 2, '\0');
      uint32_t read_buffer_size = static_cast<uint32_t>(read_buffer.size());
      CHECK_EQ(mp->ReadMessage(0,
                               &read_buffer[0], &read_buffer_size,
                               NULL, NULL,
                               MOJO_READ_MESSAGE_FLAG_NONE),
               MOJO_RESULT_OK);
      read_buffer.resize(read_buffer_size);
  
      EXPECT_EQ(std::string(i * 2, 'A' + (i % 26)), read_buffer);
    }
  
    // Wait for it to become readable, which should fail (since we sent
    // "quitquitquit").
    EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
              WaitIfNecessary(mp, MOJO_WAIT_FLAG_READABLE));
  
    mp->Close(0);
  
    EXPECT_EQ(static_cast<int>(kNumMessages % 100), WaitForChildShutdown());
  }
  
  }  // namespace
  }  // namespace system
  }  // namespace mojo
  
  #endif  // defined(OS_POSIX)
  