remoting/protocol/message_reader.cc

   1 // Copyright (c) 2012 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 "remoting/protocol/message_reader.h"
   6
   7 #include "base/bind.h"
   8 #include "base/callback.h"
   9 #include "base/compiler_specific.h"
  10 #include "base/location.h"
  11 #include "base/thread_task_runner_handle.h"
  12 #include "base/single_thread_task_runner.h"
  13 #include "net/base/io_buffer.h"
  14 #include "net/base/net_errors.h"
  15 #include "remoting/base/compound_buffer.h"
  16 #include "remoting/proto/internal.pb.h"
  17 #include "remoting/protocol/p2p_stream_socket.h"
  18
  19 namespace remoting {
  20 namespace protocol {
  21
  22 static const int kReadBufferSize = 4096;
  23
  24 MessageReader::MessageReader()
  25     : socket_(nullptr),
  26       read_pending_(false),
  27       pending_messages_(0),
  28       closed_(false),
  29       weak_factory_(this) {
  30 }
  31
  32 MessageReader::~MessageReader() {
  33 }
  34
  35 void MessageReader::SetMessageReceivedCallback(
  36     const MessageReceivedCallback& callback) {
  37   DCHECK(CalledOnValidThread());
  38   message_received_callback_ = callback;
  39 }
  40
  41 void MessageReader::StartReading(
  42     P2PStreamSocket* socket,
  43     const ReadFailedCallback& read_failed_callback) {
  44   DCHECK(CalledOnValidThread());
  45   DCHECK(socket);
  46   DCHECK(!read_failed_callback.is_null());
  47
  48   socket_ = socket;
  49   read_failed_callback_ = read_failed_callback;
  50   DoRead();
  51 }
  52
  53 void MessageReader::DoRead() {
  54   DCHECK(CalledOnValidThread());
  55   // Don't try to read again if there is another read pending or we
  56   // have messages that we haven't finished processing yet.
  57   bool read_succeeded = true;
  58   while (read_succeeded && !closed_ && !read_pending_ &&
  59          pending_messages_ == 0) {
  60     read_buffer_ = new net::IOBuffer(kReadBufferSize);
  61     int result = socket_->Read(
  62         read_buffer_.get(),
  63         kReadBufferSize,
  64         base::Bind(&MessageReader::OnRead, weak_factory_.GetWeakPtr()));
  65
  66     HandleReadResult(result, &read_succeeded);
  67   }
  68 }
  69
  70 void MessageReader::OnRead(int result) {
  71   DCHECK(CalledOnValidThread());
  72   DCHECK(read_pending_);
  73   read_pending_ = false;
  74
  75   if (!closed_) {
  76     bool read_succeeded;
  77     HandleReadResult(result, &read_succeeded);
  78     if (read_succeeded)
  79       DoRead();
  80   }
  81 }
  82
  83 void MessageReader::HandleReadResult(int result, bool* read_succeeded) {
  84   DCHECK(CalledOnValidThread());
  85   if (closed_)
  86     return;
  87
  88   *read_succeeded = true;
  89
  90   if (result > 0) {
  91     OnDataReceived(read_buffer_.get(), result);
  92     *read_succeeded = true;
  93   } else if (result == net::ERR_IO_PENDING) {
  94     read_pending_ = true;
  95   } else {
  96     DCHECK_LT(result, 0);
  97
  98     // Stop reading after any error.
  99     closed_ = true;
 100     *read_succeeded = false;
 101
 102     LOG(ERROR) << "Read() returned error " << result;
 103     read_failed_callback_.Run(result);
 104   }
 105 }
 106
 107 void MessageReader::OnDataReceived(net::IOBuffer* data, int data_size) {
 108   DCHECK(CalledOnValidThread());
 109   message_decoder_.AddData(data, data_size);
 110
 111   // Get list of all new messages first, and then call the callback
 112   // for all of them.
 113   while (true) {
 114     CompoundBuffer* buffer = message_decoder_.GetNextMessage();
 115     if (!buffer)
 116       break;
 117     pending_messages_++;
 118     base::ThreadTaskRunnerHandle::Get()->PostTask(
 119         FROM_HERE,
 120         base::Bind(&MessageReader::RunCallback,
 121                    weak_factory_.GetWeakPtr(),
 122                    base::Passed(make_scoped_ptr(buffer))));
 123   }
 124 }
 125
 126 void MessageReader::RunCallback(scoped_ptr<CompoundBuffer> message) {
 127   if (!message_received_callback_.is_null()){
 128     message_received_callback_.Run(
 129         message.Pass(),
 130         base::Bind(&MessageReader::OnMessageDone, weak_factory_.GetWeakPtr()));
 131   }
 132 }
 133
 134 void MessageReader::OnMessageDone() {
 135   DCHECK(CalledOnValidThread());
 136   pending_messages_--;
 137   DCHECK_GE(pending_messages_, 0);
 138
 139   // Start next read if necessary.
 140   DoRead();
 141 }
 142
 143 }  // namespace protocol
 144 }  // namespace remoting