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