Fix WebSocket{Read,Write}Queue.

mojo/services/network/public/cpp/web_socket_read_queue.cc

 // Copyright 2014 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 "network/public/cpp/web_socket_read_queue.h"
 
 #include "base/bind.h"
+#include "third_party/mojo/src/mojo/public/cpp/environment/logging.h"
 
 namespace mojo {
 
 struct WebSocketReadQueue::Operation {
   uint32_t num_bytes_;
   base::Callback<void(const char*)> callback_;
 };
 
 WebSocketReadQueue::WebSocketReadQueue(DataPipeConsumerHandle handle)
-    : handle_(handle), is_waiting_(false) {
+    : handle_(handle), is_busy_(false), destructed_(nullptr) {
 }
 
 WebSocketReadQueue::~WebSocketReadQueue() {
+  if (destructed_)
+    *destructed_ = true;
 }
 
 void WebSocketReadQueue::Read(uint32_t num_bytes,
                               base::Callback<void(const char*)> callback) {
   Operation* op = new Operation;
   op->num_bytes_ = num_bytes;
   op->callback_ = callback;
   queue_.push_back(op);
 
-  if (!is_waiting_)
-    TryToRead();
+  if (is_busy_)
+    return;
+
+  is_busy_ = true;
+  TryToRead();
 }
 
 void WebSocketReadQueue::TryToRead() {
-  Operation* op = queue_[0];
-  const void* buffer = NULL;
-  uint32_t bytes_read = op->num_bytes_;
-  MojoResult result = BeginReadDataRaw(
-      handle_, &buffer, &bytes_read, MOJO_READ_DATA_FLAG_ALL_OR_NONE);
-  if (result == MOJO_RESULT_SHOULD_WAIT) {
-    EndReadDataRaw(handle_, bytes_read);
-    Wait();
-    return;
-  }
+  MOJO_DCHECK(is_busy_);
+  MOJO_DCHECK(!queue_.empty());
+  do {
+    Operation* op = queue_[0];
+    const void* buffer = NULL;
+    uint32_t bytes_read = op->num_bytes_;
+    MojoResult result = BeginReadDataRaw(
+        handle_, &buffer, &bytes_read, MOJO_READ_DATA_FLAG_ALL_OR_NONE);
+    if (result == MOJO_RESULT_SHOULD_WAIT) {
+      Wait();
+      return;
+    }
 
-  // Ensure |op| is deleted, whether or not |this| goes away.
-  scoped_ptr<Operation> op_deleter(op);
-  queue_.weak_erase(queue_.begin());
-  if (result != MOJO_RESULT_OK)
-    return;
-  DataPipeConsumerHandle handle = handle_;
-  op->callback_.Run(static_cast<const char*>(buffer));  // may delete |this|
-  EndReadDataRaw(handle, bytes_read);
+    // Ensure |op| is deleted, whether or not |this| goes away.
+    scoped_ptr<Operation> op_deleter(op);
+    queue_.weak_erase(queue_.begin());
+
+    // http://crbug.com/490193 This should run callback as well. May need to
+    // change the callback signature.
+    if (result != MOJO_RESULT_OK)
+      return;
+
+    uint32_t num_bytes = op_deleter->num_bytes_;
+    MOJO_DCHECK(num_bytes <= bytes_read);
+    DataPipeConsumerHandle handle = handle_;
+
+    bool destructed = false;
+    destructed_ = &destructed;

[jam, 2015/05/20 18:23:13]

checkout this pattern:
https://code.google.com/p/chromium/codesearch#chromium/src/components/html_vi...

I had used booleans like you did in the past, but the weak factory approach is a
bit cleaner.

[yzshen1, 2015/05/20 19:23:32]

I agree that the pattern is cleaner.

One issue is that this folder is supposed to be consumed by WebSocket Mojo API
clients which may not depend on the chromium repo (base/ in particular). There
were discussions before about getting rid of existing base/ dependencies from
WebSocket{Read,Write}Queue.

There is presubmit check preventing me from adding new base/ includes.

If you think it is okay, I would like to leave it as it is. What do you think?
Thanks!

+
+    // May delete |this|. In that case, |destructed| will be set to true.
+    // Because |is_busy_| is true during the whole process, even if Read() is
+    // called by the callback, TryToRead() won't be re-entered. So we don't have
+    // to worry about |destructed_| being re-written by nested calls.
+    op->callback_.Run(static_cast<const char*>(buffer));
+
+    destructed_ = nullptr;
+
+    EndReadDataRaw(handle, num_bytes);
+
+    if (destructed)
+      return;
+  } while (!queue_.empty());
+  is_busy_ = false;
 }
 
 void WebSocketReadQueue::Wait() {
-  is_waiting_ = true;
+  MOJO_DCHECK(is_busy_);
   handle_watcher_.Start(
       handle_,
       MOJO_HANDLE_SIGNAL_READABLE,
       MOJO_DEADLINE_INDEFINITE,
       base::Bind(&WebSocketReadQueue::OnHandleReady, base::Unretained(this)));
 }
 
 void WebSocketReadQueue::OnHandleReady(MojoResult result) {
-  is_waiting_ = false;
+  MOJO_DCHECK(is_busy_);
   TryToRead();
 }
 
 }  // namespace mojo