demuxer_stream_adapter_unittest.cc 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335
  1. // Copyright 2016 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 "media/remoting/demuxer_stream_adapter.h"
  5. #include <memory>
  6. #include <vector>
  7. #include "base/bind.h"
  8. #include "base/callback_helpers.h"
  9. #include "base/run_loop.h"
  10. #include "base/task/single_thread_task_runner.h"
  11. #include "base/test/task_environment.h"
  12. #include "components/cast_streaming/public/remoting_proto_utils.h"
  13. #include "media/base/decoder_buffer.h"
  14. #include "media/base/demuxer_stream.h"
  15. #include "media/remoting/fake_media_resource.h"
  16. #include "media/remoting/fake_remoter.h"
  17. #include "testing/gmock/include/gmock/gmock.h"
  18. #include "testing/gtest/include/gtest/gtest.h"
  19. using openscreen::cast::RpcMessenger;
  20. using testing::_;
  21. using testing::Invoke;
  22. using testing::Return;
  23. namespace media {
  24. namespace remoting {
  25. class MockDemuxerStreamAdapter {
  26. public:
  27. MockDemuxerStreamAdapter(
  28. scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
  29. scoped_refptr<base::SingleThreadTaskRunner> media_task_runner,
  30. const std::string& name,
  31. DemuxerStream* demuxer_stream,
  32. mojo::PendingRemote<mojom::RemotingDataStreamSender> stream_sender_remote,
  33. mojo::ScopedDataPipeProducerHandle producer_handle) {
  34. rpc_messenger_ = std::make_unique<RpcMessenger>(
  35. [cb =
  36. base::BindRepeating(&MockDemuxerStreamAdapter::OnSendMessageToSink,
  37. weak_factory_.GetWeakPtr())](
  38. std::vector<uint8_t> message) { cb.Run(std::move(message)); });
  39. demuxer_stream_adapter_ = std::make_unique<DemuxerStreamAdapter>(
  40. std::move(main_task_runner), std::move(media_task_runner), name,
  41. demuxer_stream, rpc_messenger_->GetWeakPtr(),
  42. rpc_messenger_->GetUniqueHandle(), std::move(stream_sender_remote),
  43. std::move(producer_handle),
  44. base::BindOnce(&MockDemuxerStreamAdapter::OnError,
  45. weak_factory_.GetWeakPtr()));
  46. // Faking initialization with random callback handle to start mojo watcher.
  47. demuxer_stream_adapter_->Initialize(3);
  48. }
  49. MockDemuxerStreamAdapter(const MockDemuxerStreamAdapter&) = delete;
  50. MockDemuxerStreamAdapter& operator=(const MockDemuxerStreamAdapter&) = delete;
  51. ~MockDemuxerStreamAdapter() {
  52. // Make sure unit tests that did not expect errors did not cause any errors.
  53. EXPECT_TRUE(errors_.empty());
  54. }
  55. int rpc_handle() const { return demuxer_stream_adapter_->rpc_handle(); }
  56. base::WeakPtr<MockDemuxerStreamAdapter> GetWeakPtr() {
  57. return weak_factory_.GetWeakPtr();
  58. }
  59. void DoDuplicateInitialize() { demuxer_stream_adapter_->Initialize(999); }
  60. void TakeErrors(std::vector<StopTrigger>* errors) {
  61. errors->swap(errors_);
  62. errors_.clear();
  63. }
  64. // Fake to signal that it's in reading state.
  65. void FakeReadUntil(int read_until_count, int callback_handle) {
  66. std::unique_ptr<openscreen::cast::RpcMessage> rpc(
  67. new openscreen::cast::RpcMessage());
  68. rpc->set_handle(rpc_handle());
  69. rpc->set_proc(openscreen::cast::RpcMessage::RPC_DS_READUNTIL);
  70. auto* read_message = rpc->mutable_demuxerstream_readuntil_rpc();
  71. read_message->set_callback_handle(
  72. callback_handle); // Given an unique callback handle.
  73. read_message->set_count(read_until_count); // Request 1 frame
  74. demuxer_stream_adapter_->OnReceivedRpc(std::move(rpc));
  75. }
  76. void OnNewBuffer(scoped_refptr<DecoderBuffer> frame) {
  77. demuxer_stream_adapter_->OnNewBuffer(DemuxerStream::kOk, frame);
  78. }
  79. void SignalFlush(bool flush) { demuxer_stream_adapter_->SignalFlush(flush); }
  80. openscreen::cast::RpcMessage* last_received_rpc() const {
  81. return last_received_rpc_.get();
  82. }
  83. private:
  84. void OnSendMessageToSink(std::vector<uint8_t> message) {
  85. last_received_rpc_ = std::make_unique<openscreen::cast::RpcMessage>();
  86. CHECK(last_received_rpc_->ParseFromArray(message.data(), message.size()));
  87. }
  88. void OnError(StopTrigger stop_trigger) { errors_.push_back(stop_trigger); }
  89. std::unique_ptr<RpcMessenger> rpc_messenger_;
  90. std::unique_ptr<DemuxerStreamAdapter> demuxer_stream_adapter_;
  91. std::unique_ptr<openscreen::cast::RpcMessage> last_received_rpc_;
  92. std::vector<StopTrigger> errors_;
  93. base::WeakPtrFactory<MockDemuxerStreamAdapter> weak_factory_{this};
  94. };
  95. class DemuxerStreamAdapterTest : public ::testing::Test {
  96. public:
  97. DemuxerStreamAdapterTest() = default;
  98. DemuxerStreamAdapterTest(const DemuxerStreamAdapterTest&) = delete;
  99. DemuxerStreamAdapterTest& operator=(const DemuxerStreamAdapterTest&) = delete;
  100. ~DemuxerStreamAdapterTest() override = default;
  101. void SetUpDataPipe() {
  102. constexpr size_t kDataPipeCapacity = 256;
  103. demuxer_stream_ = std::make_unique<FakeDemuxerStream>(true); // audio.
  104. const MojoCreateDataPipeOptions data_pipe_options{
  105. sizeof(MojoCreateDataPipeOptions), MOJO_CREATE_DATA_PIPE_FLAG_NONE, 1,
  106. kDataPipeCapacity};
  107. mojo::PendingRemote<mojom::RemotingDataStreamSender> stream_sender;
  108. mojo::ScopedDataPipeProducerHandle producer_end;
  109. mojo::ScopedDataPipeConsumerHandle consumer_end;
  110. CHECK_EQ(MOJO_RESULT_OK, mojo::CreateDataPipe(&data_pipe_options,
  111. producer_end, consumer_end));
  112. data_stream_sender_ = std::make_unique<FakeRemotingDataStreamSender>(
  113. stream_sender.InitWithNewPipeAndPassReceiver(),
  114. std::move(consumer_end));
  115. demuxer_stream_adapter_ = std::make_unique<MockDemuxerStreamAdapter>(
  116. task_environment_.GetMainThreadTaskRunner(),
  117. task_environment_.GetMainThreadTaskRunner(), "test",
  118. demuxer_stream_.get(), std::move(stream_sender),
  119. std::move(producer_end));
  120. // DemuxerStreamAdapter constructor posts task to main thread to
  121. // register MessageReceiverCallback. Therefore it should call
  122. // RunPendingTasks() to make sure task is executed.
  123. RunPendingTasks();
  124. }
  125. void TearDown() override { base::RunLoop().RunUntilIdle(); }
  126. void RunPendingTasks() { base::RunLoop().RunUntilIdle(); }
  127. protected:
  128. void SetUp() override { SetUpDataPipe(); }
  129. base::test::SingleThreadTaskEnvironment task_environment_;
  130. std::unique_ptr<FakeDemuxerStream> demuxer_stream_;
  131. std::unique_ptr<FakeRemotingDataStreamSender> data_stream_sender_;
  132. std::unique_ptr<MockDemuxerStreamAdapter> demuxer_stream_adapter_;
  133. };
  134. TEST_F(DemuxerStreamAdapterTest, SingleReadUntil) {
  135. // Read will be called once since it doesn't return frame buffer in the dummy
  136. // implementation.
  137. EXPECT_CALL(*demuxer_stream_, Read(_)).Times(1);
  138. demuxer_stream_adapter_->FakeReadUntil(3, 999);
  139. RunPendingTasks();
  140. }
  141. TEST_F(DemuxerStreamAdapterTest, MultiReadUntil) {
  142. // Read will be called once since it doesn't return frame buffer in the dummy
  143. // implementation, and 2nd one will not proceed when there is ongoing read.
  144. EXPECT_CALL(*demuxer_stream_, Read(_)).Times(1);
  145. demuxer_stream_adapter_->FakeReadUntil(1, 100);
  146. RunPendingTasks();
  147. demuxer_stream_adapter_->FakeReadUntil(2, 101);
  148. RunPendingTasks();
  149. }
  150. TEST_F(DemuxerStreamAdapterTest, WriteOneFrameSmallerThanCapacity) {
  151. EXPECT_CALL(*demuxer_stream_, Read(_)).Times(1);
  152. // Sends a frame with size 50 bytes, pts = 1 and key frame.
  153. demuxer_stream_->CreateFakeFrame(50, true, 1 /* pts */);
  154. demuxer_stream_adapter_->FakeReadUntil(1, 999);
  155. RunPendingTasks();
  156. // Checks if it's sent to consumer side and data is correct
  157. ASSERT_EQ(data_stream_sender_->send_frame_count(), 1U);
  158. ASSERT_TRUE(data_stream_sender_->ValidateFrameBuffer(0, 50, true, 1));
  159. openscreen::cast::RpcMessage* last_rpc =
  160. demuxer_stream_adapter_->last_received_rpc();
  161. ASSERT_TRUE(last_rpc);
  162. ASSERT_EQ(last_rpc->proc(),
  163. openscreen::cast::RpcMessage::RPC_DS_READUNTIL_CALLBACK);
  164. ASSERT_EQ(last_rpc->handle(), 999);
  165. data_stream_sender_->ResetHistory();
  166. }
  167. TEST_F(DemuxerStreamAdapterTest, WriteOneFrameLargerThanCapacity) {
  168. EXPECT_CALL(*demuxer_stream_, Read(_)).Times(1);
  169. // Sends a frame with size 800 bytes, pts = 1 and key frame.
  170. demuxer_stream_->CreateFakeFrame(800, true, 1 /* pts */);
  171. demuxer_stream_adapter_->FakeReadUntil(1, 999);
  172. RunPendingTasks();
  173. // Checks if it's sent to consumer side and data is correct
  174. ASSERT_EQ(data_stream_sender_->send_frame_count(), 1U);
  175. ASSERT_TRUE(data_stream_sender_->ValidateFrameBuffer(0, 800, true, 1));
  176. openscreen::cast::RpcMessage* last_rpc =
  177. demuxer_stream_adapter_->last_received_rpc();
  178. ASSERT_TRUE(last_rpc);
  179. ASSERT_EQ(last_rpc->proc(),
  180. openscreen::cast::RpcMessage::RPC_DS_READUNTIL_CALLBACK);
  181. ASSERT_EQ(last_rpc->handle(), 999);
  182. data_stream_sender_->ResetHistory();
  183. }
  184. TEST_F(DemuxerStreamAdapterTest, SendFrameAndSignalFlushMix) {
  185. EXPECT_CALL(*demuxer_stream_, Read(_)).Times(4);
  186. // Sends a frame with size 50 bytes, pts = 1 and key frame.
  187. demuxer_stream_->CreateFakeFrame(50, true, 1 /* pts */);
  188. // Issues ReadUntil request with frame count up to 1 (fetch #0).
  189. demuxer_stream_adapter_->FakeReadUntil(1, 100);
  190. RunPendingTasks();
  191. ASSERT_EQ(data_stream_sender_->send_frame_count(), 1U);
  192. ASSERT_TRUE(data_stream_sender_->ValidateFrameBuffer(0, 50, true, 1));
  193. openscreen::cast::RpcMessage* last_rpc =
  194. demuxer_stream_adapter_->last_received_rpc();
  195. ASSERT_TRUE(last_rpc);
  196. ASSERT_EQ(last_rpc->proc(),
  197. openscreen::cast::RpcMessage::RPC_DS_READUNTIL_CALLBACK);
  198. ASSERT_EQ(last_rpc->handle(), 100);
  199. data_stream_sender_->ResetHistory();
  200. // Sends two frames with size 100 + 150 bytes
  201. demuxer_stream_->CreateFakeFrame(100, false, 2 /* pts */);
  202. demuxer_stream_->CreateFakeFrame(150, false, 3 /* pts */);
  203. // Issues ReadUntil request with frame count up to 3 (fetch #1 and #2).
  204. demuxer_stream_adapter_->FakeReadUntil(3, 101);
  205. RunPendingTasks();
  206. ASSERT_EQ(data_stream_sender_->send_frame_count(), 2U);
  207. ASSERT_TRUE(data_stream_sender_->ValidateFrameBuffer(0, 100, false, 2));
  208. ASSERT_TRUE(data_stream_sender_->ValidateFrameBuffer(1, 150, false, 3));
  209. last_rpc = demuxer_stream_adapter_->last_received_rpc();
  210. ASSERT_TRUE(last_rpc);
  211. ASSERT_EQ(last_rpc->proc(),
  212. openscreen::cast::RpcMessage::RPC_DS_READUNTIL_CALLBACK);
  213. ASSERT_EQ(last_rpc->handle(), 101);
  214. data_stream_sender_->ResetHistory();
  215. // Signal flush
  216. ASSERT_EQ(data_stream_sender_->cancel_in_flight_count(), 0U);
  217. demuxer_stream_adapter_->SignalFlush(true);
  218. RunPendingTasks();
  219. ASSERT_EQ(data_stream_sender_->cancel_in_flight_count(), 1U);
  220. // ReadUntil request after flush signaling should be ignored.
  221. demuxer_stream_->CreateFakeFrame(100, false, 4 /* pts */);
  222. demuxer_stream_->CreateFakeFrame(100, false, 5 /* pts */);
  223. // Issues ReadUntil request with frame count up to 5 (fetch #3 and #4).
  224. demuxer_stream_adapter_->FakeReadUntil(5, 102);
  225. RunPendingTasks();
  226. ASSERT_EQ(data_stream_sender_->send_frame_count(), 0U);
  227. // Signal flush done
  228. demuxer_stream_adapter_->SignalFlush(false);
  229. RunPendingTasks();
  230. ASSERT_EQ(data_stream_sender_->cancel_in_flight_count(), 1U);
  231. data_stream_sender_->ResetHistory();
  232. // Re-issues ReadUntil request with frame count up to 4 (fetch #3).
  233. demuxer_stream_adapter_->FakeReadUntil(4, 103);
  234. RunPendingTasks();
  235. ASSERT_EQ(data_stream_sender_->send_frame_count(), 1U);
  236. ASSERT_TRUE(data_stream_sender_->ValidateFrameBuffer(0, 100, false, 4));
  237. last_rpc = demuxer_stream_adapter_->last_received_rpc();
  238. ASSERT_TRUE(last_rpc);
  239. ASSERT_EQ(last_rpc->proc(),
  240. openscreen::cast::RpcMessage::RPC_DS_READUNTIL_CALLBACK);
  241. ASSERT_EQ(last_rpc->handle(), 103);
  242. data_stream_sender_->ResetHistory();
  243. }
  244. TEST_F(DemuxerStreamAdapterTest, DuplicateInitializeCausesFatalError) {
  245. std::vector<StopTrigger> errors;
  246. demuxer_stream_adapter_->TakeErrors(&errors);
  247. ASSERT_TRUE(errors.empty());
  248. demuxer_stream_adapter_->DoDuplicateInitialize();
  249. demuxer_stream_adapter_->TakeErrors(&errors);
  250. ASSERT_EQ(1u, errors.size());
  251. EXPECT_EQ(PEERS_OUT_OF_SYNC, errors[0]);
  252. }
  253. TEST_F(DemuxerStreamAdapterTest, ClosingMessagePipeCausesMojoDisconnected) {
  254. std::vector<StopTrigger> errors;
  255. demuxer_stream_adapter_->TakeErrors(&errors);
  256. ASSERT_TRUE(errors.empty());
  257. // Closes one end of mojo message pipes.
  258. data_stream_sender_.reset();
  259. RunPendingTasks(); // Allow notification from mojo to propagate.
  260. demuxer_stream_adapter_->TakeErrors(&errors);
  261. ASSERT_EQ(1u, errors.size());
  262. EXPECT_EQ(MOJO_DISCONNECTED, errors[0]);
  263. }
  264. TEST_F(DemuxerStreamAdapterTest, ClosingDataPipeCausesWriteError) {
  265. EXPECT_CALL(*demuxer_stream_, Read(_)).Times(1);
  266. std::vector<StopTrigger> errors;
  267. demuxer_stream_adapter_->TakeErrors(&errors);
  268. ASSERT_TRUE(errors.empty());
  269. // Closes the consumer end of the data pipe.
  270. data_stream_sender_->CloseDataPipe();
  271. demuxer_stream_->CreateFakeFrame(100, true /* key frame */, 1 /* pts */);
  272. demuxer_stream_adapter_->FakeReadUntil(1, 999);
  273. RunPendingTasks(); // Allow notification from mojo to propagate.
  274. demuxer_stream_adapter_->TakeErrors(&errors);
  275. ASSERT_EQ(1u, errors.size());
  276. EXPECT_EQ(DATA_PIPE_WRITE_ERROR, errors[0]);
  277. }
  278. } // namespace remoting
  279. } // namespace media