mojo/services/network/public/cpp/web_socket_read_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_read_queue.h"
   6
   7 #include "base/bind.h"
   8 #include "base/logging.h"
   9
  10 namespace mojo {
  11
  12 struct WebSocketReadQueue::Operation {
  13   uint32_t num_bytes_;
  14   base::Callback<void(const char*)> callback_;
  15 };
  16
  17 WebSocketReadQueue::WebSocketReadQueue(DataPipeConsumerHandle handle)
  18     : handle_(handle), is_busy_(false), weak_factory_(this) {
  19 }
  20
  21 WebSocketReadQueue::~WebSocketReadQueue() {
  22 }
  23
  24 void WebSocketReadQueue::Read(uint32_t num_bytes,
  25                               base::Callback<void(const char*)> callback) {
  26   Operation* op = new Operation;
  27   op->num_bytes_ = num_bytes;
  28   op->callback_ = callback;
  29   queue_.push_back(op);
  30
  31   if (is_busy_)
  32     return;
  33
  34   is_busy_ = true;
  35   TryToRead();
  36 }
  37
  38 void WebSocketReadQueue::TryToRead() {
  39   DCHECK(is_busy_);
  40   DCHECK(!queue_.empty());
  41   do {
  42     Operation* op = queue_[0];
  43     const void* buffer = NULL;
  44     uint32_t bytes_read = op->num_bytes_;
  45     MojoResult result = BeginReadDataRaw(
  46         handle_, &buffer, &bytes_read, MOJO_READ_DATA_FLAG_ALL_OR_NONE);
  47     if (result == MOJO_RESULT_SHOULD_WAIT) {
  48       Wait();
  49       return;
  50     }
  51
  52     // Ensure |op| is deleted, whether or not |this| goes away.
  53     scoped_ptr<Operation> op_deleter(op);
  54     queue_.weak_erase(queue_.begin());
  55
  56     // http://crbug.com/490193 This should run callback as well. May need to
  57     // change the callback signature.
  58     if (result != MOJO_RESULT_OK)
  59       return;
  60
  61     uint32_t num_bytes = op_deleter->num_bytes_;
  62     DCHECK_LE(num_bytes, bytes_read);
  63     DataPipeConsumerHandle handle = handle_;
  64
  65     base::WeakPtr<WebSocketReadQueue> self(weak_factory_.GetWeakPtr());
  66
  67     // This call may delete |this|. In that case, |self| will be invalidated.
  68     // It may re-enter Read() too. Because |is_busy_| is true during the whole
  69     // process, TryToRead() won't be re-entered.
  70     op->callback_.Run(static_cast<const char*>(buffer));
  71
  72     EndReadDataRaw(handle, num_bytes);
  73
  74     if (!self)
  75       return;
  76   } while (!queue_.empty());
  77   is_busy_ = false;
  78 }
  79
  80 void WebSocketReadQueue::Wait() {
  81   DCHECK(is_busy_);
  82   handle_watcher_.Start(
  83       handle_,
  84       MOJO_HANDLE_SIGNAL_READABLE,
  85       MOJO_DEADLINE_INDEFINITE,
  86       base::Bind(&WebSocketReadQueue::OnHandleReady, base::Unretained(this)));
  87 }
  88
  89 void WebSocketReadQueue::OnHandleReady(MojoResult result) {
  90   DCHECK(is_busy_);
  91   TryToRead();
  92 }
  93
  94 }  // namespace mojo