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

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "network/public/cpp/web_socket_write_queue.h"
   6
   7 #include "base/bind.h"
   8 #include "base/logging.h"
   9
  10 namespace mojo {
  11
  12 struct WebSocketWriteQueue::Operation {
  13   uint32_t num_bytes_;
  14   base::Callback<void(const char*)> callback_;
  15
  16   const char* data_;
  17   // Only initialized if the initial Write fails. This saves a copy in
  18   // the common case.
  19   std::vector<char> data_copy_;
  20 };
  21
  22 WebSocketWriteQueue::WebSocketWriteQueue(DataPipeProducerHandle handle)
  23     : handle_(handle), is_busy_(false), weak_factory_(this) {
  24 }
  25
  26 WebSocketWriteQueue::~WebSocketWriteQueue() {
  27 }
  28
  29 void WebSocketWriteQueue::Write(const char* data,
  30                                 uint32_t num_bytes,
  31                                 base::Callback<void(const char*)> callback) {
  32   Operation* op = new Operation;
  33   op->num_bytes_ = num_bytes;
  34   op->callback_ = callback;
  35   op->data_ = data;
  36   queue_.push_back(op);
  37
  38   if (!is_busy_) {
  39     is_busy_ = true;
  40     // This call may reset |is_busy_| to false.
  41     TryToWrite();
  42   }
  43
  44   if (is_busy_) {
  45     // If we have to wait, make a local copy of the data so we know it will
  46     // live until we need it.
  47     op->data_copy_.resize(num_bytes);
  48     memcpy(&op->data_copy_[0], data, num_bytes);
  49     op->data_ = &op->data_copy_[0];
  50   }
  51 }
  52
  53 void WebSocketWriteQueue::TryToWrite() {
  54   DCHECK(is_busy_);
  55   DCHECK(!queue_.empty());
  56   do {
  57     Operation* op = queue_[0];
  58     uint32_t bytes_written = op->num_bytes_;
  59     MojoResult result = WriteDataRaw(
  60         handle_, op->data_, &bytes_written, MOJO_WRITE_DATA_FLAG_ALL_OR_NONE);
  61     if (result == MOJO_RESULT_SHOULD_WAIT) {
  62       Wait();
  63       return;
  64     }
  65
  66     // Ensure |op| is deleted, whether or not |this| goes away.
  67     scoped_ptr<Operation> op_deleter(op);
  68     queue_.weak_erase(queue_.begin());
  69
  70     // http://crbug.com/490193 This should run callback as well. May need to
  71     // change the callback signature.
  72     if (result != MOJO_RESULT_OK)
  73       return;
  74
  75     base::WeakPtr<WebSocketWriteQueue> self(weak_factory_.GetWeakPtr());
  76
  77     // This call may delete |this|. In that case, |self| will be invalidated.
  78     // It may re-enter Write() too. Because |is_busy_| is true during the whole
  79     // process, TryToWrite() won't be re-entered.
  80     op->callback_.Run(op->data_);
  81
  82     if (!self)
  83       return;
  84   } while (!queue_.empty());
  85   is_busy_ = false;
  86 }
  87
  88 void WebSocketWriteQueue::Wait() {
  89   DCHECK(is_busy_);
  90   handle_watcher_.Start(handle_,
  91                         MOJO_HANDLE_SIGNAL_WRITABLE,
  92                         MOJO_DEADLINE_INDEFINITE,
  93                         base::Bind(&WebSocketWriteQueue::OnHandleReady,
  94                                    base::Unretained(this)));
  95 }
  96
  97 void WebSocketWriteQueue::OnHandleReady(MojoResult result) {
  98   DCHECK(is_busy_);
  99   TryToWrite();
 100 }
 101
 102 }  // namespace mojo