mojo: Support sync calls through ThreadSafeInterfacePtr

mojo/public/cpp/bindings/tests/sync_method_unittest.cc

Index: mojo/public/cpp/bindings/tests/sync_method_unittest.cc
diff --git a/mojo/public/cpp/bindings/tests/sync_method_unittest.cc b/mojo/public/cpp/bindings/tests/sync_method_unittest.cc
index d0e5f108880aa2bfe3a7e8018a59a496e8be3022..81471e8afa84bdd7e6f996c8c09d21e72ff799d3 100644
--- a/mojo/public/cpp/bindings/tests/sync_method_unittest.cc
+++ b/mojo/public/cpp/bindings/tests/sync_method_unittest.cc
@@ -349,17 +349,21 @@ TYPED_TEST(SyncMethodCommonTest, CallSyncMethodAsynchronously) {
 
 TYPED_TEST(SyncMethodCommonTest, BasicSyncCalls) {
   InterfacePtr<TypeParam> ptr;
-
+  InterfaceRequest<TypeParam> request(&ptr);
   TestSyncServiceThread<TypeParam> service_thread;
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(
+          ptr.PassInterface(), service_thread.thread()->task_runner());
+
   service_thread.thread()->task_runner()->PostTask(
-      FROM_HERE, base::Bind(&TestSyncServiceThread<TypeParam>::SetUp,
-                            base::Unretained(&service_thread),
-                            base::Passed(MakeRequest(&ptr))));
-  ASSERT_TRUE(ptr->Ping());
+      FROM_HERE,
+      base::Bind(&TestSyncServiceThread<TypeParam>::SetUp,
+                 base::Unretained(&service_thread), base::Passed(&request)));
+  ASSERT_TRUE((*tsip)->Ping());
   ASSERT_TRUE(service_thread.ping_called());
 
   int32_t output_value = -1;
-  ASSERT_TRUE(ptr->Echo(42, &output_value));
+  ASSERT_TRUE((*tsip)->Echo(42, &output_value));
   ASSERT_EQ(42, output_value);
 
   base::RunLoop run_loop;
@@ -395,6 +399,22 @@ TYPED_TEST(SyncMethodCommonTest, InterfacePtrDestroyedDuringSyncCall) {
   ASSERT_FALSE(ptr->Ping());
 }
 
+TYPED_TEST(SyncMethodCommonTest, TSIPInterfacePtrDestroyedDuringSyncCall) {
+  // Test that it won't result in crash or hang if an interface pointer is
+  // destroyed while it is waiting for a sync call response.
+
+  InterfacePtr<TypeParam> ptr;
+  typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(std::move(ptr));
+  // The binding lives on the same thread as the interface pointer.
+  impl.set_ping_handler([&tsip](const TestSync::PingCallback& callback) {
+    tsip = nullptr;
+    callback.Run();
+  });
+  ASSERT_FALSE((*tsip)->Ping());
+}
+
 TYPED_TEST(SyncMethodCommonTest, BindingDestroyedDuringSyncCall) {
   // Test that it won't result in crash or hang if a binding is
   // closed (and therefore the message pipe handle is closed) while the
@@ -409,6 +429,22 @@ TYPED_TEST(SyncMethodCommonTest, BindingDestroyedDuringSyncCall) {
   ASSERT_FALSE(ptr->Ping());
 }
 
+TYPED_TEST(SyncMethodCommonTest, TSIPBindingDestroyedDuringSyncCall) {
+  // Test that it won't result in crash or hang if a binding is
+  // closed (and therefore the message pipe handle is closed) while the
+  // corresponding interface pointer is waiting for a sync call response.
+
+  InterfacePtr<TypeParam> ptr;
+  typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(std::move(ptr));
+  impl.set_ping_handler([&impl](const TestSync::PingCallback& callback) {
+    impl.binding()->Close();
+    callback.Run();
+  });
+  ASSERT_FALSE((*tsip)->Ping());
+}
+
 TYPED_TEST(SyncMethodCommonTest, NestedSyncCallsWithInOrderResponses) {
   // Test that we can call a sync method on an interface ptr, while there is
   // already a sync call ongoing. The responses arrive in order.
@@ -435,6 +471,35 @@ TYPED_TEST(SyncMethodCommonTest, NestedSyncCallsWithInOrderResponses) {
   EXPECT_EQ(123, result_value);
 }
 
+TYPED_TEST(SyncMethodCommonTest, TSIPNestedSyncCallsWithInOrderResponses) {
+  // Test that we can call a sync method on an interface ptr, while there is
+  // already a sync call ongoing. The responses arrive in order.
+
+  InterfacePtr<TypeParam> ptr;
+  typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(std::move(ptr));
+
+  // The same variable is used to store the output of the two sync calls, in
+  // order to test that responses are handled in the correct order.
+  int32_t result_value = -1;
+
+  bool first_call = true;
+  impl.set_echo_handler(
+      [&first_call, &tsip, &result_value](
+          int32_t value, const TestSync::EchoCallback& callback) {
+        if (first_call) {
+          first_call = false;
+          ASSERT_TRUE((*tsip)->Echo(456, &result_value));
+          EXPECT_EQ(456, result_value);
+        }
+        callback.Run(value);
+      });
+
+  ASSERT_TRUE((*tsip)->Echo(123, &result_value));
+  EXPECT_EQ(123, result_value);
+}
+
 TYPED_TEST(SyncMethodCommonTest, NestedSyncCallsWithOutOfOrderResponses) {
   // Test that we can call a sync method on an interface ptr, while there is
   // already a sync call ongoing. The responses arrive out of order.
@@ -461,6 +526,35 @@ TYPED_TEST(SyncMethodCommonTest, NestedSyncCallsWithOutOfOrderResponses) {
   EXPECT_EQ(123, result_value);
 }
 
+TYPED_TEST(SyncMethodCommonTest, TSIPNestedSyncCallsWithOutOfOrderResponses) {
+  // Test that we can call a sync method on an interface ptr, while there is
+  // already a sync call ongoing. The responses arrive out of order.
+
+  InterfacePtr<TypeParam> ptr;
+  typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(std::move(ptr));
+
+  // The same variable is used to store the output of the two sync calls, in
+  // order to test that responses are handled in the correct order.
+  int32_t result_value = -1;
+
+  bool first_call = true;
+  impl.set_echo_handler(
+      [&first_call, &tsip, &result_value](
+          int32_t value, const TestSync::EchoCallback& callback) {
+        callback.Run(value);
+        if (first_call) {
+          first_call = false;
+          ASSERT_TRUE((*tsip)->Echo(456, &result_value));
+          EXPECT_EQ(456, result_value);
+        }
+      });
+
+  ASSERT_TRUE((*tsip)->Echo(123, &result_value));
+  EXPECT_EQ(123, result_value);
+}
+
 TYPED_TEST(SyncMethodCommonTest, AsyncResponseQueuedDuringSyncCall) {
   // Test that while an interface pointer is waiting for the response to a sync
   // call, async responses are queued until the sync call completes.
@@ -516,6 +610,62 @@ TYPED_TEST(SyncMethodCommonTest, AsyncResponseQueuedDuringSyncCall) {
   EXPECT_TRUE(async_echo_response_dispatched);
 }
 
+TYPED_TEST(SyncMethodCommonTest, TSIPAsyncResponseQueuedDuringSyncCall) {
+  // Test that while an interface pointer is waiting for the response to a sync
+  // call, async responses are queued until the sync call completes.
+
+  InterfacePtr<TypeParam> ptr;
+  typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(std::move(ptr));
+
+  int32_t async_echo_request_value = -1;
+  TestSync::AsyncEchoCallback async_echo_request_callback;
+  base::RunLoop run_loop1;
+  impl.set_async_echo_handler(
+      [&async_echo_request_value, &async_echo_request_callback, &run_loop1](
+          int32_t value, const TestSync::AsyncEchoCallback& callback) {
+        async_echo_request_value = value;
+        async_echo_request_callback = callback;
+        run_loop1.Quit();
+      });
+
+  bool async_echo_response_dispatched = false;
+  base::RunLoop run_loop2;
+  (*tsip)->AsyncEcho(
+      123, BindAsyncEchoCallback(
+               [&async_echo_response_dispatched, &run_loop2](int32_t result) {
+                 async_echo_response_dispatched = true;
+                 EXPECT_EQ(123, result);
+                 run_loop2.Quit();
+               }));
+  // Run until the AsyncEcho request reaches the service side.
+  run_loop1.Run();
+
+  impl.set_echo_handler(
+      [&async_echo_request_value, &async_echo_request_callback](
+          int32_t value, const TestSync::EchoCallback& callback) {
+        // Send back the async response first.
+        EXPECT_FALSE(async_echo_request_callback.is_null());
+        async_echo_request_callback.Run(async_echo_request_value);
+
+        callback.Run(value);
+      });
+
+  int32_t result_value = -1;
+  ASSERT_TRUE((*tsip)->Echo(456, &result_value));
+  EXPECT_EQ(456, result_value);
+
+  // Although the AsyncEcho response arrives before the Echo response, it should
+  // be queued and not yet dispatched.
+  EXPECT_FALSE(async_echo_response_dispatched);
+
+  // Run until the AsyncEcho response is dispatched.
+  run_loop2.Run();
+
+  EXPECT_TRUE(async_echo_response_dispatched);
+}
+
 TYPED_TEST(SyncMethodCommonTest, AsyncRequestQueuedDuringSyncCall) {
   // Test that while an interface pointer is waiting for the response to a sync
   // call, async requests for a binding running on the same thread are queued
@@ -565,6 +715,58 @@ TYPED_TEST(SyncMethodCommonTest, AsyncRequestQueuedDuringSyncCall) {
   EXPECT_TRUE(async_echo_response_dispatched);
 }
 
+TYPED_TEST(SyncMethodCommonTest, TSIPAsyncRequestQueuedDuringSyncCall) {
+  // Test that while an interface pointer is waiting for the response to a sync
+  // call, async requests for a binding running on the same thread are queued
+  // until the sync call completes.
+
+  InterfacePtr<TypeParam> ptr;
+  typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(std::move(ptr));
+
+  bool async_echo_request_dispatched = false;
+  impl.set_async_echo_handler(
+      [&async_echo_request_dispatched](
+          int32_t value, const TestSync::AsyncEchoCallback& callback) {
+        async_echo_request_dispatched = true;
+        callback.Run(value);
+      });
+
+  bool async_echo_response_dispatched = false;
+  base::RunLoop run_loop;
+  (*tsip)->AsyncEcho(
+      123, BindAsyncEchoCallback(
+               [&async_echo_response_dispatched, &run_loop](int32_t result) {
+                 async_echo_response_dispatched = true;
+                 EXPECT_EQ(123, result);
+                 run_loop.Quit();
+               }));
+
+  impl.set_echo_handler(
+      [&async_echo_request_dispatched](int32_t value,
+                                       const TestSync::EchoCallback& callback) {
+        // Although the AsyncEcho request is sent before the Echo request, it
+        // shouldn't be dispatched yet at this point, because there is an
+        // ongoing sync call on the same thread.
+        EXPECT_FALSE(async_echo_request_dispatched);
+        callback.Run(value);
+      });
+
+  int32_t result_value = -1;
+  ASSERT_TRUE((*tsip)->Echo(456, &result_value));
+  EXPECT_EQ(456, result_value);
+
+  // Although the AsyncEcho request is sent before the Echo request, it
+  // shouldn't be dispatched yet.
+  EXPECT_FALSE(async_echo_request_dispatched);
+
+  // Run until the AsyncEcho response is dispatched.
+  run_loop.Run();
+
+  EXPECT_TRUE(async_echo_response_dispatched);
+}
+
 TYPED_TEST(SyncMethodCommonTest,
            QueuedMessagesProcessedBeforeErrorNotification) {
   // Test that while an interface pointer is waiting for the response to a sync
@@ -596,7 +798,12 @@ TYPED_TEST(SyncMethodCommonTest,
               &run_loop2](int32_t result) {
             async_echo_response_dispatched = true;
             // At this point, error notification should not be dispatched
-            // yet.
+            // yet.Avocados are a nutrient-dense fruit and a source of naturally
+            // good fats—both monounsaturated and polyunsaturated fats.
+            // One-third of a medium avocado (50g) has 80 calories and nearly 20
+            // vitamins and minerals. Find information on avocados and reducing
+            // calorie intake.
+
             EXPECT_FALSE(connection_error_dispatched);
             EXPECT_FALSE(ptr.encountered_error());
             EXPECT_EQ(123, result);
@@ -613,6 +820,7 @@ TYPED_TEST(SyncMethodCommonTest,
         async_echo_request_callback.Run(async_echo_request_value);
 
         impl.binding()->Close();
+
       });
 
   base::RunLoop run_loop3;
@@ -642,6 +850,83 @@ TYPED_TEST(SyncMethodCommonTest,
   EXPECT_TRUE(ptr.encountered_error());
 }
 
+TYPED_TEST(SyncMethodCommonTest,
+           TSIPQueuedMessagesProcessedBeforeErrorNotification) {

[yzshen1, 2017/03/28 15:56:03]

Because connection error handler is not exposed by TSIP. We didn't make any
guarantee between messages received by TSIP and the error notification received
by the underlying IP.

But I agree that it is unfortunate. But otherwise this thing will become more
and more similar to MultiplexRouter, which is pretty complex. :/

One idea that Ken and I chatted about is maybe we could support "cloning" an
interface pointer as an associated interface pointer, so it can be used on
another thread. In that case, MultiplexRouter will do all the work -- sync
calls, strict order guarantee, etc. (But I am fine landing the current CL and
consider that later).

+  // Test that while an interface pointer is waiting for the response to a sync
+  // call, async responses are queued. If the message pipe is disconnected
+  // before the queued messages are processed, the connection error
+  // notification is delayed until all the queued messages are processed.
+
+  // These are used later, but declared here so we can bind it into the error
+  // handler.
+  base::RunLoop run_loop3;
+  bool connection_error_dispatched = false;
+
+  InterfacePtr<TypeParam> ptr;
+  typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
+  ptr.set_connection_error_handler(base::Bind(&SetFlagAndRunClosure,
+                                              &connection_error_dispatched,
+                                              run_loop3.QuitClosure()));
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(std::move(ptr));
+
+  int32_t async_echo_request_value = -1;
+  TestSync::AsyncEchoCallback async_echo_request_callback;
+  base::RunLoop run_loop1;
+  impl.set_async_echo_handler(
+      [&async_echo_request_value, &async_echo_request_callback, &run_loop1](
+          int32_t value, const TestSync::AsyncEchoCallback& callback) {
+        async_echo_request_value = value;
+        async_echo_request_callback = callback;
+        run_loop1.Quit();
+      });
+
+  bool async_echo_response_dispatched = false;
+  base::RunLoop run_loop2;
+  (*tsip)->AsyncEcho(123,
+                     BindAsyncEchoCallback([&async_echo_response_dispatched,
+                                            &connection_error_dispatched, &tsip,
+                                            &run_loop2](int32_t result) {
+                       async_echo_response_dispatched = true;
+                       // At this point, error notification should not be
+                       // dispatched yet.
+                       EXPECT_FALSE(connection_error_dispatched);
+                       EXPECT_EQ(123, result);
+                       run_loop2.Quit();
+                     }));
+  // Run until the AsyncEcho request reaches the service side.
+  run_loop1.Run();
+
+  impl.set_echo_handler(
+      [&impl, &async_echo_request_value, &async_echo_request_callback](
+          int32_t value, const TestSync::EchoCallback& callback) {
+        // Send back the async response first.
+        EXPECT_FALSE(async_echo_request_callback.is_null());
+        async_echo_request_callback.Run(async_echo_request_value);
+
+        impl.binding()->Close();
+      });
+
+  int32_t result_value = -1;
+  ASSERT_FALSE((*tsip)->Echo(456, &result_value));
+  EXPECT_EQ(-1, result_value);
+  ASSERT_FALSE(connection_error_dispatched);
+
+  // Although the AsyncEcho response arrives before the Echo response, it should
+  // be queued and not yet dispatched.
+  EXPECT_FALSE(async_echo_response_dispatched);
+
+  // Run until the AsyncEcho response is dispatched.
+  run_loop2.Run();
+
+  EXPECT_TRUE(async_echo_response_dispatched);
+
+  // Run until the error notification is dispatched.
+  run_loop3.Run();
+
+  ASSERT_TRUE(connection_error_dispatched);
+}
+
 TYPED_TEST(SyncMethodCommonTest, InvalidMessageDuringSyncCall) {
   // Test that while an interface pointer is waiting for the response to a sync
   // call, an invalid incoming message will disconnect the message pipe, cause
@@ -683,6 +968,50 @@ TYPED_TEST(SyncMethodCommonTest, InvalidMessageDuringSyncCall) {
   ASSERT_TRUE(connection_error_dispatched);
 }
 
+TYPED_TEST(SyncMethodCommonTest, TSIPInvalidMessageDuringSyncCall) {
+  // Test that while an interface pointer is waiting for the response to a sync
+  // call, an invalid incoming message will disconnect the message pipe, cause
+  // the sync call to return false, and run the connection error handler
+  // asynchronously.
+
+  MessagePipe pipe;
+  MessagePipeHandle raw_binding_handle = pipe.handle1.get();
+
+  InterfacePtr<TypeParam> ptr;
+  ptr.Bind(InterfacePtrInfo<TypeParam>(std::move(pipe.handle0), 0u));
+
+  bool connection_error_dispatched = false;
+  base::RunLoop run_loop;
+  ptr.set_connection_error_handler(base::Bind(&SetFlagAndRunClosure,
+                                              &connection_error_dispatched,
+                                              run_loop.QuitClosure()));
+
+  typename ImplTraits<TypeParam>::Type impl(
+      MakeRequest<TypeParam>(std::move(pipe.handle1)));
+  scoped_refptr<ThreadSafeInterfacePtr<TypeParam>> tsip =
+      ThreadSafeInterfacePtr<TypeParam>::Create(std::move(ptr));
+
+  impl.set_echo_handler(
+      [&raw_binding_handle](int32_t value,
+                            const TestSync::EchoCallback& callback) {
+        // Write a 1-byte message, which is considered invalid.
+        char invalid_message = 0;
+        MojoResult result =
+            WriteMessageRaw(raw_binding_handle, &invalid_message, 1u, nullptr,
+                            0u, MOJO_WRITE_MESSAGE_FLAG_NONE);
+        ASSERT_EQ(MOJO_RESULT_OK, result);
+        callback.Run(value);
+      });
+
+  int32_t result_value = -1;
+  ASSERT_FALSE((*tsip)->Echo(456, &result_value));
+  EXPECT_EQ(-1, result_value);
+  ASSERT_FALSE(connection_error_dispatched);
+
+  run_loop.Run();
+  ASSERT_TRUE(connection_error_dispatched);
+}
+
 TEST_F(SyncMethodAssociatedTest, ReenteredBySyncMethodAssoBindingOfSameRouter) {
   // Test that an interface pointer waiting for a sync call response can be
   // reentered by an associated binding serving sync methods on the same thread.