webrtc_data_stream_adapter.cc 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. // Copyright 2015 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/webrtc_data_stream_adapter.h"
  5. #include <stdint.h>
  6. #include <utility>
  7. #include "base/bind.h"
  8. #include "base/callback.h"
  9. #include "base/callback_helpers.h"
  10. #include "base/location.h"
  11. #include "base/memory/ptr_util.h"
  12. #include "base/threading/sequenced_task_runner_handle.h"
  13. #include "net/base/net_errors.h"
  14. #include "remoting/base/compound_buffer.h"
  15. #include "remoting/protocol/message_serialization.h"
  16. namespace remoting {
  17. namespace protocol {
  18. WebrtcDataStreamAdapter::WebrtcDataStreamAdapter(
  19. rtc::scoped_refptr<webrtc::DataChannelInterface> channel)
  20. : channel_(channel.get()) {
  21. channel_->RegisterObserver(this);
  22. DCHECK_EQ(channel_->state(), webrtc::DataChannelInterface::kConnecting);
  23. }
  24. WebrtcDataStreamAdapter::~WebrtcDataStreamAdapter() {
  25. if (channel_) {
  26. channel_->UnregisterObserver();
  27. channel_->Close();
  28. // Destroy |channel_| asynchronously as it may be on stack.
  29. // TODO(dcheng): This could probably be ReleaseSoon.
  30. base::SequencedTaskRunnerHandle::Get()->PostTask(
  31. FROM_HERE,
  32. base::BindOnce([](rtc::scoped_refptr<webrtc::DataChannelInterface>) {},
  33. std::move(channel_)));
  34. }
  35. }
  36. void WebrtcDataStreamAdapter::Start(EventHandler* event_handler) {
  37. DCHECK(!event_handler_);
  38. DCHECK(event_handler);
  39. event_handler_ = event_handler;
  40. }
  41. void WebrtcDataStreamAdapter::Send(google::protobuf::MessageLite* message,
  42. base::OnceClosure done) {
  43. rtc::CopyOnWriteBuffer buffer;
  44. buffer.SetSize(message->ByteSize());
  45. message->SerializeWithCachedSizesToArray(buffer.MutableData());
  46. pending_messages_.emplace(
  47. webrtc::DataBuffer(std::move(buffer), true /* binary */),
  48. std::move(done));
  49. SendMessagesIfReady();
  50. }
  51. void WebrtcDataStreamAdapter::SendMessagesIfReady() {
  52. // We use our own send queue instead of queuing multiple messages in the
  53. // data-channel queue so we can invoke the done callback as close to the
  54. // message actually being sent as possible and avoid overrunning the data-
  55. // channel queue. There is also lower-level buffering beneath the data-channel
  56. // queue, which we do want to keep full to ensure the link is fully utilized.
  57. // Send messages to the data channel until it has to add one to its own queue.
  58. // This ensures that the lower-level buffers remain full.
  59. while (state_ == State::OPEN && channel_->buffered_amount() == 0 &&
  60. !pending_messages_.empty()) {
  61. PendingMessage message = std::move(pending_messages_.front());
  62. pending_messages_.pop();
  63. if (!channel_->Send(std::move(message.buffer))) {
  64. LOG(ERROR) << "Send failed on data channel " << channel_->label();
  65. channel_->Close();
  66. return;
  67. }
  68. if (message.done_callback) {
  69. // Invoke callback asynchronously to avoid nested calls to Send.
  70. base::SequencedTaskRunnerHandle::Get()->PostTask(
  71. FROM_HERE, std::move(message.done_callback));
  72. }
  73. }
  74. }
  75. void WebrtcDataStreamAdapter::OnStateChange() {
  76. switch (channel_->state()) {
  77. case webrtc::DataChannelInterface::kOpen:
  78. DCHECK(state_ == State::CONNECTING);
  79. state_ = State::OPEN;
  80. base::SequencedTaskRunnerHandle::Get()->PostTask(
  81. FROM_HERE, base::BindOnce(&WebrtcDataStreamAdapter::InvokeOpenEvent,
  82. weak_ptr_factory_.GetWeakPtr()));
  83. break;
  84. case webrtc::DataChannelInterface::kClosing:
  85. if (state_ != State::CLOSED) {
  86. state_ = State::CLOSED;
  87. base::SequencedTaskRunnerHandle::Get()->PostTask(
  88. FROM_HERE,
  89. base::BindOnce(&WebrtcDataStreamAdapter::InvokeClosedEvent,
  90. weak_ptr_factory_.GetWeakPtr()));
  91. }
  92. break;
  93. case webrtc::DataChannelInterface::kConnecting:
  94. case webrtc::DataChannelInterface::kClosed:
  95. break;
  96. }
  97. }
  98. void WebrtcDataStreamAdapter::OnMessage(const webrtc::DataBuffer& rtc_buffer) {
  99. if (state_ != State::OPEN) {
  100. LOG(ERROR) << "Dropping a message received when the channel is not open.";
  101. return;
  102. }
  103. std::unique_ptr<CompoundBuffer> buffer(new CompoundBuffer());
  104. buffer->AppendCopyOf(reinterpret_cast<const char*>(rtc_buffer.data.data()),
  105. rtc_buffer.data.size());
  106. buffer->Lock();
  107. base::SequencedTaskRunnerHandle::Get()->PostTask(
  108. FROM_HERE,
  109. base::BindOnce(&WebrtcDataStreamAdapter::InvokeMessageEvent,
  110. weak_ptr_factory_.GetWeakPtr(), std::move(buffer)));
  111. }
  112. void WebrtcDataStreamAdapter::OnBufferedAmountChange(uint64_t previous_amount) {
  113. // WebRTC explicitly doesn't support sending from observer callbacks, so post
  114. // a task to let the stack unwind.
  115. base::SequencedTaskRunnerHandle::Get()->PostTask(
  116. FROM_HERE, base::BindOnce(&WebrtcDataStreamAdapter::SendMessagesIfReady,
  117. weak_ptr_factory_.GetWeakPtr()));
  118. }
  119. void WebrtcDataStreamAdapter::InvokeOpenEvent() {
  120. event_handler_->OnMessagePipeOpen();
  121. }
  122. void WebrtcDataStreamAdapter::InvokeClosedEvent() {
  123. event_handler_->OnMessagePipeClosed();
  124. }
  125. void WebrtcDataStreamAdapter::InvokeMessageEvent(
  126. std::unique_ptr<CompoundBuffer> buffer) {
  127. event_handler_->OnMessageReceived(std::move(buffer));
  128. }
  129. WebrtcDataStreamAdapter::PendingMessage::PendingMessage(
  130. webrtc::DataBuffer buffer,
  131. base::OnceClosure done_callback)
  132. : buffer(std::move(buffer)), done_callback(std::move(done_callback)) {}
  133. WebrtcDataStreamAdapter::PendingMessage&
  134. WebrtcDataStreamAdapter::PendingMessage::operator=(PendingMessage&&) = default;
  135. WebrtcDataStreamAdapter::PendingMessage::PendingMessage(PendingMessage&&) =
  136. default;
  137. WebrtcDataStreamAdapter::PendingMessage::~PendingMessage() = default;
  138. } // namespace protocol
  139. } // namespace remoting