Decouple Blimp transport output from BlimpConnections using MessagePort.

blimp/net/tcp_transport_unittest.cc

Index: blimp/net/tcp_transport_unittest.cc
diff --git a/blimp/net/tcp_transport_unittest.cc b/blimp/net/tcp_transport_unittest.cc
index a692fd880c9b70b0435163d6bc2b84fa470b0e09..97ead2f9cfbb49ce049e87765a5d8cc529f8e34e 100644
--- a/blimp/net/tcp_transport_unittest.cc
+++ b/blimp/net/tcp_transport_unittest.cc
@@ -2,6 +2,7 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include <algorithm>
 #include <memory>
 #include <string>
 
@@ -31,24 +32,34 @@ namespace {
 // Integration test for TCPEngineTransport and TCPClientTransport.
 class TCPTransportTest : public testing::Test {
  protected:
-  TCPTransportTest() {
-    net::IPEndPoint local_address(net::IPAddress(127, 0, 0, 1), 0);
-    engine_.reset(new TCPEngineTransport(local_address, nullptr));
+  TCPTransportTest()
+      : local_address_(net::IPAddress(127, 0, 0, 1), 0),
+        engine_(local_address_, nullptr),
+        read_buffer_(new net::GrowableIOBuffer) {
+    size_t buf_size = std::max(payload_1_.size(), payload_2_.size());
+    write_buffer_ = make_scoped_refptr(
+        new net::DrainableIOBuffer(new net::IOBuffer(buf_size), buf_size));
+    read_buffer_->SetCapacity(buf_size);
   }
 
   net::IPEndPoint GetLocalEndpoint() const {
     net::IPEndPoint local_address;
-    CHECK_EQ(net::OK, engine_->GetLocalAddress(&local_address));
+    CHECK_EQ(net::OK, engine_.GetLocalAddress(&local_address));
     return local_address;
   }
 
+  std::string payload_1_ = "foo";
+  std::string payload_2_ = "bar";
   base::MessageLoopForIO message_loop_;
-  std::unique_ptr<TCPEngineTransport> engine_;
+  net::IPEndPoint local_address_;
+  TCPEngineTransport engine_;
+  scoped_refptr<net::DrainableIOBuffer> write_buffer_;
+  scoped_refptr<net::GrowableIOBuffer> read_buffer_;
 };
 
 TEST_F(TCPTransportTest, Connect) {
   net::TestCompletionCallback accept_callback;
-  engine_->Connect(accept_callback.callback());
+  engine_.Connect(accept_callback.callback());
 
   net::TestCompletionCallback connect_callback;
   TCPClientTransport client(GetLocalEndpoint(), nullptr);
@@ -56,90 +67,67 @@ TEST_F(TCPTransportTest, Connect) {
 
   EXPECT_EQ(net::OK, connect_callback.WaitForResult());
   EXPECT_EQ(net::OK, accept_callback.WaitForResult());
-  EXPECT_TRUE(engine_->TakeConnection() != nullptr);
+  EXPECT_NE(nullptr, client.TakeMessagePort());
 }
 
 TEST_F(TCPTransportTest, TwoClientConnections) {
   net::TestCompletionCallback accept_callback1;
-  engine_->Connect(accept_callback1.callback());
+  engine_.Connect(accept_callback1.callback());
 
   net::TestCompletionCallback connect_callback1;
-  TCPClientTransport client1(GetLocalEndpoint(), nullptr);
-  client1.Connect(connect_callback1.callback());
-
-  net::TestCompletionCallback connect_callback2;
-  TCPClientTransport client2(GetLocalEndpoint(), nullptr);
-  client2.Connect(connect_callback2.callback());
-
+  TCPClientTransport client(GetLocalEndpoint(), nullptr);
+  client.Connect(connect_callback1.callback());
   EXPECT_EQ(net::OK, connect_callback1.WaitForResult());
   EXPECT_EQ(net::OK, accept_callback1.WaitForResult());
-  EXPECT_TRUE(engine_->TakeConnection() != nullptr);
+  EXPECT_NE(nullptr, engine_.TakeMessagePort());
 
   net::TestCompletionCallback accept_callback2;
-  engine_->Connect(accept_callback2.callback());
+  engine_.Connect(accept_callback2.callback());
+
+  net::TestCompletionCallback connect_callback2;
+  TCPClientTransport client2(GetLocalEndpoint(), nullptr);
+  client2.Connect(connect_callback2.callback());
   EXPECT_EQ(net::OK, connect_callback2.WaitForResult());
   EXPECT_EQ(net::OK, accept_callback2.WaitForResult());
-  EXPECT_TRUE(engine_->TakeConnection() != nullptr);
+  EXPECT_NE(nullptr, engine_.TakeMessagePort());
 }
 
 TEST_F(TCPTransportTest, ExchangeMessages) {
   // Start the Engine transport and connect a client to it.
   net::TestCompletionCallback accept_callback;
-  engine_->Connect(accept_callback.callback());
+  engine_.Connect(accept_callback.callback());
   net::TestCompletionCallback client_connect_callback;
   TCPClientTransport client(GetLocalEndpoint(), nullptr);
   client.Connect(client_connect_callback.callback());
-  EXPECT_EQ(net::OK, client_connect_callback.WaitForResult());
   EXPECT_EQ(net::OK, accept_callback.WaitForResult());
+  EXPECT_EQ(net::OK, client_connect_callback.WaitForResult());
 
-  // Expect the engine to get two messages from the client, and the client to
-  // get one from the engine.
-  MockBlimpMessageProcessor engine_incoming_processor;
-  MockBlimpMessageProcessor client_incoming_processor;
-  net::CompletionCallback engine_process_message_cb;
-  std::unique_ptr<BlimpMessage> client_message1 =
-      CreateStartConnectionMessage("", 0);
-  int client_message1_size = client_message1->ByteSize();
-  std::unique_ptr<BlimpMessage> client_message2 = CreateCheckpointAckMessage(5);
-  std::unique_ptr<BlimpMessage> engine_message = CreateCheckpointAckMessage(10);
-  EXPECT_CALL(engine_incoming_processor,
-              MockableProcessMessage(EqualsProto(*client_message1), _))
-      .WillOnce(SaveArg<1>(&engine_process_message_cb));
-  EXPECT_CALL(engine_incoming_processor,
-              MockableProcessMessage(EqualsProto(*client_message2), _))
-      .Times(1);
-  EXPECT_CALL(client_incoming_processor,
-              MockableProcessMessage(EqualsProto(*engine_message), _))
-      .Times(1);
-
-  // Attach the ends of the connection to our mock message-processors.
-  std::unique_ptr<BlimpConnection> engine_connnection =
-      engine_->TakeConnection();
-  std::unique_ptr<BlimpConnection> client_connnection = client.TakeConnection();
-  engine_connnection->SetIncomingMessageProcessor(&engine_incoming_processor);
-  client_connnection->SetIncomingMessageProcessor(&client_incoming_processor);
-
-  // Client sends the first message.
-  net::TestCompletionCallback client_send_callback1;
-  client_connnection->GetOutgoingMessageProcessor()->ProcessMessage(
-      std::move(client_message1), client_send_callback1.callback());
-  EXPECT_EQ(net::OK, client_send_callback1.WaitForResult());
-
-  // Engine finishes processing the client message.
-  EXPECT_FALSE(engine_process_message_cb.is_null());
-  engine_process_message_cb.Run(client_message1_size);
-
-  // Engine sends one message.
-  net::TestCompletionCallback engine_send_callback;
-  engine_connnection->GetOutgoingMessageProcessor()->ProcessMessage(
-      std::move(engine_message), engine_send_callback.callback());
-  EXPECT_EQ(net::OK, engine_send_callback.WaitForResult());
-
-  // Client sends the second message.
-  net::TestCompletionCallback client_send_callback2;
-  client_connnection->GetOutgoingMessageProcessor()->ProcessMessage(
-      std::move(client_message2), client_send_callback2.callback());
-  EXPECT_EQ(net::OK, client_send_callback2.WaitForResult());
+  std::unique_ptr<MessagePort> engine_message_port = engine_.TakeMessagePort();
+  std::unique_ptr<MessagePort> clientmessage_port = client.TakeMessagePort();
+
+  // Engine sends payload_1_ to client.
+  net::TestCompletionCallback read_cb1;
+  net::TestCompletionCallback write_cb1;
+  memcpy(write_buffer_->data(), payload_1_.data(), payload_1_.size());
+  engine_message_port->writer()->WritePacket(write_buffer_,
+                                             write_cb1.callback());
+  clientmessage_port->reader()->ReadPacket(read_buffer_, read_cb1.callback());
+  EXPECT_EQ(payload_1_.size(), static_cast<size_t>(read_cb1.WaitForResult()));
+  EXPECT_EQ(net::OK, write_cb1.WaitForResult());
+  EXPECT_TRUE(
+      BufferStartsWith(read_buffer_.get(), payload_1_.size(), payload_1_));
+
+  // Client sends payload_2_ to engine.
+  net::TestCompletionCallback read_cb2;
+  net::TestCompletionCallback write_cb2;
+  memcpy(write_buffer_->data(), payload_2_.data(), payload_2_.size());
+  clientmessage_port->writer()->WritePacket(write_buffer_,
+                                            write_cb2.callback());
+  engine_message_port->reader()->ReadPacket(read_buffer_, read_cb2.callback());
+  EXPECT_EQ(payload_2_.size(), static_cast<size_t>(read_cb2.WaitForResult()));
+  EXPECT_EQ(net::OK, write_cb2.WaitForResult());
+  EXPECT_TRUE(
+      BufferStartsWith(read_buffer_.get(), payload_2_.size(), payload_2_));
 }
 
 }  // namespace