Fixes bug in pausing connector with multiple messages queued up

Prior to this patch if there were multiple messages queued up and in
processing one of the queued messages the connector was paused, we
would continue processing the messages. After this patch we stop
processing if paused after dispatch.

BUG=none
TEST=covered by test
R=yzshen@chromium.org

Review URL: https://codereview.chromium.org/1565763003

Cr-Commit-Position: refs/heads/master@{#367902}

3 files changed, 44 insertions, 4 deletions

diff --git a/mojo/public/cpp/bindings/lib/connector.cc b/mojo/public/cpp/bindings/lib/connector.cc
index e6f2d83..854a80c 100644
--- a/mojo/public/cpp/bindings/lib/connector.cc
+++ b/mojo/public/cpp/bindings/lib/connector.cc
@@ -212,6 +212,7 @@ void Connector::OnHandleReady(MojoResult result) {
 
 void Connector::WaitToReadMore() {
   CHECK(!async_wait_id_);
+  CHECK(!paused_);
   async_wait_id_ = waiter_->AsyncWait(message_pipe_.get().value(),
                                       MOJO_HANDLE_SIGNAL_READABLE,
                                       MOJO_DEADLINE_INDEFINITE,
@@ -220,6 +221,8 @@ void Connector::WaitToReadMore() {
 }
 
 bool Connector::ReadSingleMessage(MojoResult* read_result) {
+  CHECK(!paused_);
+
   bool receiver_result = false;
 
   // Detect if |this| was destroyed during message dispatch. Allow for the
@@ -263,6 +266,9 @@ void Connector::ReadAllAvailableMessages() {
     if (!ReadSingleMessage(&rv))
       return;
 
+    if (paused_)
+      return;
+
     if (rv == MOJO_RESULT_SHOULD_WAIT) {
       // ReadSingleMessage could end up calling HandleError which resets
       // message_pipe_ to a dummy one that is closed. The old EDK will see the
diff --git a/mojo/public/cpp/bindings/tests/connector_unittest.cc b/mojo/public/cpp/bindings/tests/connector_unittest.cc
index 6669764..879abfa 100644
--- a/mojo/public/cpp/bindings/tests/connector_unittest.cc
+++ b/mojo/public/cpp/bindings/tests/connector_unittest.cc
@@ -24,14 +24,13 @@ namespace {
 class MessageAccumulator : public MessageReceiver {
  public:
   MessageAccumulator() {}
-  explicit MessageAccumulator(const base::Closure& closure)
-      : closure_(closure) {}
+  explicit MessageAccumulator(const Closure& closure) : closure_(closure) {}
 
   bool Accept(Message* message) override {
     queue_.Push(message);
     if (!closure_.is_null())
       closure_.Run();
-    closure_.Reset();
+    closure_.reset();
     return true;
   }
 
@@ -43,9 +42,11 @@ class MessageAccumulator : public MessageReceiver {
     closure_ = closure;
   }
 
+  size_t size() const { return queue_.size(); }
+
  private:
   MessageQueue queue_;
-  base::Closure closure_;
+  Closure closure_;
 };
 
 class ConnectorDeletingMessageAccumulator : public MessageAccumulator {
@@ -494,6 +495,37 @@ TEST_F(ConnectorTest, RaiseError) {
   EXPECT_TRUE(connector1.is_valid());
 }
 
+TEST_F(ConnectorTest, PauseWithQueuedMessages) {
+  internal::Connector connector0(std::move(handle0_),
+                                 internal::Connector::SINGLE_THREADED_SEND);
+  internal::Connector connector1(std::move(handle1_),
+                                 internal::Connector::SINGLE_THREADED_SEND);
+
+  const char kText[] = "hello world";
+
+  Message message;
+  AllocMessage(kText, &message);
+
+  // Queue up two messages.
+  connector0.Accept(&message);
+  connector0.Accept(&message);
+
+  base::RunLoop run_loop;
+  // Configure the accumulator such that it pauses after the first message is
+  // received.
+  MessageAccumulator accumulator([&connector1, &run_loop]() {
+    connector1.PauseIncomingMethodCallProcessing();
+    run_loop.Quit();
+  });
+  connector1.set_incoming_receiver(&accumulator);
+
+  run_loop.Run();
+
+  // As we paused after the first message we should only have gotten one
+  // message.
+  ASSERT_EQ(1u, accumulator.size());
+}
+
 }  // namespace
 }  // namespace test
 }  // namespace mojo
diff --git a/mojo/public/cpp/bindings/tests/message_queue.h b/mojo/public/cpp/bindings/tests/message_queue.h
index c3091db..823758d 100644
--- a/mojo/public/cpp/bindings/tests/message_queue.h
+++ b/mojo/public/cpp/bindings/tests/message_queue.h
@@ -29,6 +29,8 @@ class MessageQueue {
   // ownership of its handles to the given |message|.
   void Pop(Message* message);
 
+  size_t size() const { return queue_.size(); }
+
  private:
   void Pop();