[Cast Streaming] Add EnableBitstreamConverter Support This CL updates the existing Remoting code to support calls to EnableBitstreamConverter() by the custom DemuxerStream used to inject frames from Openscreen. Bug: b/209844961 Change-Id: I693383059fb50a7ac4f8c1624dbcff5bc56a845b Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/3656579 Commit-Queue: Ryan Keane <rwkeane@google.com> Reviewed-by: Ken Buchanan <kenrb@chromium.org> Reviewed-by: Fabrice de Gans <fdegans@chromium.org> Cr-Commit-Position: refs/heads/main@{#1008097}
diff --git a/components/cast_streaming/browser/BUILD.gn b/components/cast_streaming/browser/BUILD.gn index 07f344c61..a69837fa 100644 --- a/components/cast_streaming/browser/BUILD.gn +++ b/components/cast_streaming/browser/BUILD.gn
@@ -72,6 +72,7 @@ source_set("streaming_session") { deps = [ ":core", + ":demuxer_stream_data_provider", ":receiver_session_public", ":renderer_controls", ":streaming_initialization_info", @@ -176,6 +177,7 @@ source_set("demuxer_stream_data_provider") { deps = [ + ":demuxer_stream_client", "//base", "//components/cast_streaming/public/mojom", "//media", @@ -199,7 +201,7 @@ source_set("streaming_initialization_info") { public_deps = [ - ":demuxer_stream_data_provider", + ":demuxer_stream_client", "//media", "//third_party/openscreen/src/cast/streaming:receiver", ] @@ -210,8 +212,15 @@ ] } +source_set("demuxer_stream_client") { + visibility = [ ":*" ] + sources = [ "demuxer_stream_client.h" ] + public_deps = [ "//base" ] +} + source_set("renderer_controls") { public_deps = [ + ":demuxer_stream_client", ":streaming_initialization_info", "//base", "//components/cast_streaming/public:remoting_utils", @@ -254,6 +263,7 @@ testonly = true deps = [ ":browser", + ":demuxer_stream_client", ":demuxer_stream_data_provider", ":renderer_controls", "//base",
diff --git a/components/cast_streaming/browser/demuxer_stream_client.h b/components/cast_streaming/browser/demuxer_stream_client.h new file mode 100644 index 0000000..c320bff --- /dev/null +++ b/components/cast_streaming/browser/demuxer_stream_client.h
@@ -0,0 +1,32 @@ +// Copyright 2022 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 COMPONENTS_CAST_STREAMING_BROWSER_DEMUXER_STREAM_CLIENT_H_ +#define COMPONENTS_CAST_STREAMING_BROWSER_DEMUXER_STREAM_CLIENT_H_ + +#include "base/callback.h" + +namespace cast_streaming { + +// This class acts as a client to act upon state changes of the DemuxerStream +// and DemuxerStreamDataProvider with which it is associated. +class DemuxerStreamClient { + public: + virtual ~DemuxerStreamClient() = default; + + // Enables the bitstream converter for the data provider associated with this + // demuxer stream. + using BitstreamConverterEnabledCB = base::OnceCallback<void(bool)>; + virtual void EnableBitstreamConverter(BitstreamConverterEnabledCB cb) = 0; + + // Called when no buffers are available for reading. + virtual void OnNoBuffersAvailable() = 0; + + // Called when a fatal error occurs. Only called once. + virtual void OnError() = 0; +}; + +} // namespace cast_streaming + +#endif // COMPONENTS_CAST_STREAMING_BROWSER_DEMUXER_STREAM_CLIENT_H_
diff --git a/components/cast_streaming/browser/demuxer_stream_data_provider.h b/components/cast_streaming/browser/demuxer_stream_data_provider.h index 7d024cb..6cc294cd 100644 --- a/components/cast_streaming/browser/demuxer_stream_data_provider.h +++ b/components/cast_streaming/browser/demuxer_stream_data_provider.h
@@ -6,7 +6,9 @@ #define COMPONENTS_CAST_STREAMING_BROWSER_DEMUXER_STREAM_DATA_PROVIDER_H_ #include "base/callback_forward.h" +#include "base/memory/weak_ptr.h" #include "base/sequence_checker.h" +#include "components/cast_streaming/browser/demuxer_stream_client.h" #include "components/cast_streaming/public/mojom/demuxer_connector.mojom.h" #include "mojo/public/cpp/bindings/message.h" #include "mojo/public/cpp/bindings/pending_receiver.h" @@ -90,23 +92,21 @@ return config_; } - void SetOnNoBuffersAvailableCallback( - base::RepeatingClosure on_no_buffers_available) { + void SetClient(base::WeakPtr<DemuxerStreamClient> client) { DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); - on_no_buffers_available_ = std::move(on_no_buffers_available); - } - - void SetOnErrorCallback(base::OnceClosure on_error) { - DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); - on_error_ = std::move(on_error); + client_ = std::move(client); } private: using GetBufferCallback = typename TMojoReceiverType::GetBufferCallback; + using EnableBitstreamConverterCallback = + typename TMojoReceiverType::EnableBitstreamConverterCallback; void OnFatalError() { DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); - std::move(on_error_).Run(); + if (client_) { + client_->OnError(); + } std::move(on_mojo_disconnect_).Run(); } @@ -123,7 +123,19 @@ } current_callback_ = std::move(callback); - request_buffer_.Run(on_no_buffers_available_); + request_buffer_.Run( + base::BindOnce(&DemuxerStreamClient::OnNoBuffersAvailable, client_)); + } + + void EnableBitstreamConverter( + EnableBitstreamConverterCallback callback) override { + if (client_) { + client_->EnableBitstreamConverter(std::move(callback)); + } else { + std::move(callback).Run(false); + LOG(WARNING) + << "EnableBitstreamConverter() called when no client was available"; + } } // The most recently set config. @@ -139,12 +151,9 @@ // Callback to request a new buffer be read from the receiver. RequestBufferCB request_buffer_; - // Callback called when no buffers are available for reading. - base::RepeatingClosure on_no_buffers_available_; - - // Callback called on an unrecoverable error, prior to shutdown of the - // component. - base::OnceClosure on_error_; + // Client to use when the associated DemuxerStream requires an action be + // performed. + base::WeakPtr<DemuxerStreamClient> client_; // Callback called upon a mojo disconnection. base::OnceClosure on_mojo_disconnect_;
diff --git a/components/cast_streaming/browser/demuxer_stream_data_provider_unittest.cc b/components/cast_streaming/browser/demuxer_stream_data_provider_unittest.cc index 15fc978..992a1c7 100644 --- a/components/cast_streaming/browser/demuxer_stream_data_provider_unittest.cc +++ b/components/cast_streaming/browser/demuxer_stream_data_provider_unittest.cc
@@ -6,8 +6,10 @@ #include <utility> +#include "base/memory/weak_ptr.h" #include "base/test/bind.h" #include "base/test/task_environment.h" +#include "components/cast_streaming/browser/demuxer_stream_client.h" #include "media/base/audio_codecs.h" #include "media/base/audio_decoder_config.h" #include "media/base/channel_layout.h" @@ -47,12 +49,7 @@ base::Unretained(&callbacks_)), second_config_); - data_provider_->SetOnNoBuffersAvailableCallback(base::BindRepeating( - &DemuxerStreamDataProviderTest::Callbacks::OnNoBuffers, - base::Unretained(&callbacks_))); - data_provider_->SetOnErrorCallback( - base::BindRepeating(&DemuxerStreamDataProviderTest::Callbacks::OnError, - base::Unretained(&callbacks_))); + data_provider_->SetClient(client_.weak_factory_.GetWeakPtr()); std::vector<uint8_t> data = {1, 2, 3}; first_buffer_ = media::DecoderBuffer::CopyFrom(data.data(), 3); @@ -77,8 +74,6 @@ protected: class Callbacks { public: - MOCK_METHOD0(OnNoBuffers, void()); - MOCK_METHOD0(OnError, void()); MOCK_METHOD1(RequestBuffer, void(base::OnceClosure)); MOCK_METHOD0(OnMojoDisconnect, void()); @@ -101,6 +96,17 @@ } }; + class MockDemuxerStreamClient : public DemuxerStreamClient { + public: + ~MockDemuxerStreamClient() override = default; + + MOCK_METHOD1(EnableBitstreamConverter, void(BitstreamConverterEnabledCB)); + MOCK_METHOD0(OnNoBuffersAvailable, void()); + MOCK_METHOD0(OnError, void()); + + base::WeakPtrFactory<MockDemuxerStreamClient> weak_factory_{this}; + }; + using MojoPipePair = std::pair<mojo::ScopedDataPipeProducerHandle, mojo::ScopedDataPipeConsumerHandle>; MojoPipePair GetMojoPipePair() { @@ -112,6 +118,7 @@ } testing::StrictMock<Callbacks> callbacks_; + testing::StrictMock<MockDemuxerStreamClient> client_; base::test::SingleThreadTaskEnvironment task_environment_{ base::test::TaskEnvironment::TimeSource::MOCK_TIME}; @@ -178,26 +185,24 @@ task_environment_.RunUntilIdle(); } -TEST_F(DemuxerStreamDataProviderTest, NoBuffersCallsWithCallback) { +TEST_F(DemuxerStreamDataProviderTest, NoBuffersCallback) { EXPECT_CALL(callbacks_, RequestBuffer(testing::_)) .WillOnce([](base::OnceClosure no_buffers_cb) { std::move(no_buffers_cb).Run(); }); - EXPECT_CALL(callbacks_, OnNoBuffers()); + EXPECT_CALL(client_, OnNoBuffersAvailable()); remote_->GetBuffer(base::BindOnce( &DemuxerStreamDataProviderTest::Callbacks::OnGetBufferDone, base::Unretained(&callbacks_), first_config_, first_buffer_)); task_environment_.RunUntilIdle(); } -TEST_F(DemuxerStreamDataProviderTest, NoBuffersCallsNoCallback) { - data_provider_->SetOnNoBuffersAvailableCallback(base::RepeatingClosure()); - EXPECT_CALL(callbacks_, RequestBuffer(testing::_)) +TEST_F(DemuxerStreamDataProviderTest, EnableBitstreamConverter) { + EXPECT_CALL(client_, EnableBitstreamConverter(testing::_)) .WillOnce( - [](base::OnceClosure no_buffers_cb) { ASSERT_FALSE(no_buffers_cb); }); - remote_->GetBuffer(base::BindOnce( - &DemuxerStreamDataProviderTest::Callbacks::OnGetBufferDone, - base::Unretained(&callbacks_), first_config_, first_buffer_)); + [](base::OnceCallback<void(bool)> cb) { std::move(cb).Run(true); }); + ; + remote_->EnableBitstreamConverter(base::OnceCallback<void(bool)>()); task_environment_.RunUntilIdle(); }
diff --git a/components/cast_streaming/browser/playback_command_dispatcher.cc b/components/cast_streaming/browser/playback_command_dispatcher.cc index 0e11e163..1a53541 100644 --- a/components/cast_streaming/browser/playback_command_dispatcher.cc +++ b/components/cast_streaming/browser/playback_command_dispatcher.cc
@@ -90,39 +90,15 @@ 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)); + receivers.audio_receiver); } 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)); + receivers.video_receiver); } streaming_init_info_.emplace(receiver_session_, std::move(audio_stream_info), @@ -276,10 +252,18 @@ DCHECK(demuxer_stream_handler_); if (streaming_init_info_->audio_stream_info) { - demuxer_stream_handler_->RequestMoreAudioBuffers(); + auto client = demuxer_stream_handler_->GetAudioClient(); + DCHECK(client); + client->OnNoBuffersAvailable(); + streaming_init_info_->audio_stream_info->demuxer_stream_client = + std::move(client); } if (streaming_init_info_->video_stream_info) { - demuxer_stream_handler_->RequestMoreVideoBuffers(); + auto client = demuxer_stream_handler_->GetVideoClient(); + DCHECK(client); + client->OnNoBuffersAvailable(); + streaming_init_info_->video_stream_info->demuxer_stream_client = + std::move(client); } // |streaming_init_info_| is intentionally copied here.
diff --git a/components/cast_streaming/browser/receiver_session_impl.cc b/components/cast_streaming/browser/receiver_session_impl.cc index 05a40204..a09ddfc 100644 --- a/components/cast_streaming/browser/receiver_session_impl.cc +++ b/components/cast_streaming/browser/receiver_session_impl.cc
@@ -95,11 +95,8 @@ base::BindOnce(&ReceiverSessionImpl::OnMojoDisconnect, weak_factory_.GetWeakPtr()), std::move(initialization_info.audio_stream_info->config)); - audio_demuxer_stream_data_provider_->SetOnNoBuffersAvailableCallback( - std::move( - initialization_info.audio_stream_info->on_no_buffers_callback)); - audio_demuxer_stream_data_provider_->SetOnErrorCallback( - std::move(initialization_info.audio_stream_info->on_error_callback)); + audio_demuxer_stream_data_provider_->SetClient(std::move( + initialization_info.audio_stream_info->demuxer_stream_client)); audio_info = mojom::AudioStreamInitializationInfo::New( std::move(audio_receiver), mojom::AudioStreamInfo::New( @@ -117,11 +114,8 @@ base::BindOnce(&ReceiverSessionImpl::OnMojoDisconnect, weak_factory_.GetWeakPtr()), std::move(initialization_info.video_stream_info->config)); - video_demuxer_stream_data_provider_->SetOnNoBuffersAvailableCallback( - std::move( - initialization_info.video_stream_info->on_no_buffers_callback)); - video_demuxer_stream_data_provider_->SetOnErrorCallback( - std::move(initialization_info.video_stream_info->on_error_callback)); + video_demuxer_stream_data_provider_->SetClient(std::move( + initialization_info.video_stream_info->demuxer_stream_client)); video_info = mojom::VideoStreamInitializationInfo::New( std::move(video_receiver), mojom::VideoStreamInfo::New( @@ -163,11 +157,8 @@ if (audio_pipe_consumer) { if (!audio_demuxer_stream_data_provider_->config().Matches( initialization_info.audio_stream_info->config)) { - audio_demuxer_stream_data_provider_->SetOnNoBuffersAvailableCallback( - std::move( - initialization_info.audio_stream_info->on_no_buffers_callback)); - audio_demuxer_stream_data_provider_->SetOnErrorCallback( - std::move(initialization_info.audio_stream_info->on_error_callback)); + audio_demuxer_stream_data_provider_->SetClient(std::move( + initialization_info.audio_stream_info->demuxer_stream_client)); audio_demuxer_stream_data_provider_->OnNewStreamInfo( std::move(initialization_info.audio_stream_info->config), std::move(*audio_pipe_consumer)); @@ -180,11 +171,8 @@ if (video_pipe_consumer) { if (!video_demuxer_stream_data_provider_->config().Matches( initialization_info.video_stream_info->config)) { - video_demuxer_stream_data_provider_->SetOnNoBuffersAvailableCallback( - std::move( - initialization_info.video_stream_info->on_no_buffers_callback)); - video_demuxer_stream_data_provider_->SetOnErrorCallback( - std::move(initialization_info.video_stream_info->on_error_callback)); + video_demuxer_stream_data_provider_->SetClient(std::move( + initialization_info.video_stream_info->demuxer_stream_client)); video_demuxer_stream_data_provider_->OnNewStreamInfo( std::move(initialization_info.video_stream_info->config), std::move(*video_pipe_consumer));
diff --git a/components/cast_streaming/browser/rpc_demuxer_stream_handler.cc b/components/cast_streaming/browser/rpc_demuxer_stream_handler.cc index 70eb6ca..4aba944 100644 --- a/components/cast_streaming/browser/rpc_demuxer_stream_handler.cc +++ b/components/cast_streaming/browser/rpc_demuxer_stream_handler.cc
@@ -20,13 +20,12 @@ RpcDemuxerStreamHandler::RpcDemuxerStreamHandler( Client* client, HandleFactory handle_factory, - RpcProcessMessageCB message_processor) + RpcProcessMessageCB process_message_cb) : client_(client), handle_factory_(std::move(handle_factory)), - message_processor_(std::move(message_processor)), - weak_factory_(this) { + process_message_cb_(std::move(process_message_cb)) { DCHECK(handle_factory_); - DCHECK(message_processor_); + DCHECK(process_message_cb_); } RpcDemuxerStreamHandler::~RpcDemuxerStreamHandler() = default; @@ -38,83 +37,27 @@ // initialize the DemuxerStreams. if (audio_stream_handle != openscreen::cast::RpcMessenger::kInvalidHandle) { audio_message_processor_ = std::make_unique<MessageProcessor>( - client_, handle_factory_.Run(), audio_stream_handle, - MessageProcessor::Type::kAudio); + client_, process_message_cb_, handle_factory_.Run(), + audio_stream_handle, MessageProcessor::Type::kAudio); std::unique_ptr<openscreen::cast::RpcMessage> message = remoting::CreateMessageForDemuxerStreamInitialize( audio_message_processor_->local_handle()); - message_processor_.Run(audio_message_processor_->remote_handle(), - std::move(message)); + process_message_cb_.Run(audio_message_processor_->remote_handle(), + std::move(message)); } if (video_stream_handle != openscreen::cast::RpcMessenger::kInvalidHandle) { video_message_processor_ = std::make_unique<MessageProcessor>( - client_, handle_factory_.Run(), video_stream_handle, - MessageProcessor::Type::kVideo); + client_, process_message_cb_, handle_factory_.Run(), + video_stream_handle, MessageProcessor::Type::kVideo); std::unique_ptr<openscreen::cast::RpcMessage> message = remoting::CreateMessageForDemuxerStreamInitialize( video_message_processor_->local_handle()); - message_processor_.Run(video_message_processor_->remote_handle(), - std::move(message)); + process_message_cb_.Run(video_message_processor_->remote_handle(), + std::move(message)); } } -void RpcDemuxerStreamHandler::RequestMoreAudioBuffers() { - if (!audio_message_processor_) { - return; - } - - RequestMoreBuffers(audio_message_processor_.get()); -} - -void RpcDemuxerStreamHandler::RequestMoreVideoBuffers() { - if (!video_message_processor_) { - return; - } - - RequestMoreBuffers(video_message_processor_.get()); -} - -void RpcDemuxerStreamHandler::RequestMoreBuffers( - MessageProcessor* message_processor) { - if (message_processor->is_read_until_call_pending()) { - return; - } - - message_processor->set_read_until_call_pending(); - auto message = CreateMessageForDemuxerStreamReadUntil( - message_processor->local_handle(), - message_processor->total_frames_received() + kNumFramesInEachReadUntil); - message_processor_.Run(message_processor->remote_handle(), - std::move(message)); -} - -void RpcDemuxerStreamHandler::OnAudioError() { - if (!audio_message_processor_) { - return; - } - - OnError(audio_message_processor_.get()); -} - -void RpcDemuxerStreamHandler::OnVideoError() { - if (!video_message_processor_) { - return; - } - - OnError(video_message_processor_.get()); -} - -void RpcDemuxerStreamHandler::OnError(MessageProcessor* message_processor) { - auto message = CreateMessageForDemuxerStreamError(); - message_processor_.Run(message_processor->remote_handle(), - std::move(message)); -} - -base::WeakPtr<RpcDemuxerStreamHandler> RpcDemuxerStreamHandler::GetWeakPtr() { - return weak_factory_.GetWeakPtr(); -} - void RpcDemuxerStreamHandler::OnRpcInitializeCallback( openscreen::cast::RpcMessenger::Handle handle, absl::optional<media::AudioDecoderConfig> audio_config, @@ -152,22 +95,58 @@ } } +void RpcDemuxerStreamHandler::OnRpcBitstreamConverterEnabled( + openscreen::cast::RpcMessenger::Handle handle, + bool success) { + if (audio_message_processor_ && + handle == audio_message_processor_->local_handle()) { + audio_message_processor_->OnBitstreamConverterEnabled(success); + } else if (video_message_processor_ && + handle == video_message_processor_->local_handle()) { + video_message_processor_->OnBitstreamConverterEnabled(success); + } else { + LOG(WARNING) + << "OnRpcBitstreamConverterEnabled received for invalid handle"; + } +} + +base::WeakPtr<DemuxerStreamClient> RpcDemuxerStreamHandler::GetAudioClient() { + if (!audio_message_processor_) { + return nullptr; + } + + return audio_message_processor_->GetWeakPtr(); +} + +base::WeakPtr<DemuxerStreamClient> RpcDemuxerStreamHandler::GetVideoClient() { + if (!video_message_processor_) { + return nullptr; + } + + return video_message_processor_->GetWeakPtr(); +} + RpcDemuxerStreamHandler::Client::~Client() = default; RpcDemuxerStreamHandler::MessageProcessor::MessageProcessor( Client* client, + RpcProcessMessageCB process_message_cb, openscreen::cast::RpcMessenger::Handle local_handle, openscreen::cast::RpcMessenger::Handle remote_handle, Type type) : client_(client), + process_message_cb_(std::move(process_message_cb)), local_handle_(local_handle), remote_handle_(remote_handle), - type_(type) { + type_(type), + weak_factory_(this) { DCHECK(client_); DCHECK_NE(local_handle_, openscreen::cast::RpcMessenger::kInvalidHandle); DCHECK_NE(remote_handle_, openscreen::cast::RpcMessenger::kInvalidHandle); } +RpcDemuxerStreamHandler::MessageProcessor::~MessageProcessor() = default; + bool RpcDemuxerStreamHandler::MessageProcessor::OnRpcInitializeCallback( absl::optional<media::AudioDecoderConfig> audio_config, absl::optional<media::VideoDecoderConfig> video_config) { @@ -194,6 +173,7 @@ uint32_t total_frames_received) { if (!OnRpcInitializeCallback(std::move(audio_config), std::move(video_config))) { + LOG(WARNING) << "Failed to process OnRpcReadUntilCallback."; return false; } @@ -202,4 +182,43 @@ return true; } +void RpcDemuxerStreamHandler::MessageProcessor::OnBitstreamConverterEnabled( + bool success) { + if (!bitstream_converter_enabled_cb_) { + return; + } + + std::move(bitstream_converter_enabled_cb_).Run(success); +} + +base::WeakPtr<RpcDemuxerStreamHandler::MessageProcessor> +RpcDemuxerStreamHandler::MessageProcessor::GetWeakPtr() { + return weak_factory_.GetWeakPtr(); +} + +void RpcDemuxerStreamHandler::MessageProcessor::EnableBitstreamConverter( + BitstreamConverterEnabledCB cb) { + DCHECK(!bitstream_converter_enabled_cb_); + bitstream_converter_enabled_cb_ = std::move(cb); + + auto message = CreateMessageForDemuxerStreamEnableBitstreamConverter(); + process_message_cb_.Run(remote_handle(), std::move(message)); +} + +void RpcDemuxerStreamHandler::MessageProcessor::OnNoBuffersAvailable() { + if (is_read_until_call_pending()) { + return; + } + + set_read_until_call_pending(); + auto message = CreateMessageForDemuxerStreamReadUntil( + local_handle(), total_frames_received() + kNumFramesInEachReadUntil); + process_message_cb_.Run(remote_handle(), std::move(message)); +} + +void RpcDemuxerStreamHandler::MessageProcessor::OnError() { + auto message = CreateMessageForDemuxerStreamError(); + process_message_cb_.Run(remote_handle(), std::move(message)); +} + } // namespace cast_streaming::remoting
diff --git a/components/cast_streaming/browser/rpc_demuxer_stream_handler.h b/components/cast_streaming/browser/rpc_demuxer_stream_handler.h index 1a5b8a5..c1b2f1d 100644 --- a/components/cast_streaming/browser/rpc_demuxer_stream_handler.h +++ b/components/cast_streaming/browser/rpc_demuxer_stream_handler.h
@@ -9,6 +9,7 @@ #include "base/callback.h" #include "base/memory/weak_ptr.h" +#include "components/cast_streaming/browser/demuxer_stream_client.h" #include "components/cast_streaming/public/rpc_call_message_handler.h" #include "media/base/audio_decoder_config.h" #include "media/base/video_decoder_config.h" @@ -44,7 +45,7 @@ std::unique_ptr<openscreen::cast::RpcMessage>)>; RpcDemuxerStreamHandler(Client* client, HandleFactory handle_factory, - RpcProcessMessageCB message_processor); + RpcProcessMessageCB process_message_cb); ~RpcDemuxerStreamHandler() override; @@ -55,19 +56,17 @@ openscreen::cast::RpcMessenger::Handle audio_stream_handle, openscreen::cast::RpcMessenger::Handle video_stream_handle); - // To be called when no further buffers are available for reading to request - // more be sent. - void RequestMoreAudioBuffers(); - void RequestMoreVideoBuffers(); + // To be called when the RPC_DS_ENABLEBITSTREAMCONVERTER_CALLBACK message is + // received. + void OnRpcBitstreamConverterEnabled( + openscreen::cast::RpcMessenger::Handle handle, + bool success); - // Called when a DemuxerStream error occurs to inform the sender device. - void OnAudioError(); - void OnVideoError(); - - base::WeakPtr<RpcDemuxerStreamHandler> GetWeakPtr(); + base::WeakPtr<DemuxerStreamClient> GetAudioClient(); + base::WeakPtr<DemuxerStreamClient> GetVideoClient(); private: - class MessageProcessor { + class MessageProcessor : public DemuxerStreamClient { public: enum class Type { kUnknown = 0, kAudio, kVideo }; @@ -79,9 +78,11 @@ // OnRpcAcquiredDemuxer() call and used as the handle for sending messages // back to the sender. MessageProcessor(Client* client, + RpcProcessMessageCB process_message_cb, openscreen::cast::RpcMessenger::Handle local_handle, openscreen::cast::RpcMessenger::Handle remote_handle, Type type); + ~MessageProcessor() override; bool OnRpcInitializeCallback( absl::optional<media::AudioDecoderConfig> audio_config, @@ -90,6 +91,9 @@ absl::optional<media::AudioDecoderConfig> audio_config, absl::optional<media::VideoDecoderConfig> video_config, uint32_t total_frames_received); + void OnBitstreamConverterEnabled(bool success); + + base::WeakPtr<MessageProcessor> GetWeakPtr(); uint32_t total_frames_received() const { return total_frames_received_; } @@ -106,7 +110,12 @@ void set_read_until_call_pending() { is_read_until_call_pending_ = true; } private: + void EnableBitstreamConverter(BitstreamConverterEnabledCB cb) override; + void OnNoBuffersAvailable() override; + void OnError() override; + Client* client_; + RpcProcessMessageCB process_message_cb_; openscreen::cast::RpcMessenger::Handle local_handle_; openscreen::cast::RpcMessenger::Handle remote_handle_; Type type_ = Type::kUnknown; @@ -114,6 +123,11 @@ uint32_t total_frames_received_ = 0; bool is_read_until_call_pending_ = false; + + // Most recent callback for EnableBitstreamConverter(). + BitstreamConverterEnabledCB bitstream_converter_enabled_cb_; + + base::WeakPtrFactory<MessageProcessor> weak_factory_; }; // Helpers for the above methods of the same name. @@ -133,12 +147,10 @@ Client* const client_; HandleFactory handle_factory_; - RpcProcessMessageCB message_processor_; + RpcProcessMessageCB process_message_cb_; std::unique_ptr<MessageProcessor> audio_message_processor_; std::unique_ptr<MessageProcessor> video_message_processor_; - - base::WeakPtrFactory<RpcDemuxerStreamHandler> weak_factory_; }; } // namespace cast_streaming::remoting
diff --git a/components/cast_streaming/browser/rpc_demuxer_stream_handler_unittests.cc b/components/cast_streaming/browser/rpc_demuxer_stream_handler_unittests.cc index 8305553..3e0e57c 100644 --- a/components/cast_streaming/browser/rpc_demuxer_stream_handler_unittests.cc +++ b/components/cast_streaming/browser/rpc_demuxer_stream_handler_unittests.cc
@@ -62,6 +62,16 @@ EXPECT_EQ(handle, remote_handle); } +ACTION_P(CheckEnableBistreamConverterCall, remote_handle) { + const openscreen::cast::RpcMessenger::Handle handle = arg0; + const std::unique_ptr<openscreen::cast::RpcMessage>& rpc = arg1; + + ASSERT_TRUE(rpc); + EXPECT_EQ(rpc->proc(), + openscreen::cast::RpcMessage::RPC_DS_ENABLEBITSTREAMCONVERTER); + EXPECT_EQ(handle, remote_handle); +} + } // namespace class RpcDemuxerStreamHandlerTest : public testing::Test { @@ -119,6 +129,8 @@ MOCK_METHOD0(GetHandle, openscreen::cast::RpcMessenger::Handle()); + MOCK_METHOD1(OnBitstreamConverterEnabled, void(bool)); + void OnRpcInitializeCallback( openscreen::cast::RpcMessenger::Handle handle, absl::optional<media::AudioDecoderConfig> audio_config, @@ -139,6 +151,46 @@ total_frames_received); } + void RequestMoreAudioBuffers() { + auto client = stream_handler_.GetAudioClient(); + ASSERT_TRUE(!!client); + client->OnNoBuffersAvailable(); + } + + void RequestMoreVideoBuffers() { + auto client = stream_handler_.GetVideoClient(); + ASSERT_TRUE(!!client); + client->OnNoBuffersAvailable(); + } + + void OnAudioError() { + auto client = stream_handler_.GetAudioClient(); + ASSERT_TRUE(!!client); + client->OnError(); + } + + void OnVideoError() { + auto client = stream_handler_.GetVideoClient(); + ASSERT_TRUE(!!client); + client->OnError(); + } + + void EnableAudioBitstreamConverter() { + auto client = stream_handler_.GetAudioClient(); + ASSERT_TRUE(!!client); + client->EnableBitstreamConverter(base::BindOnce( + &RpcDemuxerStreamHandlerTest::OnBitstreamConverterEnabled, + base::Unretained(this))); + } + + void EnableVideoBitstreamConverter() { + auto client = stream_handler_.GetVideoClient(); + ASSERT_TRUE(!!client); + client->EnableBitstreamConverter(base::BindOnce( + &RpcDemuxerStreamHandlerTest::OnBitstreamConverterEnabled, + base::Unretained(this))); + } + openscreen::cast::RpcMessenger::Handle audio_remote_handle_ = 123; openscreen::cast::RpcMessenger::Handle video_remote_handle_ = 456; @@ -219,7 +271,7 @@ EXPECT_CALL(*this, SendMessage(_, _)) .WillOnce( CheckReadUntilCall(audio_remote_handle_, audio_local_handle_, 1)); - stream_handler_.RequestMoreAudioBuffers(); + RequestMoreAudioBuffers(); EXPECT_CALL(client_, OnNewAudioConfig(_)) .WillOnce([this](media::AudioDecoderConfig config) { @@ -230,7 +282,7 @@ EXPECT_CALL(*this, SendMessage(_, _)) .WillOnce( CheckReadUntilCall(audio_remote_handle_, audio_local_handle_, 17)); - stream_handler_.RequestMoreAudioBuffers(); + RequestMoreAudioBuffers(); } TEST_F(RpcDemuxerStreamHandlerTest, RequestMoreVideoBuffers) { @@ -243,7 +295,7 @@ EXPECT_CALL(*this, SendMessage(_, _)) .WillOnce( CheckReadUntilCall(video_remote_handle_, video_local_handle_, 12)); - stream_handler_.RequestMoreVideoBuffers(); + RequestMoreVideoBuffers(); EXPECT_CALL(client_, OnNewVideoConfig(_)) .WillOnce([this](media::VideoDecoderConfig config) { @@ -254,19 +306,37 @@ EXPECT_CALL(*this, SendMessage(_, _)) .WillOnce( CheckReadUntilCall(video_remote_handle_, video_local_handle_, 42)); - stream_handler_.RequestMoreVideoBuffers(); + RequestMoreVideoBuffers(); } TEST_F(RpcDemuxerStreamHandlerTest, OnAudioError) { EXPECT_CALL(*this, SendMessage(_, _)) .WillOnce(CheckOnErrorCall(audio_remote_handle_)); - stream_handler_.OnAudioError(); + OnAudioError(); } TEST_F(RpcDemuxerStreamHandlerTest, OnVideoError) { EXPECT_CALL(*this, SendMessage(_, _)) .WillOnce(CheckOnErrorCall(video_remote_handle_)); - stream_handler_.OnVideoError(); + OnVideoError(); +} + +TEST_F(RpcDemuxerStreamHandlerTest, OnEnableAudioBitstreamConverter) { + EXPECT_CALL(*this, SendMessage(_, _)) + .WillOnce(CheckEnableBistreamConverterCall(audio_remote_handle_)); + EnableAudioBitstreamConverter(); + + EXPECT_CALL(*this, OnBitstreamConverterEnabled(true)); + stream_handler_.OnRpcBitstreamConverterEnabled(audio_local_handle_, true); +} + +TEST_F(RpcDemuxerStreamHandlerTest, OnEnableVideoBitstreamConverter) { + EXPECT_CALL(*this, SendMessage(_, _)) + .WillOnce(CheckEnableBistreamConverterCall(video_remote_handle_)); + EnableVideoBitstreamConverter(); + + EXPECT_CALL(*this, OnBitstreamConverterEnabled(false)); + stream_handler_.OnRpcBitstreamConverterEnabled(video_local_handle_, false); } } // namespace cast_streaming::remoting
diff --git a/components/cast_streaming/browser/streaming_initialization_info.cc b/components/cast_streaming/browser/streaming_initialization_info.cc index f6e0ac0a..711473b 100644 --- a/components/cast_streaming/browser/streaming_initialization_info.cc +++ b/components/cast_streaming/browser/streaming_initialization_info.cc
@@ -28,32 +28,18 @@ StreamingInitializationInfo::AudioStreamInfo::AudioStreamInfo( media::AudioDecoderConfig audio_config, openscreen::cast::Receiver* cast_receiver) - : AudioStreamInfo(std::move(audio_config), - cast_receiver, - base::RepeatingClosure(), - base::OnceClosure()) {} + : AudioStreamInfo(std::move(audio_config), cast_receiver, nullptr) {} StreamingInitializationInfo::AudioStreamInfo::AudioStreamInfo( media::AudioDecoderConfig audio_config, openscreen::cast::Receiver* cast_receiver, - base::RepeatingClosure on_no_buffers_cb, - base::OnceClosure on_error_cb) + base::WeakPtr<DemuxerStreamClient> ds_client) : config(std::move(audio_config)), receiver(cast_receiver), - on_no_buffers_callback(std::move(on_no_buffers_cb)), - on_error_callback(std::move(on_error_cb)) {} + demuxer_stream_client(std::move(ds_client)) {} StreamingInitializationInfo::AudioStreamInfo::AudioStreamInfo( - const StreamingInitializationInfo::AudioStreamInfo& other) { - auto& old = const_cast<AudioStreamInfo&>(other); - auto cb_pair = base::SplitOnceCallback(std::move(old.on_error_callback)); - old.on_error_callback = std::move(cb_pair.first); - - config = other.config; - receiver = other.receiver; - on_no_buffers_callback = other.on_no_buffers_callback; - on_error_callback = std::move(cb_pair.second); -} + const StreamingInitializationInfo::AudioStreamInfo& other) = default; StreamingInitializationInfo::AudioStreamInfo::~AudioStreamInfo() = default; @@ -62,32 +48,18 @@ StreamingInitializationInfo::VideoStreamInfo::VideoStreamInfo( media::VideoDecoderConfig video_config, openscreen::cast::Receiver* cast_receiver) - : VideoStreamInfo(std::move(video_config), - cast_receiver, - base::RepeatingClosure(), - base::OnceClosure()) {} + : VideoStreamInfo(std::move(video_config), cast_receiver, nullptr) {} StreamingInitializationInfo::VideoStreamInfo::VideoStreamInfo( media::VideoDecoderConfig video_config, openscreen::cast::Receiver* cast_receiver, - base::RepeatingClosure on_no_buffers_cb, - base::OnceClosure on_error_cb) + base::WeakPtr<DemuxerStreamClient> ds_client) : config(std::move(video_config)), receiver(cast_receiver), - on_no_buffers_callback(std::move(on_no_buffers_cb)), - on_error_callback(std::move(on_error_cb)) {} + demuxer_stream_client(std::move(ds_client)) {} StreamingInitializationInfo::VideoStreamInfo::VideoStreamInfo( - const StreamingInitializationInfo::VideoStreamInfo& other) { - auto& old = const_cast<VideoStreamInfo&>(other); - auto cb_pair = base::SplitOnceCallback(std::move(old.on_error_callback)); - old.on_error_callback = std::move(cb_pair.first); - - config = other.config; - receiver = other.receiver; - on_no_buffers_callback = other.on_no_buffers_callback; - on_error_callback = std::move(cb_pair.second); -} + const StreamingInitializationInfo::VideoStreamInfo& other) = default; StreamingInitializationInfo::VideoStreamInfo::~VideoStreamInfo() = default;
diff --git a/components/cast_streaming/browser/streaming_initialization_info.h b/components/cast_streaming/browser/streaming_initialization_info.h index 188f2faf..2fd89a7 100644 --- a/components/cast_streaming/browser/streaming_initialization_info.h +++ b/components/cast_streaming/browser/streaming_initialization_info.h
@@ -5,8 +5,8 @@ #ifndef COMPONENTS_CAST_STREAMING_BROWSER_STREAMING_INITIALIZATION_INFO_H_ #define COMPONENTS_CAST_STREAMING_BROWSER_STREAMING_INITIALIZATION_INFO_H_ -#include "base/callback.h" -#include "base/callback_forward.h" +#include "base/memory/weak_ptr.h" +#include "components/cast_streaming/browser/demuxer_stream_client.h" #include "media/base/audio_decoder_config.h" #include "media/base/video_decoder_config.h" #include "third_party/abseil-cpp/absl/types/optional.h" @@ -27,8 +27,7 @@ openscreen::cast::Receiver* cast_receiver); AudioStreamInfo(media::AudioDecoderConfig audio_config, openscreen::cast::Receiver* cast_receiver, - base::RepeatingClosure on_no_buffers_cb, - base::OnceClosure on_error_cb); + base::WeakPtr<DemuxerStreamClient> ds_client); AudioStreamInfo(); AudioStreamInfo(const AudioStreamInfo& other); ~AudioStreamInfo(); @@ -40,11 +39,9 @@ // duration of the streaming session. openscreen::cast::Receiver* receiver; - // Callback to be called when no buffers are available for reading. - base::RepeatingClosure on_no_buffers_callback; - - // Callback to be called when a non-recoverable error occurs. - base::OnceClosure on_error_callback; + // Client with methods to be called when the DemuxerStream requires an + // action be executed. + base::WeakPtr<DemuxerStreamClient> demuxer_stream_client; }; struct VideoStreamInfo { @@ -52,8 +49,7 @@ openscreen::cast::Receiver* cast_receiver); VideoStreamInfo(media::VideoDecoderConfig video_config, openscreen::cast::Receiver* cast_receiver, - base::RepeatingClosure on_no_buffers_cb, - base::OnceClosure on_error_cb); + base::WeakPtr<DemuxerStreamClient> ds_client); VideoStreamInfo(); VideoStreamInfo(const VideoStreamInfo& other); ~VideoStreamInfo(); @@ -65,11 +61,9 @@ // duration of the streaming session. openscreen::cast::Receiver* receiver; - // Callback to be called when no buffers are available for reading. - base::RepeatingClosure on_no_buffers_callback; - - // Callback to be called when a non-recoverable error occurs. - base::OnceClosure on_error_callback; + // Client with methods to be called when the DemuxerStream requires an + // action be executed. + base::WeakPtr<DemuxerStreamClient> demuxer_stream_client; }; StreamingInitializationInfo(
diff --git a/components/cast_streaming/public/mojom/demuxer_connector.mojom b/components/cast_streaming/public/mojom/demuxer_connector.mojom index d7c8ae0a..00ef810 100644 --- a/components/cast_streaming/public/mojom/demuxer_connector.mojom +++ b/components/cast_streaming/public/mojom/demuxer_connector.mojom
@@ -47,6 +47,10 @@ // returns. GetBuffer() => (AudioStreamInfo? stream_info, media.mojom.DecoderBuffer buffer); + + // Requests that the data source providing audio buffers enable its bitstream + // converter. Returns whether the operation was successful. + EnableBitstreamConverter() => (bool success); }; // Provides a "pull" mechanism to request DecoderBuffer frames of video data. @@ -56,6 +60,9 @@ // As AudioBufferRequester::GetBuffer() above. GetBuffer() => (VideoStreamInfo? stream_info, media.mojom.DecoderBuffer buffer); + + // As AudioBufferRequester::EnableBitstreamConverter() above. + EnableBitstreamConverter() => (bool success); }; // Initialization information for an audio DemuxerStream.
diff --git a/components/cast_streaming/public/remoting_message_factories.cc b/components/cast_streaming/public/remoting_message_factories.cc index da597e17..2dcb3a0 100644 --- a/components/cast_streaming/public/remoting_message_factories.cc +++ b/components/cast_streaming/public/remoting_message_factories.cc
@@ -150,6 +150,13 @@ } std::unique_ptr<openscreen::cast::RpcMessage> +CreateMessageForDemuxerStreamEnableBitstreamConverter() { + auto rpc = std::make_unique<openscreen::cast::RpcMessage>(); + rpc->set_proc(openscreen::cast::RpcMessage::RPC_DS_ENABLEBITSTREAMCONVERTER); + return rpc; +} + +std::unique_ptr<openscreen::cast::RpcMessage> CreateMessageForDemuxerStreamError() { auto rpc = std::make_unique<openscreen::cast::RpcMessage>(); rpc->set_proc(openscreen::cast::RpcMessage::RPC_DS_ONERROR);
diff --git a/components/cast_streaming/public/remoting_message_factories.h b/components/cast_streaming/public/remoting_message_factories.h index 980de9fa..9d00b066 100644 --- a/components/cast_streaming/public/remoting_message_factories.h +++ b/components/cast_streaming/public/remoting_message_factories.h
@@ -80,6 +80,9 @@ uint32_t buffers_requested); std::unique_ptr<openscreen::cast::RpcMessage> +CreateMessageForDemuxerStreamEnableBitstreamConverter(); + +std::unique_ptr<openscreen::cast::RpcMessage> CreateMessageForDemuxerStreamError(); } // namespace remoting
diff --git a/components/cast_streaming/public/remoting_message_factories_unittest.cc b/components/cast_streaming/public/remoting_message_factories_unittest.cc index 59b18a1..17c825a 100644 --- a/components/cast_streaming/public/remoting_message_factories_unittest.cc +++ b/components/cast_streaming/public/remoting_message_factories_unittest.cc
@@ -208,6 +208,13 @@ EXPECT_EQ(message.callback_handle(), kTestHandle); } +TEST_F(RemotingMessageFactoriesTest, + CreateMessageForDemuxerStreamEnableBitstreamConverter) { + const auto rpc = CreateMessageForDemuxerStreamEnableBitstreamConverter(); + EXPECT_EQ(rpc->proc(), + openscreen::cast::RpcMessage::RPC_DS_ENABLEBITSTREAMCONVERTER); +} + TEST_F(RemotingMessageFactoriesTest, CreateMessageForDemuxerStreamError) { const auto rpc = CreateMessageForDemuxerStreamError(); EXPECT_EQ(rpc->proc(), openscreen::cast::RpcMessage::RPC_DS_ONERROR);
diff --git a/components/cast_streaming/renderer/cast_streaming_demuxer.cc b/components/cast_streaming/renderer/cast_streaming_demuxer.cc index 50e2610d..2016431 100644 --- a/components/cast_streaming/renderer/cast_streaming_demuxer.cc +++ b/components/cast_streaming/renderer/cast_streaming_demuxer.cc
@@ -182,6 +182,8 @@ } } + void OnBitstreamConverterEnabled(bool success) { NOTIMPLEMENTED_LOG_ONCE(); } + // DemuxerStream partial implementation. void Read(ReadCB read_cb) final { DVLOG(3) << __func__; @@ -212,6 +214,12 @@ buffer_reader_->ReadBufferAsync(); } + void EnableBitstreamConverter() final { + remote_->EnableBitstreamConverter( + base::BindOnce(&CastStreamingDemuxerStream::OnBitstreamConverterEnabled, + weak_factory_.GetWeakPtr())); + } + media::StreamLiveness liveness() const final { return media::StreamLiveness::kLive; }