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 "net/socket/socket.h"
  16 #include "remoting/base/compound_buffer.h"
  17 #include "remoting/proto/internal.pb.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(net::Socket* socket) {
  42   DCHECK(CalledOnValidThread());
  43   DCHECK(socket);
  44   socket_ = socket;
  45   DoRead();
  46 }
  47
  48 void MessageReader::DoRead() {
  49   DCHECK(CalledOnValidThread());
  50   // Don't try to read again if there is another read pending or we
  51   // have messages that we haven't finished processing yet.
  52   while (!closed_ && !read_pending_ && pending_messages_ == 0) {
  53     read_buffer_ = new net::IOBuffer(kReadBufferSize);
  54     int result = socket_->Read(
  55         read_buffer_.get(),
  56         kReadBufferSize,
  57         base::Bind(&MessageReader::OnRead, weak_factory_.GetWeakPtr()));
  58     HandleReadResult(result);
  59   }
  60 }
  61
  62 void MessageReader::OnRead(int result) {
  63   DCHECK(CalledOnValidThread());
  64   DCHECK(read_pending_);
  65   read_pending_ = false;
  66
  67   if (!closed_) {
  68     HandleReadResult(result);
  69     DoRead();
  70   }
  71 }
  72
  73 void MessageReader::HandleReadResult(int result) {
  74   DCHECK(CalledOnValidThread());
  75   if (closed_)
  76     return;
  77
  78   if (result > 0) {
  79     OnDataReceived(read_buffer_.get(), result);
  80   } else if (result == net::ERR_IO_PENDING) {
  81     read_pending_ = true;
  82   } else {
  83     if (result != net::ERR_CONNECTION_CLOSED) {
  84       LOG(ERROR) << "Read() returned error " << result;
  85     }
  86     // Stop reading after any error.
  87     closed_ = true;
  88   }
  89 }
  90
  91 void MessageReader::OnDataReceived(net::IOBuffer* data, int data_size) {
  92   DCHECK(CalledOnValidThread());
  93   message_decoder_.AddData(data, data_size);
  94
  95   // Get list of all new messages first, and then call the callback
  96   // for all of them.
  97   while (true) {
  98     CompoundBuffer* buffer = message_decoder_.GetNextMessage();
  99     if (!buffer)
 100       break;
 101     pending_messages_++;
 102     base::ThreadTaskRunnerHandle::Get()->PostTask(
 103         FROM_HERE,
 104         base::Bind(&MessageReader::RunCallback,
 105                    weak_factory_.GetWeakPtr(),
 106                    base::Passed(make_scoped_ptr(buffer))));
 107   }
 108 }
 109
 110 void MessageReader::RunCallback(scoped_ptr<CompoundBuffer> message) {
 111   if (!message_received_callback_.is_null()){
 112     message_received_callback_.Run(
 113         message.Pass(),
 114         base::Bind(&MessageReader::OnMessageDone, weak_factory_.GetWeakPtr()));
 115   }
 116 }
 117
 118 void MessageReader::OnMessageDone() {
 119   DCHECK(CalledOnValidThread());
 120   pending_messages_--;
 121   DCHECK_GE(pending_messages_, 0);
 122
 123   // Start next read if necessary.
 124   DoRead();
 125 }
 126
 127 }  // namespace protocol
 128 }  // namespace remoting