ftl_message_reception_channel.cc 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218
  1. // Copyright 2019 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/signaling/ftl_message_reception_channel.h"
  5. #include <utility>
  6. #include "base/callback.h"
  7. #include "base/callback_helpers.h"
  8. #include "base/logging.h"
  9. #include "remoting/base/protobuf_http_status.h"
  10. #include "remoting/base/scoped_protobuf_http_request.h"
  11. #include "remoting/proto/ftl/v1/ftl_messages.pb.h"
  12. #include "remoting/signaling/ftl_services_context.h"
  13. namespace remoting {
  14. constexpr base::TimeDelta FtlMessageReceptionChannel::kPongTimeout;
  15. FtlMessageReceptionChannel::FtlMessageReceptionChannel(
  16. SignalingTracker* signaling_tracker)
  17. : reconnect_retry_backoff_(&FtlServicesContext::GetBackoffPolicy()),
  18. signaling_tracker_(signaling_tracker) {}
  19. FtlMessageReceptionChannel::~FtlMessageReceptionChannel() = default;
  20. void FtlMessageReceptionChannel::Initialize(
  21. const StreamOpener& stream_opener,
  22. const MessageCallback& on_incoming_msg) {
  23. DCHECK(stream_opener);
  24. DCHECK(on_incoming_msg);
  25. DCHECK(!stream_opener_);
  26. DCHECK(!on_incoming_msg_);
  27. stream_opener_ = stream_opener;
  28. on_incoming_msg_ = on_incoming_msg;
  29. }
  30. void FtlMessageReceptionChannel::StartReceivingMessages(
  31. base::OnceClosure on_ready,
  32. DoneCallback on_closed) {
  33. stream_closed_callbacks_.push_back(std::move(on_closed));
  34. if (state_ == State::STARTED) {
  35. std::move(on_ready).Run();
  36. return;
  37. }
  38. stream_ready_callbacks_.push_back(std::move(on_ready));
  39. if (state_ == State::STARTING) {
  40. return;
  41. }
  42. state_ = State::STARTING;
  43. RetryStartReceivingMessagesWithBackoff();
  44. }
  45. void FtlMessageReceptionChannel::StopReceivingMessages() {
  46. if (state_ == State::STOPPED) {
  47. return;
  48. }
  49. // Current stream callbacks shouldn't receive notification for future streams.
  50. stream_ready_callbacks_.clear();
  51. stream_closed_callbacks_.clear();
  52. StopReceivingMessagesInternal();
  53. }
  54. const net::BackoffEntry&
  55. FtlMessageReceptionChannel::GetReconnectRetryBackoffEntryForTesting() const {
  56. return reconnect_retry_backoff_;
  57. }
  58. void FtlMessageReceptionChannel::OnReceiveMessagesStreamReady() {
  59. DCHECK_EQ(State::STARTING, state_);
  60. state_ = State::STARTED;
  61. if (signaling_tracker_) {
  62. signaling_tracker_->OnSignalingActive();
  63. }
  64. RunStreamReadyCallbacks();
  65. BeginStreamTimers();
  66. }
  67. void FtlMessageReceptionChannel::OnReceiveMessagesStreamClosed(
  68. const ProtobufHttpStatus& status) {
  69. if (state_ == State::STOPPED) {
  70. // Previously closed by the caller.
  71. return;
  72. }
  73. if (status.ok()) {
  74. // The backend closes the stream. This is not an error so we restart it
  75. // without backoff.
  76. VLOG(1) << "Stream has been closed by the server. Reconnecting...";
  77. reconnect_retry_backoff_.Reset();
  78. RetryStartReceivingMessages();
  79. return;
  80. }
  81. reconnect_retry_backoff_.InformOfRequest(false);
  82. if (status.error_code() == ProtobufHttpStatus::Code::ABORTED ||
  83. status.error_code() == ProtobufHttpStatus::Code::UNAVAILABLE) {
  84. // These are 'soft' connection errors that should be retried.
  85. // Other errors should be ignored.
  86. RetryStartReceivingMessagesWithBackoff();
  87. return;
  88. }
  89. stream_ready_callbacks_.clear();
  90. StopReceivingMessagesInternal();
  91. RunStreamClosedCallbacks(status);
  92. }
  93. void FtlMessageReceptionChannel::OnMessageReceived(
  94. std::unique_ptr<ftl::ReceiveMessagesResponse> response) {
  95. switch (response->body_case()) {
  96. case ftl::ReceiveMessagesResponse::BodyCase::kInboxMessage: {
  97. VLOG(1) << "Received message";
  98. on_incoming_msg_.Run(response->inbox_message());
  99. break;
  100. }
  101. case ftl::ReceiveMessagesResponse::BodyCase::kPong:
  102. VLOG(1) << "Received pong";
  103. stream_pong_timer_->Reset();
  104. if (signaling_tracker_) {
  105. signaling_tracker_->OnSignalingActive();
  106. }
  107. break;
  108. case ftl::ReceiveMessagesResponse::BodyCase::kStartOfBatch:
  109. VLOG(1) << "Received start of batch";
  110. break;
  111. case ftl::ReceiveMessagesResponse::BodyCase::kEndOfBatch:
  112. VLOG(1) << "Received end of batch";
  113. break;
  114. default:
  115. LOG(WARNING) << "Received unknown message type: "
  116. << response->body_case();
  117. break;
  118. }
  119. }
  120. void FtlMessageReceptionChannel::RunStreamReadyCallbacks() {
  121. if (stream_ready_callbacks_.empty()) {
  122. return;
  123. }
  124. // The swap is to make StartReceivingMessages() reentrant.
  125. std::list<base::OnceClosure> callbacks;
  126. callbacks.swap(stream_ready_callbacks_);
  127. for (base::OnceClosure& callback : callbacks) {
  128. std::move(callback).Run();
  129. }
  130. }
  131. void FtlMessageReceptionChannel::RunStreamClosedCallbacks(
  132. const ProtobufHttpStatus& status) {
  133. if (stream_closed_callbacks_.empty()) {
  134. return;
  135. }
  136. // The swap is to make StartReceivingMessages() reentrant.
  137. std::list<DoneCallback> callbacks;
  138. callbacks.swap(stream_closed_callbacks_);
  139. for (DoneCallback& callback : callbacks) {
  140. std::move(callback).Run(status);
  141. }
  142. }
  143. void FtlMessageReceptionChannel::RetryStartReceivingMessagesWithBackoff() {
  144. VLOG(1) << "RetryStartReceivingMessages will be called with backoff: "
  145. << reconnect_retry_backoff_.GetTimeUntilRelease();
  146. reconnect_retry_timer_.Start(
  147. FROM_HERE, reconnect_retry_backoff_.GetTimeUntilRelease(),
  148. base::BindOnce(&FtlMessageReceptionChannel::RetryStartReceivingMessages,
  149. base::Unretained(this)));
  150. }
  151. void FtlMessageReceptionChannel::RetryStartReceivingMessages() {
  152. VLOG(1) << "RetryStartReceivingMessages called";
  153. StopReceivingMessagesInternal();
  154. StartReceivingMessagesInternal();
  155. }
  156. void FtlMessageReceptionChannel::StartReceivingMessagesInternal() {
  157. DCHECK_EQ(State::STOPPED, state_);
  158. state_ = State::STARTING;
  159. receive_messages_stream_ = stream_opener_.Run(
  160. base::BindOnce(&FtlMessageReceptionChannel::OnReceiveMessagesStreamReady,
  161. weak_factory_.GetWeakPtr()),
  162. base::BindRepeating(&FtlMessageReceptionChannel::OnMessageReceived,
  163. weak_factory_.GetWeakPtr()),
  164. base::BindOnce(&FtlMessageReceptionChannel::OnReceiveMessagesStreamClosed,
  165. weak_factory_.GetWeakPtr()));
  166. }
  167. void FtlMessageReceptionChannel::StopReceivingMessagesInternal() {
  168. DCHECK_NE(State::STOPPED, state_);
  169. state_ = State::STOPPED;
  170. receive_messages_stream_.reset();
  171. reconnect_retry_timer_.Stop();
  172. stream_pong_timer_.reset();
  173. }
  174. bool FtlMessageReceptionChannel::IsReceivingMessages() const {
  175. return receive_messages_stream_.get() != nullptr;
  176. }
  177. void FtlMessageReceptionChannel::BeginStreamTimers() {
  178. reconnect_retry_backoff_.Reset();
  179. stream_pong_timer_ = std::make_unique<base::DelayTimer>(
  180. FROM_HERE, kPongTimeout, this,
  181. &FtlMessageReceptionChannel::OnPongTimeout);
  182. stream_pong_timer_->Reset();
  183. }
  184. void FtlMessageReceptionChannel::OnPongTimeout() {
  185. LOG(WARNING) << "Timed out waiting for PONG message from server.";
  186. reconnect_retry_backoff_.InformOfRequest(false);
  187. RetryStartReceivingMessagesWithBackoff();
  188. }
  189. } // namespace remoting