components/cast_streaming/browser/playback_command_dispatcher.cc

// Copyright 2021 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 "components/cast_streaming/browser/playback_command_dispatcher.h"

#include "base/bind.h"
#include "base/task/bind_post_task.h"
#include "components/cast_streaming/browser/renderer_rpc_call_translator.h"
#include "components/cast_streaming/public/rpc_call_message_handler.h"
#include "media/mojo/mojom/renderer.mojom.h"
#include "mojo/public/cpp/bindings/remote.h"

namespace cast_streaming {

PlaybackCommandDispatcher::PlaybackCommandDispatcher(
    scoped_refptr<base::SequencedTaskRunner> task_runner,
    mojo::AssociatedRemote<mojom::RendererController> control_configuration)
    : RpcInitializationCallHandlerBase(base::BindRepeating(
          &PlaybackCommandDispatcher::SendRemotingRpcMessageToRemote,
          base::Unretained(this))),
      task_runner_(std::move(task_runner)),
      weak_factory_(this) {
  // Create a muxer using the "real" media::mojom::Renderer instance that
  // connects to the remote media::Renderer.
  mojo::Remote<media::mojom::Renderer> renderer;
  control_configuration->SetPlaybackController(
      renderer.BindNewPipeAndPassReceiver(),
      base::BindOnce(&PlaybackCommandDispatcher::OnSetPlaybackControllerDone,
                     weak_factory_.GetWeakPtr()));
  muxer_ = std::make_unique<RendererControlMultiplexer>(std::move(renderer),
                                                        task_runner_);

  // Create a "fake" media::mojom::Renderer so that the RpcCallTranslator can
  // pass commands to the |muxer_|.
  mojo::Remote<media::mojom::Renderer> translators_renderer;
  RegisterCommandSource(translators_renderer.BindNewPipeAndPassReceiver());

  call_translator_ = std::make_unique<remoting::RendererRpcCallTranslator>(
      std::move(translators_renderer));
}

PlaybackCommandDispatcher::~PlaybackCommandDispatcher() {
  OnRemotingSessionEnded();
}

void PlaybackCommandDispatcher::RegisterCommandSource(
    mojo::PendingReceiver<media::mojom::Renderer> controls) {
  muxer_->RegisterController(std::move(controls));
}

void PlaybackCommandDispatcher::OnRemotingSessionNegotiated(
    openscreen::cast::RpcMessenger* messenger) {
  DCHECK(messenger);

  messenger_ = messenger;
  handle_ = messenger_->GetUniqueHandle();

  // Include the |handle_| in the callback so that it will persist even upon
  // re-negotiation.
  auto message_processor_callback = base::BindPostTask(
      task_runner_,
      base::BindRepeating(
          &PlaybackCommandDispatcher::SendRemotingRpcMessageToRemote,
          weak_factory_.GetWeakPtr(), handle_),
      FROM_HERE);
  call_translator_->SetMessageProcessor(std::move(message_processor_callback));

  auto message_receiver_callback = base::BindPostTask(
      task_runner_,
      base::BindRepeating(
          &PlaybackCommandDispatcher::ProcessRemotingRpcMessageFromRemote,
          weak_factory_.GetWeakPtr()),
      FROM_HERE);
  messenger_->RegisterMessageReceiverCallback(
      handle_, [cb = std::move(message_receiver_callback)](
                   std::unique_ptr<openscreen::cast::RpcMessage> message) {
        cb.Run(std::move(message));
      });

  demuxer_stream_handler_ = std::make_unique<remoting::RpcDemuxerStreamHandler>(
      this, messenger_,
      base::BindRepeating(
          &PlaybackCommandDispatcher::SendRemotingRpcMessageToRemote,
          base::Unretained(this)));
}

void PlaybackCommandDispatcher::ConfigureRemotingAsync(
    Dispatcher* dispatcher,
    const openscreen::cast::ReceiverSession* session,
    openscreen::cast::ReceiverSession::ConfiguredReceivers receivers) {
  DCHECK(dispatcher);
  DCHECK(session);
  DCHECK(demuxer_stream_handler_);
  DCHECK(!streaming_init_info_);

  streaming_dispatcher_ = dispatcher;
  receiver_session_ = session;

  absl::optional<StreamingInitializationInfo::AudioStreamInfo>
      audio_stream_info;
  if (receivers.audio_receiver) {
    auto no_buffers_cb = base::BindPostTask(
        task_runner_,
        base::BindRepeating(
            &remoting::RpcDemuxerStreamHandler::RequestMoreAudioBuffers,
            demuxer_stream_handler_->GetWeakPtr()),
        FROM_HERE);
    auto error_cb = base::BindPostTask(
        task_runner_,
        base::BindRepeating(&remoting::RpcDemuxerStreamHandler::OnAudioError,
                            demuxer_stream_handler_->GetWeakPtr()),
        FROM_HERE);
    audio_stream_info.emplace(media::AudioDecoderConfig(),
                              receivers.audio_receiver,
                              std::move(no_buffers_cb), std::move(error_cb));
  }

  absl::optional<StreamingInitializationInfo::VideoStreamInfo>
      video_stream_info;
  if (receivers.video_receiver) {
    auto no_buffers_cb = base::BindPostTask(
        task_runner_,
        base::BindRepeating(
            &remoting::RpcDemuxerStreamHandler::RequestMoreVideoBuffers,
            demuxer_stream_handler_->GetWeakPtr()),
        FROM_HERE);
    auto error_cb = base::BindPostTask(
        task_runner_,
        base::BindRepeating(&remoting::RpcDemuxerStreamHandler::OnVideoError,
                            demuxer_stream_handler_->GetWeakPtr()),
        FROM_HERE);
    video_stream_info.emplace(media::VideoDecoderConfig(),
                              receivers.video_receiver,
                              std::move(no_buffers_cb), std::move(error_cb));
  }

  streaming_init_info_.emplace(receiver_session_, std::move(audio_stream_info),
                               std::move(video_stream_info));
}

void PlaybackCommandDispatcher::OnRemotingSessionEnded() {
  demuxer_stream_handler_.reset();
  if (messenger_) {
    messenger_->UnregisterMessageReceiverCallback(handle_);
    messenger_ = nullptr;
  }
  streaming_init_info_ = absl::nullopt;
}

void PlaybackCommandDispatcher::SendRemotingRpcMessageToRemote(
    openscreen::cast::RpcMessenger::Handle handle,
    std::unique_ptr<openscreen::cast::RpcMessage> message) {
  DCHECK(message);
  DCHECK(task_runner_->RunsTasksInCurrentSequence());

  if (!messenger_) {
    return;
  }

  message->set_handle(handle);
  messenger_->SendMessageToRemote(*message);
}

void PlaybackCommandDispatcher::ProcessRemotingRpcMessageFromRemote(
    std::unique_ptr<openscreen::cast::RpcMessage> message) {
  DCHECK(message);
  DCHECK(task_runner_->RunsTasksInCurrentSequence());

  const bool did_dispatch_as_initialization_call =
      remoting::DispatchInitializationRpcCall(message.get(), this);
  if (did_dispatch_as_initialization_call) {
    return;
  }

  const bool did_dispatch_as_renderer_call =
      call_translator_ &&
      remoting::DispatchRendererRpcCall(message.get(), call_translator_.get());
  if (did_dispatch_as_renderer_call) {
    return;
  }

  const bool did_dispatch_as_demuxer_stream_callback =
      demuxer_stream_handler_ &&
      remoting::DispatchDemuxerStreamCBRpcCall(message.get(),
                                               demuxer_stream_handler_.get());
  if (did_dispatch_as_demuxer_stream_callback) {
    return;
  }

  LOG(ERROR) << "Unhandled RPC Message for command " << message->proc();
}

void PlaybackCommandDispatcher::OnSetPlaybackControllerDone() {
  has_set_playback_controller_call_returned_ = true;

  if (acquire_renderer_cb_) {
    std::move(acquire_renderer_cb_).Run();
  }
}

void PlaybackCommandDispatcher::RpcAcquireRendererAsync(AcquireRendererCB cb) {
  acquire_renderer_cb_ = base::BindOnce(std::move(cb), handle_);

  if (has_set_playback_controller_call_returned_) {
    std::move(acquire_renderer_cb_).Run();
  }
}

void PlaybackCommandDispatcher::OnRpcAcquireDemuxer(
    openscreen::cast::RpcMessenger::Handle audio_stream_handle,
    openscreen::cast::RpcMessenger::Handle video_stream_handle) {
  if (demuxer_stream_handler_) {
    demuxer_stream_handler_->OnRpcAcquireDemuxer(audio_stream_handle,
                                                 video_stream_handle);
  }
}

void PlaybackCommandDispatcher::OnNewAudioConfig(
    media::AudioDecoderConfig config) {
  DCHECK(streaming_init_info_);
  DCHECK(streaming_dispatcher_);
  if (!streaming_init_info_->audio_stream_info) {
    LOG(ERROR) << "Received Audio config for a remoting session where audio is "
                  "not supported";
    return;
  }

  streaming_init_info_->audio_stream_info->config = std::move(config);
  if (!streaming_init_info_->video_stream_info ||
      !streaming_init_info_->video_stream_info->config.Matches(
          media::VideoDecoderConfig())) {
    // |streaming_init_info_| is intentionally copied here.
    streaming_dispatcher_->StartStreamingSession(streaming_init_info_.value());
  }
}

void PlaybackCommandDispatcher::OnNewVideoConfig(
    media::VideoDecoderConfig config) {
  DCHECK(streaming_init_info_);
  DCHECK(streaming_dispatcher_);
  if (!streaming_init_info_->video_stream_info) {
    LOG(ERROR) << "Received Video config for a remoting session where video is "
                  "not supported";
    return;
  }

  streaming_init_info_->video_stream_info->config = std::move(config);
  if (!streaming_init_info_->audio_stream_info ||
      !streaming_init_info_->audio_stream_info->config.Matches(
          media::AudioDecoderConfig())) {
    // |streaming_init_info_| is intentionally copied here.
    streaming_dispatcher_->StartStreamingSession(streaming_init_info_.value());
  }
}

}  // namespace cast_streaming