mojo: Support sync calls through ThreadSafeInterfacePtr

mojo/public/cpp/bindings/tests/sync_method_unittest.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 <utility>
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/message_loop/message_loop.h"
@@ skipping 331 matching lines @@
   typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
 
   base::RunLoop run_loop;
   ptr->Echo(123, base::Bind(&ExpectValueAndRunClosure, 123,
                             run_loop.QuitClosure()));
   run_loop.Run();
 }
 
 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());
 
-  TestSyncServiceThread<TypeParam> service_thread;
   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;
   service_thread.thread()->task_runner()->PostTaskAndReply(
       FROM_HERE, base::Bind(&TestSyncServiceThread<TypeParam>::TearDown,
                             base::Unretained(&service_thread)),
       run_loop.QuitClosure());
   run_loop.Run();
 }
 
@@ skipping 15 matching lines @@
 
   InterfacePtr<TypeParam> ptr;
   typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
   impl.set_ping_handler([&ptr](const TestSync::PingCallback& callback) {
     ptr.reset();
     callback.Run();
   });
   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
   // corresponding interface pointer is waiting for a sync call response.
 
   InterfacePtr<TypeParam> ptr;
   typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
   impl.set_ping_handler([&impl](const TestSync::PingCallback& callback) {
     impl.binding()->Close();
     callback.Run();
   });
   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.
 
   InterfacePtr<TypeParam> ptr;
   typename ImplTraits<TypeParam>::Type impl(MakeRequest(&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, &ptr, &result_value](
       int32_t value, const TestSync::EchoCallback& callback) {
     if (first_call) {
       first_call = false;
       ASSERT_TRUE(ptr->Echo(456, &result_value));
       EXPECT_EQ(456, result_value);
     }
     callback.Run(value);
   });
 
   ASSERT_TRUE(ptr->Echo(123, &result_value));
   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.
 
   InterfacePtr<TypeParam> ptr;
   typename ImplTraits<TypeParam>::Type impl(MakeRequest(&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, &ptr, &result_value](
       int32_t value, const TestSync::EchoCallback& callback) {
     callback.Run(value);
     if (first_call) {
       first_call = false;
       ASSERT_TRUE(ptr->Echo(456, &result_value));
       EXPECT_EQ(456, result_value);
     }
   });
 
   ASSERT_TRUE(ptr->Echo(123, &result_value));
   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.
 
   InterfacePtr<TypeParam> ptr;
   typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
 
   int32_t async_echo_request_value = -1;
   TestSync::AsyncEchoCallback async_echo_request_callback;
   base::RunLoop run_loop1;
@@ skipping 35 matching lines @@
   // 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, 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
   // until the sync call completes.
 
   InterfacePtr<TypeParam> ptr;
   typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
 
   bool async_echo_request_dispatched = false;
   impl.set_async_echo_handler([&async_echo_request_dispatched](
@@ skipping 29 matching lines @@
   // 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, 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
   // 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.
 
   InterfacePtr<TypeParam> ptr;
   typename ImplTraits<TypeParam>::Type impl(MakeRequest(&ptr));
 
@@ skipping 11 matching lines @@
   bool async_echo_response_dispatched = false;
   bool connection_error_dispatched = false;
   base::RunLoop run_loop2;
   ptr->AsyncEcho(
       123,
       BindAsyncEchoCallback(
           [&async_echo_response_dispatched, &connection_error_dispatched, &ptr,
               &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);
             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();
+
       });
 
   base::RunLoop run_loop3;
   ptr.set_connection_error_handler(
       base::Bind(&SetFlagAndRunClosure, &connection_error_dispatched,
                  run_loop3.QuitClosure()));
 
   int32_t result_value = -1;
   ASSERT_FALSE(ptr->Echo(456, &result_value));
   EXPECT_EQ(-1, result_value);
   ASSERT_FALSE(connection_error_dispatched);
   EXPECT_FALSE(ptr.encountered_error());
 
   // 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);
   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
   // the sync call to return false, and run the connection error handler
   // asynchronously.
 
   MessagePipe pipe;
 
   InterfacePtr<TypeParam> ptr;
   ptr.Bind(InterfacePtrInfo<TypeParam>(std::move(pipe.handle0), 0u));
@@ skipping 21 matching lines @@
 
   int32_t result_value = -1;
   ASSERT_FALSE(ptr->Echo(456, &result_value));
   EXPECT_EQ(-1, result_value);
   ASSERT_FALSE(connection_error_dispatched);
 
   run_loop.Run();
   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.
   // The associated binding belongs to the same MultiplexRouter as the waiting
   // interface pointer.
 
   TestSyncAssociatedImpl opposite_asso_impl(std::move(opposite_asso_request_));
   TestSyncAssociatedPtr opposite_asso_ptr;
   opposite_asso_ptr.Bind(std::move(opposite_asso_ptr_info_));
 
@@ skipping 34 matching lines @@
   int32_t result_value = -1;
   ASSERT_TRUE(master_ptr_->Echo(456, &result_value));
   EXPECT_EQ(456, result_value);
 }
 
 // TODO(yzshen): Add more tests related to associated interfaces.
 
 }  // namespace
 }  // namespace test
 }  // namespace mojo