[cast_streaming] Add a Cast Streaming test receiver target

This adds a standalone cast streaming receiver implementation intended
to be used for tests involving cast streaming.

Bug: 1110490
Change-Id: Ide0e33587d2e269e8808102342cc30aa7513cb7c
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2583103
Commit-Queue: Fabrice de Gans-Riberi <fdegans@chromium.org>
Reviewed-by: Sergey Ulanov <sergeyu@chromium.org>
Cr-Commit-Position: refs/heads/master@{#836783}

fuchsia/cast_streaming/BUILD.gn

@@ -102,6 +102,27 @@ source_set("cast_streaming_test_sender") {
   ]
 }
 
+source_set("cast_streaming_test_receiver") {
+  testonly = true
+  deps = [
+    ":cast_streaming_session",
+    "//base",
+    "//components/openscreen_platform",
+    "//media",
+    "//media/mojo/common",
+    "//mojo/public/cpp/system",
+    "//third_party/openscreen/src/cast/common:public",
+    "//third_party/openscreen/src/cast/streaming:receiver",
+    "//third_party/openscreen/src/platform:api",
+    "//third_party/openscreen/src/util",
+  ]
+  visibility = [ ":*" ]
+  sources = [
+    "test/cast_streaming_test_receiver.cc",
+    "test/cast_streaming_test_receiver.h",
+  ]
+}
+
 test("cast_streaming_unittests") {
   sources = [ "cast_message_port_impl_unittest.cc" ]
   deps = [

fuchsia/cast_streaming/test/cast_streaming_test_receiver.cc

+// Copyright 2020 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 "fuchsia/cast_streaming/test/cast_streaming_test_receiver.h"
+
+#include "base/logging.h"
+#include "base/run_loop.h"
+#include "base/threading/sequenced_task_runner_handle.h"
+
+namespace cast_streaming {
+
+CastStreamingTestReceiver::CastStreamingTestReceiver() = default;
+CastStreamingTestReceiver::~CastStreamingTestReceiver() = default;
+
+void CastStreamingTestReceiver::Start(
+    std::unique_ptr<cast_api_bindings::MessagePort> message_port) {
+  VLOG(1) << __func__;
+  receiver_session_.Start(this, std::move(message_port),
+                          base::SequencedTaskRunnerHandle::Get());
+}
+
+void CastStreamingTestReceiver::Stop() {
+  VLOG(1) << __func__;
+  is_active_ = false;
+  receiver_session_.Stop();
+
+  // Clear the configuration.
+  audio_config_.reset();
+  audio_decoder_buffer_reader_.reset();
+  audio_buffers_.clear();
+  video_config_.reset();
+  video_decoder_buffer_reader_.reset();
+  video_buffers_.clear();
+
+  // Clear any pending callback.
+  if (receiver_updated_closure_) {
+    std::move(receiver_updated_closure_).Run();
+  }
+}
+
+void CastStreamingTestReceiver::RunUntilStarted() {
+  VLOG(1) << __func__;
+  CHECK(!receiver_updated_closure_);
+  while (!is_active_) {
+    base::RunLoop run_loop;
+    receiver_updated_closure_ = run_loop.QuitClosure();
+    run_loop.Run();
+  }
+}
+
+void CastStreamingTestReceiver::RunUntilStopped() {
+  VLOG(1) << __func__;
+  CHECK(!receiver_updated_closure_);
+  while (is_active_) {
+    base::RunLoop run_loop;
+    receiver_updated_closure_ = run_loop.QuitClosure();
+    run_loop.Run();
+  }
+}
+
+bool CastStreamingTestReceiver::RunUntilAudioFramesCountIsAtLeast(
+    size_t audio_frames_count) {
+  VLOG(1) << __func__;
+  CHECK(!receiver_updated_closure_);
+  while (is_active_ && audio_buffers_.size() < audio_frames_count) {
+    base::RunLoop run_loop;
+    receiver_updated_closure_ = run_loop.QuitClosure();
+    run_loop.Run();
+  }
+  return is_active_;
+}
+
+bool CastStreamingTestReceiver::RunUntilVideoFramesCountIsAtLeast(
+    size_t video_frames_count) {
+  VLOG(1) << __func__;
+  CHECK(!receiver_updated_closure_);
+  while (is_active_ && video_buffers_.size() < video_frames_count) {
+    base::RunLoop run_loop;
+    receiver_updated_closure_ = run_loop.QuitClosure();
+    run_loop.Run();
+  }
+  return is_active_;
+}
+
+void CastStreamingTestReceiver::OnAudioBufferRead(
+    scoped_refptr<media::DecoderBuffer> buffer) {
+  VLOG(3) << __func__;
+
+  // The pending buffer reads are cancelled when we reset the data pipe on a
+  // configuration change. Just ignore them and return early here.
+  if (!buffer)
+    return;
+
+  audio_buffers_.push_back(buffer);
+  if (receiver_updated_closure_) {
+    std::move(receiver_updated_closure_).Run();
+  }
+}
+
+void CastStreamingTestReceiver::OnVideoBufferRead(
+    scoped_refptr<media::DecoderBuffer> buffer) {
+  VLOG(3) << __func__;
+
+  // The pending buffer reads are cancelled when we reset the data pipe on a
+  // configuration change. Just ignore them and return early here.
+  if (!buffer)
+    return;
+
+  video_buffers_.push_back(buffer);
+  if (receiver_updated_closure_) {
+    std::move(receiver_updated_closure_).Run();
+  }
+}
+
+void CastStreamingTestReceiver::OnSessionInitialization(
+    base::Optional<CastStreamingSession::AudioStreamInfo> audio_stream_info,
+    base::Optional<CastStreamingSession::VideoStreamInfo> video_stream_info) {
+  VLOG(1) << __func__;
+  if (audio_stream_info) {
+    audio_decoder_buffer_reader_ =
+        std::make_unique<media::MojoDecoderBufferReader>(
+            std::move(audio_stream_info->data_pipe));
+    audio_config_ = audio_stream_info->decoder_config;
+  }
+
+  if (video_stream_info) {
+    video_decoder_buffer_reader_ =
+        std::make_unique<media::MojoDecoderBufferReader>(
+            std::move(video_stream_info->data_pipe));
+    video_config_ = video_stream_info->decoder_config;
+  }
+
+  is_active_ = true;
+  if (receiver_updated_closure_) {
+    std::move(receiver_updated_closure_).Run();
+  }
+}
+
+void CastStreamingTestReceiver::OnAudioBufferReceived(
+    media::mojom::DecoderBufferPtr buffer) {
+  VLOG(3) << __func__;
+  audio_decoder_buffer_reader_->ReadDecoderBuffer(
+      std::move(buffer),
+      base::BindOnce(&CastStreamingTestReceiver::OnAudioBufferRead,
+                     base::Unretained(this)));
+}
+
+void CastStreamingTestReceiver::OnVideoBufferReceived(
+    media::mojom::DecoderBufferPtr buffer) {
+  VLOG(3) << __func__;
+  video_decoder_buffer_reader_->ReadDecoderBuffer(
+      std::move(buffer),
+      base::BindOnce(&CastStreamingTestReceiver::OnVideoBufferRead,
+                     base::Unretained(this)));
+}
+
+void CastStreamingTestReceiver::OnSessionReinitialization(
+    base::Optional<CastStreamingSession::AudioStreamInfo> audio_stream_info,
+    base::Optional<CastStreamingSession::VideoStreamInfo> video_stream_info) {
+  VLOG(1) << __func__;
+
+  // TODO(crbug.com/1110490): Add tests handling the session reinitialization
+  // case.
+  Stop();
+}
+
+void CastStreamingTestReceiver::OnSessionEnded() {
+  VLOG(1) << __func__;
+  if (is_active_) {
+    // Do not call Stop() if the session is already ending.
+    Stop();
+  }
+}
+
+}  // namespace cast_streaming

fuchsia/cast_streaming/test/cast_streaming_test_receiver.h

+// Copyright 2020 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.
+
+#ifndef FUCHSIA_CAST_STREAMING_TEST_CAST_STREAMING_TEST_RECEIVER_H_
+#define FUCHSIA_CAST_STREAMING_TEST_CAST_STREAMING_TEST_RECEIVER_H_
+
+#include "base/callback.h"
+#include "base/optional.h"
+#include "components/cast/message_port/message_port.h"
+#include "fuchsia/cast_streaming/public/cast_streaming_session.h"
+#include "media/mojo/common/mojo_decoder_buffer_converter.h"
+
+namespace cast_streaming {
+
+// Cast Streaming Receiver implementation for testing. This class provides basic
+// functionality for starting a Cast Streaming Receiver and receiving audio and
+// video frames to a Cast Streaming Sender. Example usage:
+//
+// std::unique_ptr<cast_api_bindings::MessagePort> sender_message_port;
+// std::unique_ptr<cast_api_bindings::MessagePort> receiver_message_port;
+// cast_api_bindings::MessagePort::CreatePair(&sender_message_port,
+//                                            &receiver_message_port);
+//
+// // Send |sender_message_port| to a Sender and start it.
+//
+// CastStreamingTestReceiver receiver;
+// receiver.Start(std::move(receiver_message_port));
+// receiver.RunUntilStarted();
+//
+// if (!receiver.RunUntilAudioFramesCountIsAtLeast(1u)) {
+//   return;
+// }
+// // Handle receiver.audio_frames()[0] here.
+//
+// if (!receiver.RunUntilVideoFramesCountIsAtLeast(1u)) {
+//   return;
+// }
+// // Handle receiver.video_frames()[0] here.
+//
+// receiver.Stop();
+// receiver.RunUntilStopped();
+class CastStreamingTestReceiver : public CastStreamingSession::Client {
+ public:
+  CastStreamingTestReceiver();
+  ~CastStreamingTestReceiver() final;
+
+  CastStreamingTestReceiver(const CastStreamingTestReceiver&) = delete;
+  CastStreamingTestReceiver& operator=(const CastStreamingTestReceiver&) =
+      delete;
+
+  // Uses |message_port| as the Receiverr-end of a Cast Streaming MessagePort to
+  // start a Cast Streaming Session.
+  void Start(std::unique_ptr<cast_api_bindings::MessagePort> message_port);
+  void Stop();
+
+  // Helper methods running the Receiver until it is properly started or
+  // stopped. Only one of the Run*() methods can be called at any given time.
+  void RunUntilStarted();
+  void RunUntilStopped();
+
+  // Helper methods running the Receiver until |*frames_count| audio or video
+  // frames have been received or until the Receiver has stopped. Returns true
+  // if the session is still active. Only one of the Run*() methods can be
+  // called at any given time.
+  bool RunUntilAudioFramesCountIsAtLeast(size_t audio_frames_count);
+  bool RunUntilVideoFramesCountIsAtLeast(size_t video_frames_count);
+
+  bool is_active() { return is_active_; }
+  base::Optional<media::AudioDecoderConfig> audio_config() {
+    return audio_config_;
+  }
+  base::Optional<media::VideoDecoderConfig> video_config() {
+    return video_config_;
+  }
+  const std::vector<scoped_refptr<media::DecoderBuffer>>& audio_buffers() {
+    return audio_buffers_;
+  }
+  const std::vector<scoped_refptr<media::DecoderBuffer>>& video_buffers() {
+    return video_buffers_;
+  }
+
+ private:
+  // Callbacks for |audio_decoder_buffer_reader_| and
+  // |video_decoder_buffer_reader_|, respectively.
+  void OnAudioBufferRead(scoped_refptr<media::DecoderBuffer> buffer);
+  void OnVideoBufferRead(scoped_refptr<media::DecoderBuffer> buffer);
+
+  // CastStreamingSession::Client implementation.
+  void OnSessionInitialization(
+      base::Optional<CastStreamingSession::AudioStreamInfo> audio_stream_info,
+      base::Optional<CastStreamingSession::VideoStreamInfo> video_stream_info)
+      final;
+  void OnAudioBufferReceived(media::mojom::DecoderBufferPtr buffer) final;
+  void OnVideoBufferReceived(media::mojom::DecoderBufferPtr buffer) final;
+  void OnSessionReinitialization(
+      base::Optional<CastStreamingSession::AudioStreamInfo> audio_stream_info,
+      base::Optional<CastStreamingSession::VideoStreamInfo> video_stream_info)
+      final;
+  void OnSessionEnded() final;
+
+  CastStreamingSession receiver_session_;
+  bool is_active_ = false;
+  base::OnceClosure receiver_updated_closure_;
+
+  base::Optional<media::AudioDecoderConfig> audio_config_;
+  std::unique_ptr<media::MojoDecoderBufferReader> audio_decoder_buffer_reader_;
+  std::vector<scoped_refptr<media::DecoderBuffer>> audio_buffers_;
+
+  base::Optional<media::VideoDecoderConfig> video_config_;
+  std::unique_ptr<media::MojoDecoderBufferReader> video_decoder_buffer_reader_;
+  std::vector<scoped_refptr<media::DecoderBuffer>> video_buffers_;
+};
+
+}  // namespace cast_streaming
+
+#endif  // FUCHSIA_CAST_STREAMING_TEST_CAST_STREAMING_TEST_RECEIVER_H_