Index: mojo/services/network/public/cpp/web_socket_write_queue.cc |
diff --git a/mojo/services/network/public/cpp/web_socket_write_queue.cc b/mojo/services/network/public/cpp/web_socket_write_queue.cc |
index 9f2d13237ef0da16533ba320cb87ea1db1dd0f67..9a0079c4609278f6b3290084ee120321477718cf 100644 |
--- a/mojo/services/network/public/cpp/web_socket_write_queue.cc |
+++ b/mojo/services/network/public/cpp/web_socket_write_queue.cc |
@@ -5,6 +5,7 @@ |
#include "network/public/cpp/web_socket_write_queue.h" |
#include "base/bind.h" |
+#include "base/logging.h" |
namespace mojo { |
@@ -19,7 +20,7 @@ struct WebSocketWriteQueue::Operation { |
}; |
WebSocketWriteQueue::WebSocketWriteQueue(DataPipeProducerHandle handle) |
- : handle_(handle), is_waiting_(false) { |
+ : handle_(handle), is_busy_(false), weak_factory_(this) { |
} |
WebSocketWriteQueue::~WebSocketWriteQueue() { |
@@ -34,41 +35,58 @@ void WebSocketWriteQueue::Write(const char* data, |
op->data_ = data; |
queue_.push_back(op); |
- MojoResult result = MOJO_RESULT_SHOULD_WAIT; |
- if (!is_waiting_) |
- result = TryToWrite(); |
+ if (!is_busy_) { |
+ is_busy_ = true; |
+ // This call may reset |is_busy_| to false. |
+ TryToWrite(); |
+ } |
- // If we have to wait, make a local copy of the data so we know it will |
- // live until we need it. |
- if (result == MOJO_RESULT_SHOULD_WAIT) { |
+ if (is_busy_) { |
+ // If we have to wait, make a local copy of the data so we know it will |
+ // live until we need it. |
op->data_copy_.resize(num_bytes); |
memcpy(&op->data_copy_[0], data, num_bytes); |
op->data_ = &op->data_copy_[0]; |
} |
} |
-MojoResult WebSocketWriteQueue::TryToWrite() { |
- Operation* op = queue_[0]; |
- uint32_t bytes_written = op->num_bytes_; |
- MojoResult result = WriteDataRaw( |
- handle_, op->data_, &bytes_written, MOJO_WRITE_DATA_FLAG_ALL_OR_NONE); |
- if (result == MOJO_RESULT_SHOULD_WAIT) { |
- Wait(); |
- return result; |
- } |
- |
- // 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 result; |
- |
- op->callback_.Run(op->data_); // may delete |this| |
- return result; |
+void WebSocketWriteQueue::TryToWrite() { |
+ DCHECK(is_busy_); |
+ DCHECK(!queue_.empty()); |
+ do { |
+ Operation* op = queue_[0]; |
+ uint32_t bytes_written = op->num_bytes_; |
+ MojoResult result = WriteDataRaw( |
+ handle_, op->data_, &bytes_written, MOJO_WRITE_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()); |
+ |
+ // http://crbug.com/490193 This should run callback as well. May need to |
+ // change the callback signature. |
+ if (result != MOJO_RESULT_OK) |
+ return; |
+ |
+ base::WeakPtr<WebSocketWriteQueue> self(weak_factory_.GetWeakPtr()); |
+ |
+ // This call may delete |this|. In that case, |self| will be invalidated. |
+ // It may re-enter Write() too. Because |is_busy_| is true during the whole |
+ // process, TryToWrite() won't be re-entered. |
+ op->callback_.Run(op->data_); |
+ |
+ if (!self) |
+ return; |
+ } while (!queue_.empty()); |
+ is_busy_ = false; |
} |
void WebSocketWriteQueue::Wait() { |
- is_waiting_ = true; |
+ DCHECK(is_busy_); |
handle_watcher_.Start(handle_, |
MOJO_HANDLE_SIGNAL_WRITABLE, |
MOJO_DEADLINE_INDEFINITE, |
@@ -77,7 +95,7 @@ void WebSocketWriteQueue::Wait() { |
} |
void WebSocketWriteQueue::OnHandleReady(MojoResult result) { |
- is_waiting_ = false; |
+ DCHECK(is_busy_); |
TryToWrite(); |
} |