Media Remoting: UMAs to track session events and measurements.

media/remoting/remote_demuxer_stream_adapter_unittest.cc

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "media/remoting/remote_demuxer_stream_adapter.h"
 
 #include <memory>
+#include <vector>
 
 #include "base/callback_helpers.h"
 #include "base/run_loop.h"
 #include "media/base/decoder_buffer.h"
 #include "media/base/demuxer_stream.h"
 #include "media/remoting/fake_remoting_controller.h"
 #include "media/remoting/fake_remoting_demuxer_stream_provider.h"
 #include "media/remoting/rpc/proto_utils.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ skipping 15 matching lines @@
       mojom::RemotingDataStreamSenderPtrInfo stream_sender_info,
       mojo::ScopedDataPipeProducerHandle producer_handle)
       : weak_factory_(this) {
     rpc_broker_.reset(new RpcBroker(
         base::Bind(&MockRemoteDemuxerStreamAdapter::OnSendMessageToSink,
                    weak_factory_.GetWeakPtr())));
     demuxer_stream_adapter_.reset(new RemoteDemuxerStreamAdapter(
         std::move(main_task_runner), std::move(media_task_runner), name,
         demuxer_stream, rpc_broker_->GetWeakPtr(),
         rpc_broker_->GetUniqueHandle(), std::move(stream_sender_info),
-        std::move(producer_handle)));
+        std::move(producer_handle),
+        base::Bind(&MockRemoteDemuxerStreamAdapter::OnError,
+                   weak_factory_.GetWeakPtr())));
 
     // Faking initialization with random callback handle to start mojo watcher.
     demuxer_stream_adapter_->Initialize(3);
   }
 
+  ~MockRemoteDemuxerStreamAdapter() {
+    // Make sure unit tests that did not expect errors did not cause any errors.
+    EXPECT_TRUE(errors_.empty());
+  }
+
   int rpc_handle() const { return demuxer_stream_adapter_->rpc_handle(); }
+
   base::WeakPtr<MockRemoteDemuxerStreamAdapter> GetWeakPtr() {
     return weak_factory_.GetWeakPtr();
   }
 
+  void DoDuplicateInitialize() { demuxer_stream_adapter_->Initialize(999); }
+
+  void TakeErrors(std::vector<StopTrigger>* errors) {
+    errors->swap(errors_);
+    errors_.clear();
+  }
+
   // Fake to signal that it's in reading state.
   void FakeReadUntil(int read_until_count, int callback_handle) {
     std::unique_ptr<pb::RpcMessage> rpc(new pb::RpcMessage());
     rpc->set_handle(rpc_handle());
     rpc->set_proc(pb::RpcMessage::RPC_DS_READUNTIL);
     auto* read_message = rpc->mutable_demuxerstream_readuntil_rpc();
     read_message->set_callback_handle(
         callback_handle);  // Given an unique callback handle.
     read_message->set_count(read_until_count);  // Request 1 frame
 
     demuxer_stream_adapter_->OnReceivedRpc(std::move(rpc));
   }
   void OnNewBuffer(const scoped_refptr<::media::DecoderBuffer>& frame) {
     demuxer_stream_adapter_->OnNewBuffer(DemuxerStream::kOk, frame);
   }
 
   void SignalFlush(bool flush) { demuxer_stream_adapter_->SignalFlush(flush); }
 
   pb::RpcMessage* last_received_rpc() const { return last_received_rpc_.get(); }
 
  private:
   void OnSendMessageToSink(std::unique_ptr<std::vector<uint8_t>> message) {
     last_received_rpc_.reset(new remoting::pb::RpcMessage());
     CHECK(last_received_rpc_->ParseFromArray(message->data(), message->size()));
   }
 
+  void OnError(StopTrigger stop_trigger) { errors_.push_back(stop_trigger); }
+
   std::unique_ptr<RpcBroker> rpc_broker_;
   std::unique_ptr<RemoteDemuxerStreamAdapter> demuxer_stream_adapter_;
   std::unique_ptr<remoting::pb::RpcMessage> last_received_rpc_;
+
+  std::vector<StopTrigger> errors_;
+
   base::WeakPtrFactory<MockRemoteDemuxerStreamAdapter> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(MockRemoteDemuxerStreamAdapter);
 };
 
 class RemoteDemuxerStreamAdapterTest : public ::testing::Test {
  public:
   RemoteDemuxerStreamAdapterTest() {}
   ~RemoteDemuxerStreamAdapterTest() override = default;
 
@@ skipping 22 matching lines @@
     RunPendingTasks();
   }
 
   void TearDown() override { base::RunLoop().RunUntilIdle(); }
 
   void RunPendingTasks() { base::RunLoop().RunUntilIdle(); }
 
  protected:
   void SetUp() override { SetUpDataPipe(); }
 
+  // TODO(miu): Add separate media thread, to test threading also.
+  base::MessageLoop message_loop_;
   std::unique_ptr<DummyDemuxerStream> demuxer_stream_;
   std::unique_ptr<FakeRemotingDataStreamSender> data_stream_sender_;
   std::unique_ptr<MockRemoteDemuxerStreamAdapter> demuxer_stream_adapter_;
-  base::MessageLoop message_loop_;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(RemoteDemuxerStreamAdapterTest);
 };
 
 TEST_F(RemoteDemuxerStreamAdapterTest, SingleReadUntil) {
   // Read will be called once since it doesn't return frame buffer in the dummy
   // implementation.
   EXPECT_CALL(*demuxer_stream_, Read(_)).Times(1);
 
@@ skipping 101 matching lines @@
   RunPendingTasks();
   ASSERT_EQ(data_stream_sender_->send_frame_count(), 1U);
   ASSERT_TRUE(data_stream_sender_->ValidateFrameBuffer(0, 100, false, 4));
   last_rpc = demuxer_stream_adapter_->last_received_rpc();
   ASSERT_TRUE(last_rpc);
   ASSERT_EQ(last_rpc->proc(), pb::RpcMessage::RPC_DS_READUNTIL_CALLBACK);
   ASSERT_EQ(last_rpc->handle(), 103);
   data_stream_sender_->ResetHistory();
 }
 
+TEST_F(RemoteDemuxerStreamAdapterTest, DuplicateInitializeCausesFatalError) {
+  std::vector<StopTrigger> errors;
+  demuxer_stream_adapter_->TakeErrors(&errors);
+  ASSERT_TRUE(errors.empty());
+
+  demuxer_stream_adapter_->DoDuplicateInitialize();
+  demuxer_stream_adapter_->TakeErrors(&errors);
+  ASSERT_EQ(1u, errors.size());
+  EXPECT_EQ(PEERS_OUT_OF_SYNC, errors[0]);
+}
+
+TEST_F(RemoteDemuxerStreamAdapterTest, ClosingPipeCausesFatalError) {
+  std::vector<StopTrigger> errors;
+  demuxer_stream_adapter_->TakeErrors(&errors);
+  ASSERT_TRUE(errors.empty());
+
+  // Closes one end of mojo message and data pipes.
+  data_stream_sender_.reset();
+  RunPendingTasks();  // Allow notification from mojo to propagate.
+
+  demuxer_stream_adapter_->TakeErrors(&errors);
+  ASSERT_EQ(1u, errors.size());
+  EXPECT_EQ(MOJO_PIPE_ERROR, errors[0]);
+}
+
 }  // namesapce remoting
 }  // namespace media