Support sync mojo IPCs to and from bindings running on a sequence.

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 acd379a0cfc2824d6a430220f3bdbec9b1606192..b4f794ee35994cf51877e5bfd484b8d4927e12e0 100644
--- a/mojo/public/cpp/bindings/tests/sync_method_unittest.cc
+++ b/mojo/public/cpp/bindings/tests/sync_method_unittest.cc
@@ -9,9 +9,13 @@
 #include "base/macros.h"
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
+#include "base/sequence_token.h"
+#include "base/test/sequenced_worker_pool_owner.h"
+#include "base/threading/sequenced_worker_pool.h"
 #include "base/threading/thread.h"
 #include "mojo/public/cpp/bindings/associated_binding.h"
 #include "mojo/public/cpp/bindings/binding.h"
+#include "mojo/public/cpp/bindings/tests/thread_per_task_sequenced_task_runner.h"
 #include "mojo/public/interfaces/bindings/tests/test_sync_methods.mojom.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
@@ -227,7 +231,9 @@ class TestSyncServiceThread {
     impl_.reset();
   }
 
-  base::Thread* thread() { return &thread_; }
+  base::SequencedTaskRunner* task_runner() {
+    return thread_.task_runner().get();
+  }
   bool ping_called() const {
     base::AutoLock locker(lock_);
     return ping_called_;
@@ -244,6 +250,52 @@ class TestSyncServiceThread {
   DISALLOW_COPY_AND_ASSIGN(TestSyncServiceThread);
 };
 
+template <typename Interface>
+class TestSyncServiceSequencedPool {
+ public:
+  TestSyncServiceSequencedPool()
+      : pool_(2, "TestSyncServiceSequencedPool"),
+        token_(base::SequencedWorkerPool::GetSequenceToken()),
+        task_runner_(pool_.pool()->GetSequencedTaskRunner(token_)),
+        ping_called_(false) {}
+
+  void SetUp(InterfaceRequest<Interface> request) {
+    CHECK(task_runner_->RunsTasksOnCurrentThread());
+    impl_.reset(new typename ImplTraits<Interface>::Type(std::move(request)));
+    impl_->set_ping_handler(
+        [this](const typename Interface::PingCallback& callback) {
+          {
+            base::AutoLock locker(lock_);
+            ping_called_ = true;
+          }
+          callback.Run();
+        });
+  }
+
+  void TearDown() {
+    CHECK(task_runner_->RunsTasksOnCurrentThread());
+    impl_.reset();
+  }
+
+  base::SequencedTaskRunner* task_runner() { return task_runner_.get(); }
+  bool ping_called() const {
+    base::AutoLock locker(lock_);
+    return ping_called_;
+  }
+
+ private:
+  base::SequencedWorkerPoolOwner pool_;
+  base::SequencedWorkerPool::SequenceToken token_;
+  scoped_refptr<base::SequencedTaskRunner> task_runner_;
+
+  std::unique_ptr<typename ImplTraits<Interface>::Type> impl_;
+
+  mutable base::Lock lock_;
+  bool ping_called_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestSyncServiceSequencedPool);
+};
+
 class SyncMethodTest : public testing::Test {
  public:
   SyncMethodTest() {}
@@ -334,11 +386,87 @@ template <typename Func>
 TestSync::AsyncEchoCallback BindAsyncEchoCallback(Func func) {
   return base::Bind(&CallAsyncEchoCallback<Func>, func);
 }
+class SequencedTaskRunnerTestBase;
+
+void RunTestOnSequencedTaskRunner(
+    std::unique_ptr<SequencedTaskRunnerTestBase> test);
+
+class SequencedTaskRunnerTestBase {
+ public:
+  virtual ~SequencedTaskRunnerTestBase() = default;
+
+  void RunTest() {
+    SetUp();
+    Run();
+  }
+
+  virtual void Run() = 0;
+
+  virtual void SetUp() {}
+  virtual void TearDown() {}
+
+ protected:
+  void Done() {
+    TearDown();
+    task_runner_->PostTask(FROM_HERE, quit_closure_);
+    delete this;
+  }
+
+  base::Closure DoneClosure() {
+    return base::Bind(&SequencedTaskRunnerTestBase::Done,
+                      base::Unretained(this));
+  }
+
+ private:
+  friend void RunTestOnSequencedTaskRunner(
+      std::unique_ptr<SequencedTaskRunnerTestBase> test);
+
+  void Init(const base::Closure& quit_closure) {
+    task_runner_ = base::SequencedTaskRunnerHandle::Get();
+    quit_closure_ = quit_closure;
+  }
+
+  scoped_refptr<base::SequencedTaskRunner> task_runner_;
+  base::Closure quit_closure_;
+};
+
+template <typename TypeParam>
+class SyncMethodCommonOnSequencedTaskRunnerCommonTest : public testing::Test {
+  base::MessageLoop message_loop_;
+};
+
+template <typename TypeParam>
+class SyncMethodCommonOnSequencedTaskRunnerTest
+    : public SequencedTaskRunnerTestBase {
+ public:
+  void SetUp() override {
+    InterfacePtr<TypeParam> ptr;
+    impl_ = base::MakeUnique<typename ImplTraits<TypeParam>::Type>(
+        MakeRequest(&ptr_));
+  }
+
+ protected:
+  InterfacePtr<TypeParam> ptr_;
+  std::unique_ptr<typename ImplTraits<TypeParam>::Type> impl_;
+};
+
+void RunTestOnSequencedTaskRunner(
+    std::unique_ptr<SequencedTaskRunnerTestBase> test) {
+  ThreadPerTaskSequencedTaskRunnerOwner task_runner_owner;
+  base::RunLoop run_loop;
+  test->Init(run_loop.QuitClosure());
+  task_runner_owner.GetSequencedTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&SequencedTaskRunnerTestBase::RunTest,
+                            base::Unretained(test.release())));
+  run_loop.Run();
+}
 
 // TestSync and TestSyncMaster exercise Router and MultiplexRouter,
 // respectively.
 using InterfaceTypes = testing::Types<TestSync, TestSyncMaster>;
 TYPED_TEST_CASE(SyncMethodCommonTest, InterfaceTypes);
