message_reader.cc 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120
  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. #include "remoting/protocol/message_reader.h"
  5. #include <utility>
  6. #include "base/bind.h"
  7. #include "base/callback.h"
  8. #include "base/compiler_specific.h"
  9. #include "base/location.h"
  10. #include "base/memory/ptr_util.h"
  11. #include "base/task/single_thread_task_runner.h"
  12. #include "base/threading/thread_task_runner_handle.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. namespace remoting {
  19. namespace protocol {
  20. static const int kReadBufferSize = 4096;
  21. MessageReader::MessageReader() {}
  22. MessageReader::~MessageReader() {
  23. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  24. }
  25. void MessageReader::StartReading(
  26. P2PStreamSocket* socket,
  27. const MessageReceivedCallback& message_received_callback,
  28. ReadFailedCallback read_failed_callback) {
  29. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  30. DCHECK(!socket_);
  31. DCHECK(socket);
  32. DCHECK(message_received_callback);
  33. DCHECK(read_failed_callback);
  34. socket_ = socket;
  35. message_received_callback_ = message_received_callback;
  36. read_failed_callback_ = std::move(read_failed_callback);
  37. DoRead();
  38. }
  39. void MessageReader::DoRead() {
  40. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  41. // Don't try to read again if there is another read pending or we
  42. // have messages that we haven't finished processing yet.
  43. while (!closed_ && !read_pending_) {
  44. read_buffer_ = base::MakeRefCounted<net::IOBuffer>(kReadBufferSize);
  45. int result = socket_->Read(
  46. read_buffer_.get(), kReadBufferSize,
  47. base::BindOnce(&MessageReader::OnRead, weak_factory_.GetWeakPtr()));
  48. if (!HandleReadResult(result))
  49. break;
  50. }
  51. }
  52. void MessageReader::OnRead(int result) {
  53. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  54. DCHECK(read_pending_);
  55. read_pending_ = false;
  56. if (!closed_) {
  57. HandleReadResult(result);
  58. DoRead();
  59. }
  60. }
  61. bool MessageReader::HandleReadResult(int result) {
  62. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  63. DCHECK(!closed_);
  64. if (result > 0) {
  65. OnDataReceived(read_buffer_.get(), result);
  66. return true;
  67. }
  68. if (result == net::ERR_IO_PENDING) {
  69. read_pending_ = true;
  70. return true;
  71. }
  72. // Stop reading after any error.
  73. closed_ = true;
  74. LOG(ERROR) << "Read() returned error " << result;
  75. std::move(read_failed_callback_).Run(result);
  76. // |this| may be deleted.
  77. return false;
  78. }
  79. void MessageReader::OnDataReceived(net::IOBuffer* data, int data_size) {
  80. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  81. message_decoder_.AddData(data, data_size);
  82. // Get list of all new messages first, and then call the callback
  83. // for all of them.
  84. while (true) {
  85. CompoundBuffer* buffer = message_decoder_.GetNextMessage();
  86. if (!buffer)
  87. break;
  88. base::ThreadTaskRunnerHandle::Get()->PostTask(
  89. FROM_HERE,
  90. base::BindOnce(&MessageReader::RunCallback, weak_factory_.GetWeakPtr(),
  91. base::WrapUnique(buffer)));
  92. }
  93. }
  94. void MessageReader::RunCallback(std::unique_ptr<CompoundBuffer> message) {
  95. if (message_received_callback_) {
  96. message_received_callback_.Run(std::move(message));
  97. }
  98. }
  99. } // namespace protocol
  100. } // namespace remoting