message_stream.cc 7.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. // Copyright 2021 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 "ash/quick_pair/message_stream/message_stream.h"
  5. #include "ash/quick_pair/common/fast_pair/fast_pair_metrics.h"
  6. #include "ash/quick_pair/common/logging.h"
  7. #include "ash/services/quick_pair/quick_pair_process.h"
  8. #include "ash/services/quick_pair/quick_pair_process_manager.h"
  9. #include "base/bind.h"
  10. #include "base/strings/string_number_conversions.h"
  11. #include "device/bluetooth/bluetooth_socket.h"
  12. #include "net/base/io_buffer.h"
  13. namespace {
  14. constexpr int kMaxBufferSize = 4096;
  15. constexpr int kMaxRetryCount = 10;
  16. constexpr int kMessageStorageCapacity = 1000;
  17. } // namespace
  18. namespace ash {
  19. namespace quick_pair {
  20. MessageStream::MessageStream(const std::string& device_address,
  21. scoped_refptr<device::BluetoothSocket> socket)
  22. : device_address_(device_address), socket_(socket) {
  23. Receive();
  24. }
  25. MessageStream::~MessageStream() {
  26. if (socket_.get())
  27. socket_->Disconnect(base::DoNothing());
  28. // Notify observers for lifetime management
  29. for (auto& obs : observers_)
  30. obs.OnMessageStreamDestroyed(device_address_);
  31. }
  32. void MessageStream::AddObserver(Observer* observer) {
  33. observers_.AddObserver(observer);
  34. }
  35. void MessageStream::RemoveObserver(Observer* observer) {
  36. observers_.RemoveObserver(observer);
  37. }
  38. void MessageStream::Receive() {
  39. if (receive_retry_counter_ == kMaxRetryCount) {
  40. QP_LOG(WARNING)
  41. << __func__
  42. << ": Failed to receive or parse data from socket more than "
  43. << kMaxRetryCount << " times.";
  44. if (socket_.get()) {
  45. socket_->Disconnect(base::BindOnce(&MessageStream::OnSocketDisconnected,
  46. weak_ptr_factory_.GetWeakPtr()));
  47. }
  48. return;
  49. }
  50. // Retry receiving data.
  51. receive_retry_counter_++;
  52. socket_->Receive(/*buffer_size=*/kMaxBufferSize,
  53. base::BindOnce(&MessageStream::ReceiveDataSuccess,
  54. weak_ptr_factory_.GetWeakPtr()),
  55. base::BindOnce(&MessageStream::ReceiveDataError,
  56. weak_ptr_factory_.GetWeakPtr()));
  57. }
  58. void MessageStream::ReceiveDataSuccess(int buffer_size,
  59. scoped_refptr<net::IOBuffer> io_buffer) {
  60. RecordMessageStreamReceiveResult(/*success=*/true);
  61. receive_retry_counter_ = 0;
  62. if (!io_buffer->data()) {
  63. Receive();
  64. return;
  65. }
  66. std::vector<uint8_t> message_bytes(buffer_size);
  67. for (int i = 0; i < buffer_size; i++) {
  68. char* c = io_buffer->data() + i;
  69. message_bytes[i] = static_cast<uint8_t>(*c);
  70. }
  71. quick_pair_process::ParseMessageStreamMessages(
  72. std::move(message_bytes),
  73. base::BindOnce(&MessageStream::ParseMessageStreamSuccess,
  74. weak_ptr_factory_.GetWeakPtr()),
  75. base::BindOnce(&MessageStream::OnUtilityProcessStopped,
  76. weak_ptr_factory_.GetWeakPtr()));
  77. }
  78. void MessageStream::ReceiveDataError(device::BluetoothSocket::ErrorReason error,
  79. const std::string& error_message) {
  80. QP_LOG(INFO) << __func__ << ": Error: " << error_message;
  81. RecordMessageStreamReceiveResult(/*success=*/false);
  82. RecordMessageStreamReceiveError(error);
  83. if (error == device::BluetoothSocket::ErrorReason::kDisconnected) {
  84. OnSocketDisconnected();
  85. return;
  86. }
  87. Receive();
  88. }
  89. void MessageStream::Disconnect(base::OnceClosure on_disconnect_callback) {
  90. // If we already have disconnected the socket, then we can run the callback.
  91. // This can happen since the socket might have disconnected previously but
  92. // we kept the MessageStream instance alive to preserve messages from the
  93. // corresponding device.
  94. if (!socket_.get()) {
  95. std::move(on_disconnect_callback).Run();
  96. return;
  97. }
  98. socket_->Disconnect(base::BindOnce(
  99. &MessageStream::OnSocketDisconnectedWithCallback,
  100. weak_ptr_factory_.GetWeakPtr(), std::move(on_disconnect_callback)));
  101. }
  102. void MessageStream::OnSocketDisconnected() {
  103. for (auto& obs : observers_)
  104. obs.OnDisconnected(device_address_);
  105. }
  106. void MessageStream::OnSocketDisconnectedWithCallback(
  107. base::OnceClosure on_disconnect_callback) {
  108. OnSocketDisconnected();
  109. std::move(on_disconnect_callback).Run();
  110. }
  111. void MessageStream::ParseMessageStreamSuccess(
  112. std::vector<mojom::MessageStreamMessagePtr> messages) {
  113. QP_LOG(VERBOSE) << __func__;
  114. if (messages.empty()) {
  115. Receive();
  116. return;
  117. }
  118. // Store messages and notify observers.
  119. for (size_t i = 0; i < messages.size(); ++i) {
  120. if (messages_.size() == kMessageStorageCapacity)
  121. messages_.pop_front();
  122. messages_.push_back(std::move(messages[i]));
  123. NotifyObservers(messages_.back());
  124. }
  125. // Attempt to receive new messages from socket.
  126. Receive();
  127. }
  128. void MessageStream::NotifyObservers(
  129. const mojom::MessageStreamMessagePtr& message) {
  130. if (message->is_model_id()) {
  131. for (auto& obs : observers_)
  132. obs.OnModelIdMessage(device_address_, message->get_model_id());
  133. return;
  134. }
  135. if (message->is_ble_address_update()) {
  136. for (auto& obs : observers_)
  137. obs.OnBleAddressUpdateMessage(device_address_,
  138. message->get_ble_address_update());
  139. return;
  140. }
  141. if (message->is_battery_update()) {
  142. for (auto& obs : observers_)
  143. obs.OnBatteryUpdateMessage(device_address_,
  144. std::move(message->get_battery_update()));
  145. return;
  146. }
  147. if (message->is_remaining_battery_time()) {
  148. for (auto& obs : observers_)
  149. obs.OnRemainingBatteryTimeMessage(device_address_,
  150. message->get_remaining_battery_time());
  151. return;
  152. }
  153. if (message->is_enable_silence_mode()) {
  154. for (auto& obs : observers_)
  155. obs.OnEnableSilenceModeMessage(device_address_,
  156. message->get_enable_silence_mode());
  157. return;
  158. }
  159. if (message->is_companion_app_log_buffer_full()) {
  160. for (auto& obs : observers_)
  161. obs.OnCompanionAppLogBufferFullMessage(device_address_);
  162. return;
  163. }
  164. if (message->is_active_components_byte()) {
  165. for (auto& obs : observers_)
  166. obs.OnActiveComponentsMessage(device_address_,
  167. message->get_active_components_byte());
  168. return;
  169. }
  170. if (message->is_ring_device_event()) {
  171. for (auto& obs : observers_)
  172. obs.OnRingDeviceMessage(device_address_,
  173. std::move(message->get_ring_device_event()));
  174. return;
  175. }
  176. if (message->is_acknowledgement()) {
  177. for (auto& obs : observers_)
  178. obs.OnAcknowledgementMessage(device_address_,
  179. std::move(message->get_acknowledgement()));
  180. return;
  181. }
  182. if (message->is_sdk_version()) {
  183. for (auto& obs : observers_)
  184. obs.OnAndroidSdkVersionMessage(device_address_,
  185. message->get_sdk_version());
  186. return;
  187. }
  188. }
  189. void MessageStream::OnUtilityProcessStopped(
  190. QuickPairProcessManager::ShutdownReason shutdown_reason) {
  191. QP_LOG(INFO) << __func__ << ": Error: " << shutdown_reason;
  192. receive_retry_counter_++;
  193. Receive();
  194. }
  195. } // namespace quick_pair
  196. } // namespace ash