ftl_message_reception_channel_unittest.cc 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545
  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 <algorithm>
  6. #include <memory>
  7. #include <string>
  8. #include <utility>
  9. #include <vector>
  10. #include "base/bind.h"
  11. #include "base/callback_helpers.h"
  12. #include "base/memory/weak_ptr.h"
  13. #include "base/notreached.h"
  14. #include "base/run_loop.h"
  15. #include "base/test/bind.h"
  16. #include "base/test/mock_callback.h"
  17. #include "base/test/task_environment.h"
  18. #include "base/threading/sequenced_task_runner_handle.h"
  19. #include "base/time/time.h"
  20. #include "remoting/base/protobuf_http_status.h"
  21. #include "remoting/base/scoped_protobuf_http_request.h"
  22. #include "remoting/proto/ftl/v1/ftl_messages.pb.h"
  23. #include "remoting/signaling/ftl_services_context.h"
  24. #include "remoting/signaling/signaling_tracker.h"
  25. #include "testing/gmock/include/gmock/gmock.h"
  26. #include "testing/gtest/include/gtest/gtest.h"
  27. namespace remoting {
  28. namespace {
  29. using ::testing::_;
  30. using ::testing::Expectation;
  31. using ::testing::Invoke;
  32. using ::testing::Property;
  33. using ::testing::Return;
  34. using ReceiveMessagesResponseCallback = base::RepeatingCallback<void(
  35. std::unique_ptr<ftl::ReceiveMessagesResponse>)>;
  36. using StatusCallback = base::OnceCallback<void(const ProtobufHttpStatus&)>;
  37. class MockSignalingTracker : public SignalingTracker {
  38. public:
  39. MOCK_METHOD0(OnSignalingActive, void());
  40. };
  41. // Fake stream implementation to allow probing if a stream is closed by client.
  42. class FakeScopedProtobufHttpRequest : public ScopedProtobufHttpRequest {
  43. public:
  44. FakeScopedProtobufHttpRequest()
  45. : ScopedProtobufHttpRequest(base::DoNothing()) {}
  46. FakeScopedProtobufHttpRequest(const FakeScopedProtobufHttpRequest&) = delete;
  47. FakeScopedProtobufHttpRequest& operator=(
  48. const FakeScopedProtobufHttpRequest&) = delete;
  49. ~FakeScopedProtobufHttpRequest() override = default;
  50. base::WeakPtr<FakeScopedProtobufHttpRequest> GetWeakPtr() {
  51. return weak_factory_.GetWeakPtr();
  52. }
  53. private:
  54. base::WeakPtrFactory<FakeScopedProtobufHttpRequest> weak_factory_{this};
  55. };
  56. std::unique_ptr<FakeScopedProtobufHttpRequest> CreateFakeServerStream() {
  57. return std::make_unique<FakeScopedProtobufHttpRequest>();
  58. }
  59. // Creates a gmock EXPECT_CALL action that:
  60. // 1. Creates a fake server stream and returns it as the start stream result
  61. // 2. Posts a task to call |on_stream_opened| at the end of current sequence
  62. // 3. Writes the WeakPtr to the fake server stream to |optional_out_stream|
  63. // if it is provided.
  64. template <typename OnStreamOpenedLambda>
  65. decltype(auto) StartStream(OnStreamOpenedLambda on_stream_opened,
  66. base::WeakPtr<FakeScopedProtobufHttpRequest>*
  67. optional_out_stream = nullptr) {
  68. return [=](base::OnceClosure on_channel_ready,
  69. const ReceiveMessagesResponseCallback& on_incoming_msg,
  70. StatusCallback on_channel_closed) {
  71. auto fake_stream = CreateFakeServerStream();
  72. if (optional_out_stream) {
  73. *optional_out_stream = fake_stream->GetWeakPtr();
  74. }
  75. auto on_stream_opened_cb = base::BindLambdaForTesting(on_stream_opened);
  76. base::SequencedTaskRunnerHandle::Get()->PostTask(
  77. FROM_HERE,
  78. base::BindOnce(on_stream_opened_cb, std::move(on_channel_ready),
  79. on_incoming_msg, std::move(on_channel_closed)));
  80. return fake_stream;
  81. };
  82. }
  83. base::OnceClosure NotReachedClosure() {
  84. return base::BindOnce([]() { NOTREACHED(); });
  85. }
  86. base::RepeatingCallback<void(const ProtobufHttpStatus&)>
  87. NotReachedStatusCallback(const base::Location& location) {
  88. return base::BindLambdaForTesting([=](const ProtobufHttpStatus& status) {
  89. NOTREACHED() << "Location: " << location.ToString()
  90. << ", status code: " << static_cast<int>(status.error_code());
  91. });
  92. }
  93. base::OnceCallback<void(const ProtobufHttpStatus&)>
  94. CheckStatusThenQuitRunLoopCallback(
  95. const base::Location& from_here,
  96. ProtobufHttpStatus::Code expected_status_code,
  97. base::RunLoop* run_loop) {
  98. return base::BindLambdaForTesting([=](const ProtobufHttpStatus& status) {
  99. ASSERT_EQ(expected_status_code, status.error_code())
  100. << "Incorrect status code. Location: " << from_here.ToString();
  101. run_loop->QuitWhenIdle();
  102. });
  103. }
  104. } // namespace
  105. class FtlMessageReceptionChannelTest : public testing::Test {
  106. public:
  107. void SetUp() override;
  108. void TearDown() override;
  109. protected:
  110. base::TimeDelta GetTimeUntilRetry() const;
  111. int GetRetryFailureCount() const;
  112. base::test::TaskEnvironment task_environment_{
  113. base::test::TaskEnvironment::TimeSource::MOCK_TIME};
  114. std::unique_ptr<FtlMessageReceptionChannel> channel_;
  115. base::MockCallback<FtlMessageReceptionChannel::StreamOpener>
  116. mock_stream_opener_;
  117. base::MockCallback<base::RepeatingCallback<void(const ftl::InboxMessage&)>>
  118. mock_on_incoming_msg_;
  119. MockSignalingTracker mock_signaling_tracker_;
  120. };
  121. void FtlMessageReceptionChannelTest::SetUp() {
  122. channel_ =
  123. std::make_unique<FtlMessageReceptionChannel>(&mock_signaling_tracker_);
  124. channel_->Initialize(mock_stream_opener_.Get(), mock_on_incoming_msg_.Get());
  125. }
  126. void FtlMessageReceptionChannelTest::TearDown() {
  127. channel_.reset();
  128. task_environment_.FastForwardUntilNoTasksRemain();
  129. }
  130. base::TimeDelta FtlMessageReceptionChannelTest::GetTimeUntilRetry() const {
  131. return channel_->GetReconnectRetryBackoffEntryForTesting()
  132. .GetTimeUntilRelease();
  133. }
  134. int FtlMessageReceptionChannelTest::GetRetryFailureCount() const {
  135. return channel_->GetReconnectRetryBackoffEntryForTesting().failure_count();
  136. }
  137. TEST_F(FtlMessageReceptionChannelTest,
  138. TestStartReceivingMessages_StoppedImmediately) {
  139. base::RunLoop run_loop;
  140. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  141. .WillOnce(StartStream(
  142. [&](base::OnceClosure on_channel_ready,
  143. const ReceiveMessagesResponseCallback& on_incoming_msg,
  144. StatusCallback on_channel_closed) {
  145. channel_->StopReceivingMessages();
  146. run_loop.Quit();
  147. }));
  148. channel_->StartReceivingMessages(NotReachedClosure(),
  149. NotReachedStatusCallback(FROM_HERE));
  150. run_loop.Run();
  151. }
  152. TEST_F(FtlMessageReceptionChannelTest,
  153. TestStartReceivingMessages_NotAuthenticated) {
  154. base::RunLoop run_loop;
  155. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  156. .WillOnce(StartStream(
  157. [&](base::OnceClosure on_channel_ready,
  158. const ReceiveMessagesResponseCallback& on_incoming_msg,
  159. StatusCallback on_channel_closed) {
  160. std::move(on_channel_closed)
  161. .Run(ProtobufHttpStatus(
  162. ProtobufHttpStatus::Code::UNAUTHENTICATED, ""));
  163. }));
  164. channel_->StartReceivingMessages(
  165. NotReachedClosure(),
  166. CheckStatusThenQuitRunLoopCallback(
  167. FROM_HERE, ProtobufHttpStatus::Code::UNAUTHENTICATED, &run_loop));
  168. run_loop.Run();
  169. }
  170. TEST_F(FtlMessageReceptionChannelTest,
  171. TestStartReceivingMessages_StreamStarted) {
  172. base::RunLoop run_loop;
  173. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  174. .WillOnce(StartStream(
  175. [&](base::OnceClosure on_channel_ready,
  176. const ReceiveMessagesResponseCallback& on_incoming_msg,
  177. StatusCallback on_channel_closed) {
  178. std::move(on_channel_ready).Run();
  179. }));
  180. channel_->StartReceivingMessages(run_loop.QuitClosure(),
  181. NotReachedStatusCallback(FROM_HERE));
  182. run_loop.Run();
  183. }
  184. TEST_F(FtlMessageReceptionChannelTest,
  185. TestStartReceivingMessages_RecoverableStreamError) {
  186. base::RunLoop run_loop;
  187. base::WeakPtr<FakeScopedProtobufHttpRequest> old_stream;
  188. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  189. .WillOnce(StartStream(
  190. [&](base::OnceClosure on_channel_ready,
  191. const ReceiveMessagesResponseCallback& on_incoming_msg,
  192. StatusCallback on_channel_closed) {
  193. // The first open stream attempt fails with UNAVAILABLE error.
  194. ASSERT_EQ(0, GetRetryFailureCount());
  195. std::move(on_channel_closed)
  196. .Run(ProtobufHttpStatus(ProtobufHttpStatus::Code::UNAVAILABLE,
  197. ""));
  198. ASSERT_EQ(1, GetRetryFailureCount());
  199. ASSERT_NEAR(FtlServicesContext::kBackoffInitialDelay.InSecondsF(),
  200. GetTimeUntilRetry().InSecondsF(), 0.5);
  201. // This will make the channel reopen the stream.
  202. task_environment_.FastForwardBy(GetTimeUntilRetry());
  203. },
  204. &old_stream))
  205. .WillOnce(StartStream(
  206. [&](base::OnceClosure on_channel_ready,
  207. const ReceiveMessagesResponseCallback& on_incoming_msg,
  208. StatusCallback on_channel_closed) {
  209. // Second open stream attempt succeeds.
  210. // Assert old stream closed.
  211. ASSERT_FALSE(old_stream);
  212. std::move(on_channel_ready).Run();
  213. ASSERT_EQ(0, GetRetryFailureCount());
  214. }));
  215. channel_->StartReceivingMessages(run_loop.QuitClosure(),
  216. NotReachedStatusCallback(FROM_HERE));
  217. run_loop.Run();
  218. }
  219. TEST_F(FtlMessageReceptionChannelTest,
  220. TestStartReceivingMessages_MultipleCalls) {
  221. base::RunLoop run_loop;
  222. base::MockCallback<base::OnceClosure> stream_ready_callback;
  223. // Exits the run loop iff the callback is called three times with OK.
  224. EXPECT_CALL(stream_ready_callback, Run())
  225. .WillOnce(Return())
  226. .WillOnce(Return())
  227. .WillOnce([&]() { run_loop.Quit(); });
  228. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  229. .WillOnce(StartStream(
  230. [&](base::OnceClosure on_channel_ready,
  231. const ReceiveMessagesResponseCallback& on_incoming_msg,
  232. StatusCallback on_channel_closed) {
  233. std::move(on_channel_ready).Run();
  234. }));
  235. channel_->StartReceivingMessages(stream_ready_callback.Get(),
  236. NotReachedStatusCallback(FROM_HERE));
  237. channel_->StartReceivingMessages(stream_ready_callback.Get(),
  238. NotReachedStatusCallback(FROM_HERE));
  239. channel_->StartReceivingMessages(stream_ready_callback.Get(),
  240. NotReachedStatusCallback(FROM_HERE));
  241. run_loop.Run();
  242. }
  243. TEST_F(FtlMessageReceptionChannelTest, StreamsTwoMessages) {
  244. base::RunLoop run_loop;
  245. constexpr char kMessage1Id[] = "msg_1";
  246. constexpr char kMessage2Id[] = "msg_2";
  247. ftl::InboxMessage message_1;
  248. message_1.set_message_id(kMessage1Id);
  249. ftl::InboxMessage message_2;
  250. message_2.set_message_id(kMessage2Id);
  251. EXPECT_CALL(mock_on_incoming_msg_,
  252. Run(Property(&ftl::InboxMessage::message_id, kMessage1Id)))
  253. .WillOnce(Return());
  254. EXPECT_CALL(mock_on_incoming_msg_,
  255. Run(Property(&ftl::InboxMessage::message_id, kMessage2Id)))
  256. .WillOnce(Invoke([&](const ftl::InboxMessage&) { run_loop.Quit(); }));
  257. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  258. .WillOnce(StartStream(
  259. [&](base::OnceClosure on_channel_ready,
  260. const ReceiveMessagesResponseCallback& on_incoming_msg,
  261. StatusCallback on_channel_closed) {
  262. std::move(on_channel_ready).Run();
  263. auto response = std::make_unique<ftl::ReceiveMessagesResponse>();
  264. *response->mutable_inbox_message() = message_1;
  265. on_incoming_msg.Run(std::move(response));
  266. response = std::make_unique<ftl::ReceiveMessagesResponse>();
  267. *response->mutable_inbox_message() = message_2;
  268. on_incoming_msg.Run(std::move(response));
  269. const ProtobufHttpStatus kCancel(
  270. ProtobufHttpStatus::Code::CANCELLED, "Cancelled");
  271. std::move(on_channel_closed).Run(kCancel);
  272. }));
  273. channel_->StartReceivingMessages(
  274. base::DoNothing(),
  275. CheckStatusThenQuitRunLoopCallback(
  276. FROM_HERE, ProtobufHttpStatus::ProtobufHttpStatus::Code::CANCELLED,
  277. &run_loop));
  278. run_loop.Run();
  279. }
  280. TEST_F(FtlMessageReceptionChannelTest, ReceivedOnePong_OnSignalingActiveTwice) {
  281. base::RunLoop run_loop;
  282. base::MockCallback<base::OnceClosure> stream_ready_callback;
  283. EXPECT_CALL(mock_signaling_tracker_, OnSignalingActive())
  284. .WillOnce(Return())
  285. .WillOnce([&]() { run_loop.Quit(); });
  286. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  287. .WillOnce(StartStream(
  288. [&](base::OnceClosure on_channel_ready,
  289. const ReceiveMessagesResponseCallback& on_incoming_msg,
  290. StatusCallback on_channel_closed) {
  291. std::move(on_channel_ready).Run();
  292. auto response = std::make_unique<ftl::ReceiveMessagesResponse>();
  293. response->mutable_pong();
  294. on_incoming_msg.Run(std::move(response));
  295. }));
  296. channel_->StartReceivingMessages(base::DoNothing(),
  297. NotReachedStatusCallback(FROM_HERE));
  298. run_loop.Run();
  299. }
  300. TEST_F(FtlMessageReceptionChannelTest, NoPongWithinTimeout_ResetsStream) {
  301. base::RunLoop run_loop;
  302. base::WeakPtr<FakeScopedProtobufHttpRequest> old_stream;
  303. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  304. .WillOnce(StartStream(
  305. [&](base::OnceClosure on_channel_ready,
  306. const ReceiveMessagesResponseCallback& on_incoming_msg,
  307. StatusCallback on_channel_closed) {
  308. std::move(on_channel_ready).Run();
  309. task_environment_.FastForwardBy(
  310. FtlMessageReceptionChannel::kPongTimeout);
  311. ASSERT_EQ(1, GetRetryFailureCount());
  312. ASSERT_NEAR(FtlServicesContext::kBackoffInitialDelay.InSecondsF(),
  313. GetTimeUntilRetry().InSecondsF(), 0.5);
  314. // This will make the channel reopen the stream.
  315. task_environment_.FastForwardBy(GetTimeUntilRetry());
  316. },
  317. &old_stream))
  318. .WillOnce(StartStream(
  319. [&](base::OnceClosure on_channel_ready,
  320. const ReceiveMessagesResponseCallback& on_incoming_msg,
  321. StatusCallback on_channel_closed) {
  322. // Stream is reopened.
  323. // Assert old stream closed.
  324. ASSERT_FALSE(old_stream);
  325. std::move(on_channel_ready).Run();
  326. ASSERT_EQ(0, GetRetryFailureCount());
  327. run_loop.Quit();
  328. }));
  329. channel_->StartReceivingMessages(base::DoNothing(),
  330. NotReachedStatusCallback(FROM_HERE));
  331. run_loop.Run();
  332. }
  333. TEST_F(FtlMessageReceptionChannelTest, ServerClosesStream_ResetsStream) {
  334. base::RunLoop run_loop;
  335. base::WeakPtr<FakeScopedProtobufHttpRequest> old_stream;
  336. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  337. .WillOnce(StartStream(
  338. [&](base::OnceClosure on_channel_ready,
  339. const ReceiveMessagesResponseCallback& on_incoming_msg,
  340. StatusCallback on_channel_closed) {
  341. auto fake_server_stream = CreateFakeServerStream();
  342. std::move(on_channel_ready).Run();
  343. // Close the stream with OK.
  344. std::move(on_channel_closed).Run(ProtobufHttpStatus::OK());
  345. },
  346. &old_stream))
  347. .WillOnce(StartStream(
  348. [&](base::OnceClosure on_channel_ready,
  349. const ReceiveMessagesResponseCallback& on_incoming_msg,
  350. StatusCallback on_channel_closed) {
  351. ASSERT_FALSE(old_stream);
  352. std::move(on_channel_ready).Run();
  353. ASSERT_EQ(0, GetRetryFailureCount());
  354. run_loop.Quit();
  355. }));
  356. channel_->StartReceivingMessages(base::DoNothing(),
  357. NotReachedStatusCallback(FROM_HERE));
  358. run_loop.Run();
  359. }
  360. TEST_F(FtlMessageReceptionChannelTest, TimeoutIncreasesToMaximum) {
  361. base::RunLoop run_loop;
  362. int failure_count = 0;
  363. int hitting_max_delay_count = 0;
  364. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  365. .WillRepeatedly(StartStream(
  366. [&](base::OnceClosure on_channel_ready,
  367. const ReceiveMessagesResponseCallback& on_incoming_msg,
  368. StatusCallback on_channel_closed) {
  369. // Quit if delay is ~kBackoffMaxDelay three times.
  370. if (hitting_max_delay_count == 3) {
  371. std::move(on_channel_ready).Run();
  372. ASSERT_EQ(0, GetRetryFailureCount());
  373. run_loop.Quit();
  374. return;
  375. }
  376. // Otherwise send UNAVAILABLE to reset the stream.
  377. std::move(on_channel_closed)
  378. .Run(ProtobufHttpStatus(
  379. ProtobufHttpStatus::ProtobufHttpStatus::Code::UNAVAILABLE,
  380. ""));
  381. int new_failure_count = GetRetryFailureCount();
  382. ASSERT_LT(failure_count, new_failure_count);
  383. failure_count = new_failure_count;
  384. base::TimeDelta time_until_retry = GetTimeUntilRetry();
  385. base::TimeDelta max_delay_diff =
  386. time_until_retry - FtlServicesContext::kBackoffMaxDelay;
  387. // Adjust for fuzziness.
  388. if (max_delay_diff.magnitude() < base::Milliseconds(500)) {
  389. hitting_max_delay_count++;
  390. }
  391. // This will tail-recursively call the stream opener.
  392. task_environment_.FastForwardBy(time_until_retry);
  393. }));
  394. channel_->StartReceivingMessages(base::DoNothing(),
  395. NotReachedStatusCallback(FROM_HERE));
  396. run_loop.Run();
  397. }
  398. TEST_F(FtlMessageReceptionChannelTest,
  399. StartStreamFailsWithUnRecoverableErrorAndRetry_TimeoutApplied) {
  400. base::RunLoop run_loop;
  401. base::WeakPtr<FakeScopedProtobufHttpRequest> old_stream;
  402. EXPECT_CALL(mock_stream_opener_, Run(_, _, _))
  403. .WillOnce(StartStream(
  404. [&](base::OnceClosure on_channel_ready,
  405. const ReceiveMessagesResponseCallback& on_incoming_msg,
  406. StatusCallback on_channel_closed) {
  407. // The first open stream attempt fails with UNAUTHENTICATED error.
  408. ASSERT_EQ(0, GetRetryFailureCount());
  409. std::move(on_channel_closed)
  410. .Run(ProtobufHttpStatus(ProtobufHttpStatus::ProtobufHttpStatus::
  411. Code::UNAUTHENTICATED,
  412. ""));
  413. ASSERT_EQ(1, GetRetryFailureCount());
  414. ASSERT_NEAR(FtlServicesContext::kBackoffInitialDelay.InSecondsF(),
  415. GetTimeUntilRetry().InSecondsF(), 0.5);
  416. },
  417. &old_stream))
  418. .WillOnce(StartStream(
  419. [&](base::OnceClosure on_channel_ready,
  420. const ReceiveMessagesResponseCallback& on_incoming_msg,
  421. StatusCallback on_channel_closed) {
  422. // Second open stream attempt succeeds.
  423. // Assert old stream closed.
  424. ASSERT_FALSE(old_stream);
  425. ASSERT_EQ(1, GetRetryFailureCount());
  426. std::move(on_channel_ready).Run();
  427. ASSERT_EQ(0, GetRetryFailureCount());
  428. }));
  429. channel_->StartReceivingMessages(
  430. base::DoNothing(),
  431. base::BindLambdaForTesting([&](const ProtobufHttpStatus& status) {
  432. ASSERT_EQ(ProtobufHttpStatus::ProtobufHttpStatus::Code::UNAUTHENTICATED,
  433. status.error_code());
  434. channel_->StartReceivingMessages(run_loop.QuitClosure(),
  435. NotReachedStatusCallback(FROM_HERE));
  436. }));
  437. run_loop.Run();
  438. }
  439. } // namespace remoting