+TYPED_TEST_CASE(SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+                InterfaceTypes);
 
 TYPED_TEST(SyncMethodCommonTest, CallSyncMethodAsynchronously) {
   InterfacePtr<TypeParam> ptr;
@@ -350,11 +478,49 @@ TYPED_TEST(SyncMethodCommonTest, CallSyncMethodAsynchronously) {
   run_loop.Run();
 }
 
+#define SEQUENCED_TASK_RUNNER_TYPED_TEST_NAME(fixture_name, name) \
+  fixture_name##name##_SequencedTaskRunnerTestSuffix
+
+#define SEQUENCED_TASK_RUNNER_TYPED_TEST(base_fixture, fixture_name, name) \
+  template <typename TypeParam>                                            \
+  class SEQUENCED_TASK_RUNNER_TYPED_TEST_NAME(fixture_name, name)          \
+      : public fixture_name<TypeParam> {                                   \
+    void Run() override;                                                   \
+  };                                                                       \
+  TYPED_TEST(base_fixture, name) {                                         \
+    RunTestOnSequencedTaskRunner(                                          \
+        base::MakeUnique<SEQUENCED_TASK_RUNNER_TYPED_TEST_NAME(            \
+                             fixture_name, name) < TypeParam>> ());        \
+  }                                                                        \
+  template <typename TypeParam>                                            \
+  void SEQUENCED_TASK_RUNNER_TYPED_TEST_NAME(fixture_name,                 \
+                                             name)<TypeParam>::Run()
+
+#define SEQUENCED_TASK_RUNNER_TYPED_TEST_F(base_fixture, fixture_name, name) \
+  template <typename TypeParam>                                              \
+  class SEQUENCED_TASK_RUNNER_TYPED_TEST_NAME(fixture_name, name);           \
+  TYPED_TEST(base_fixture, name) {                                           \
+    RunTestOnSequencedTaskRunner(                                            \
+        base::MakeUnique<SEQUENCED_TASK_RUNNER_TYPED_TEST_NAME(              \
+                             fixture_name, name) < TypeParam>> ());          \
+  }                                                                          \
+  template <typename TypeParam>                                              \
+  class SEQUENCED_TASK_RUNNER_TYPED_TEST_NAME(fixture_name, name)            \
+      : public fixture_name<TypeParam>
+
+SEQUENCED_TASK_RUNNER_TYPED_TEST(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    CallSyncMethodAsynchronously) {
+  this->ptr_->Echo(
+      123, base::Bind(&ExpectValueAndRunClosure, 123, this->DoneClosure()));
+}
+
 TYPED_TEST(SyncMethodCommonTest, BasicSyncCalls) {
   InterfacePtr<TypeParam> ptr;
 
   TestSyncServiceThread<TypeParam> service_thread;
-  service_thread.thread()->task_runner()->PostTask(
+  service_thread.task_runner()->PostTask(
       FROM_HERE, base::Bind(&TestSyncServiceThread<TypeParam>::SetUp,
                             base::Unretained(&service_thread),
                             base::Passed(MakeRequest(&ptr))));
@@ -366,13 +532,36 @@ TYPED_TEST(SyncMethodCommonTest, BasicSyncCalls) {
   ASSERT_EQ(42, output_value);
 
   base::RunLoop run_loop;
-  service_thread.thread()->task_runner()->PostTaskAndReply(
+  service_thread.task_runner()->PostTaskAndReply(
       FROM_HERE, base::Bind(&TestSyncServiceThread<TypeParam>::TearDown,
                             base::Unretained(&service_thread)),
       run_loop.QuitClosure());
   run_loop.Run();
 }
 
+TYPED_TEST(SyncMethodCommonTest, BasicSyncCallsOnSequencedTaskRunner) {
+  InterfacePtr<TypeParam> ptr;
+
+  TestSyncServiceSequencedPool<TypeParam> service_pool;
+  service_pool.task_runner()->PostTask(
+      FROM_HERE, base::Bind(&TestSyncServiceSequencedPool<TypeParam>::SetUp,
+                            base::Unretained(&service_pool),
+                            base::Passed(MakeRequest(&ptr))));
+  ASSERT_TRUE(ptr->Ping());
+  ASSERT_TRUE(service_pool.ping_called());
+
+  int32_t output_value = -1;
+  ASSERT_TRUE(ptr->Echo(42, &output_value));
+  ASSERT_EQ(42, output_value);
+
+  base::RunLoop run_loop;
+  service_pool.task_runner()->PostTaskAndReply(
+      FROM_HERE, base::Bind(&TestSyncServiceSequencedPool<TypeParam>::TearDown,
+                            base::Unretained(&service_pool)),
+      run_loop.QuitClosure());
+  run_loop.Run();
+}
+
 TYPED_TEST(SyncMethodCommonTest, ReenteredBySyncMethodBinding) {
   // Test that an interface pointer waiting for a sync call response can be
   // reentered by a binding serving sync methods on the same thread.
@@ -385,6 +574,19 @@ TYPED_TEST(SyncMethodCommonTest, ReenteredBySyncMethodBinding) {
   EXPECT_EQ(42, output_value);
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    ReenteredBySyncMethodBinding) {
+  // Test that an interface pointer waiting for a sync call response can be
+  // reentered by a binding serving sync methods on the same thread.
+
+  int32_t output_value = -1;
+  ASSERT_TRUE(this->ptr_->Echo(42, &output_value));
+  EXPECT_EQ(42, output_value);
+  this->Done();
+}
+
 TYPED_TEST(SyncMethodCommonTest, InterfacePtrDestroyedDuringSyncCall) {
   // 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.
@@ -398,6 +600,22 @@ TYPED_TEST(SyncMethodCommonTest, InterfacePtrDestroyedDuringSyncCall) {
   ASSERT_FALSE(ptr->Ping());
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    InterfacePtrDestroyedDuringSyncCall) {
+  // 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.
+
+  auto* ptr = &this->ptr_;
+  this->impl_->set_ping_handler([ptr](const TestSync::PingCallback& callback) {
+    ptr->reset();
+    callback.Run();
+  });
+  ASSERT_FALSE(this->ptr_->Ping());
+  this->Done();
+}
+
 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
@@ -412,6 +630,24 @@ TYPED_TEST(SyncMethodCommonTest, BindingDestroyedDuringSyncCall) {
   ASSERT_FALSE(ptr->Ping());
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    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.
+
+  auto& impl = *this->impl_;
+  this->impl_->set_ping_handler(
+      [&impl](const TestSync::PingCallback& callback) {
+        impl.binding()->Close();
+        callback.Run();
+      });
+  ASSERT_FALSE(this->ptr_->Ping());
+  this->Done();
+}
+
 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.
@@ -438,6 +674,35 @@ TYPED_TEST(SyncMethodCommonTest, NestedSyncCallsWithInOrderResponses) {
   EXPECT_EQ(123, result_value);
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    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.
+
+  // 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;
+  auto& ptr = this->ptr_;
+  auto& impl = *this->impl_;
+  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);
+  this->Done();
+}
+
 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.
@@ -464,6 +729,35 @@ TYPED_TEST(SyncMethodCommonTest, NestedSyncCallsWithOutOfOrderResponses) {
   EXPECT_EQ(123, result_value);
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    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.
+
+  // 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;
+  auto& ptr = this->ptr_;
+  auto& impl = *this->impl_;
+  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);
+  this->Done();
+}
+
 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.
@@ -519,6 +813,54 @@ TYPED_TEST(SyncMethodCommonTest, AsyncResponseQueuedDuringSyncCall) {
   EXPECT_TRUE(async_echo_response_dispatched);
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST_F(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    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.
+
+  void Run() override {
+    this->impl_->set_async_echo_handler(
+        [this](int32_t value, const TestSync::AsyncEchoCallback& callback) {
+          async_echo_request_value_ = value;
+          async_echo_request_callback_ = callback;
+          OnAsyncEchoReceived();
+        });
+
+    this->ptr_->AsyncEcho(123, BindAsyncEchoCallback([this](int32_t result) {
+                            async_echo_response_dispatched_ = true;
+                            EXPECT_EQ(123, result);
+                            EXPECT_TRUE(async_echo_response_dispatched_);
+                            this->Done();
+                          }));
+  }
+
+  // Called when the AsyncEcho request reaches the service side.
+  void OnAsyncEchoReceived() {
+    this->impl_->set_echo_handler(
+        [this](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(this->ptr_->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_);
+  }
+
+  int32_t async_echo_request_value_ = -1;
+  TestSync::AsyncEchoCallback async_echo_request_callback_;
+  bool async_echo_response_dispatched_ = false;
+};
+
 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
@@ -568,6 +910,46 @@ TYPED_TEST(SyncMethodCommonTest, AsyncRequestQueuedDuringSyncCall) {
   EXPECT_TRUE(async_echo_response_dispatched);
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST_F(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    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.
+  void Run() override {
+    this->impl_->set_async_echo_handler(
+        [this](int32_t value, const TestSync::AsyncEchoCallback& callback) {
+          async_echo_request_dispatched_ = true;
+          callback.Run(value);
+        });
+
+    this->ptr_->AsyncEcho(123, BindAsyncEchoCallback([this](int32_t result) {
+                            EXPECT_EQ(123, result);
+                            this->Done();
+                          }));
+
+    this->impl_->set_echo_handler(
+        [this](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(this->ptr_->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_);
+  }
+  bool async_echo_request_dispatched_ = false;
+};
+
 TYPED_TEST(SyncMethodCommonTest,
            QueuedMessagesProcessedBeforeErrorNotification) {
   // Test that while an interface pointer is waiting for the response to a sync
@@ -592,19 +974,17 @@ TYPED_TEST(SyncMethodCommonTest,
   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.
-            EXPECT_FALSE(connection_error_dispatched);
-            EXPECT_FALSE(ptr.encountered_error());
-            EXPECT_EQ(123, result);
-            run_loop2.Quit();
-          }));
+  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.
+                   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();
 
@@ -619,9 +999,9 @@ TYPED_TEST(SyncMethodCommonTest,
       });
 
   base::RunLoop run_loop3;
-  ptr.set_connection_error_handler(
-      base::Bind(&SetFlagAndRunClosure, &connection_error_dispatched,
-                 run_loop3.QuitClosure()));
+  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));
@@ -645,6 +1025,75 @@ TYPED_TEST(SyncMethodCommonTest,
   EXPECT_TRUE(ptr.encountered_error());
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST_F(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    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.
+
+  void Run() override {
+    this->impl_->set_async_echo_handler(
+        [this](int32_t value, const TestSync::AsyncEchoCallback& callback) {
+          async_echo_request_value_ = value;
+          async_echo_request_callback_ = callback;
+          Run2();
+        });
+
+    this->ptr_->AsyncEcho(123, BindAsyncEchoCallback([this](int32_t result) {
+                            async_echo_response_dispatched_ = true;
+                            // At this point, error notification should not be
+                            // dispatched
+                            // yet.
+                            EXPECT_FALSE(connection_error_dispatched_);
+                            EXPECT_FALSE(this->ptr_.encountered_error());
+                            EXPECT_EQ(123, result);
+                            EXPECT_TRUE(async_echo_response_dispatched_);
+                          }));
+  }
+
+  // Invoked when the AsyncEcho request reaches the service side.
+  void Run2() {
+    this->impl_->set_echo_handler(
+        [this](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_);
+
+          this->impl_->binding()->Close();
+        });
+
+    this->ptr_.set_connection_error_handler(
+        base::Bind(&SetFlagAndRunClosure, &connection_error_dispatched_,
+                   LambdaBinder<>::BindLambda([this]() { Run4(); })));
+
+    int32_t result_value = -1;
+    ASSERT_FALSE(this->ptr_->Echo(456, &result_value));
+    EXPECT_EQ(-1, result_value);
+    ASSERT_FALSE(connection_error_dispatched_);
+    EXPECT_FALSE(this->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_);
+  }
+
+  // Invoked when the error notification is dispatched.
+  void Run4() {
+    ASSERT_TRUE(connection_error_dispatched_);
+    EXPECT_TRUE(this->ptr_.encountered_error());
+    this->Done();
+  }
+
+  int32_t async_echo_request_value_ = -1;
+  TestSync::AsyncEchoCallback async_echo_request_callback_;
+  bool async_echo_response_dispatched_ = false;
+  bool connection_error_dispatched_ = false;
+};
+
 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
@@ -686,6 +1135,49 @@ TYPED_TEST(SyncMethodCommonTest, InvalidMessageDuringSyncCall) {
   ASSERT_TRUE(connection_error_dispatched);
 }
 
+SEQUENCED_TASK_RUNNER_TYPED_TEST_F(
+    SyncMethodCommonOnSequencedTaskRunnerCommonTest,
+    SyncMethodCommonOnSequencedTaskRunnerTest,
+    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.
+
+  void Run() override {
+    MessagePipe pipe;
+
+    this->ptr_.Bind(InterfacePtrInfo<TypeParam>(std::move(pipe.handle0), 0u));
+
+    MessagePipeHandle raw_binding_handle = pipe.handle1.get();
+    this->impl_ = base::MakeUnique<typename ImplTraits<TypeParam>::Type>(
+        MakeRequest<TypeParam>(std::move(pipe.handle1)));
+
+    this->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);
+    });
+
+    this->ptr_.set_connection_error_handler(
+        LambdaBinder<>::BindLambda([this]() {
+          connection_error_dispatched_ = true;
+          this->Done();
+        }));
+
+    int32_t result_value = -1;
+    ASSERT_FALSE(this->ptr_->Echo(456, &result_value));
+    EXPECT_EQ(-1, result_value);
+    ASSERT_FALSE(connection_error_dispatched_);
+  }
+  bool connection_error_dispatched_ = false;
+};
+
 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.