media/remoting/stream_provider.h

   1 // Copyright 2017 The Chromium Authors
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #ifndef MEDIA_REMOTING_STREAM_PROVIDER_H_
   6 #define MEDIA_REMOTING_STREAM_PROVIDER_H_
   7
   8 #include "base/containers/circular_deque.h"
   9 #include "base/functional/callback_forward.h"
  10 #include "base/memory/raw_ptr.h"
  11 #include "base/memory/scoped_refptr.h"
  12 #include "base/memory/weak_ptr.h"
  13 #include "base/task/sequenced_task_runner.h"
  14 #include "base/task/sequenced_task_runner_helpers.h"
  15 #include "base/task/single_thread_task_runner.h"
  16 #include "media/base/audio_decoder_config.h"
  17 #include "media/base/demuxer.h"
  18 #include "media/base/demuxer_stream.h"
  19 #include "media/base/video_decoder_config.h"
  20 #include "media/mojo/mojom/remoting.mojom.h"
  21 #include "mojo/public/cpp/bindings/receiver.h"
  22 #include "mojo/public/cpp/bindings/remote.h"
  23 #include "third_party/openscreen/src/cast/streaming/remoting.pb.h"
  24
  25 namespace base {
  26 class SingleThreadTaskRunner;
  27 }  // namespace base
  28
  29 namespace openscreen {
  30 namespace cast {
  31 class RpcMessenger;
  32 }
  33 }  // namespace openscreen
  34
  35 namespace media {
  36
  37 class MojoDecoderBufferReader;
  38
  39 namespace remoting {
  40
  41 class ReceiverController;
  42
  43 // The media stream provider for Media Remoting receiver.
  44 class StreamProvider final : public Demuxer {
  45  public:
  46   StreamProvider(
  47       ReceiverController* receiver_controller,
  48       const scoped_refptr<base::SequencedTaskRunner>& media_task_runner);
  49
  50   // Demuxer implementation.
  51   std::vector<DemuxerStream*> GetAllStreams() override;
  52   std::string GetDisplayName() const override;
  53   DemuxerType GetDemuxerType() const override;
  54   void Initialize(DemuxerHost* host, PipelineStatusCallback status_cb) override;
  55   void AbortPendingReads() override;
  56   void StartWaitingForSeek(base::TimeDelta seek_time) override;
  57   void CancelPendingSeek(base::TimeDelta seek_time) override;
  58   void Seek(base::TimeDelta time, PipelineStatusCallback status_cb) override;
  59   bool IsSeekable() const override;
  60   void Stop() override;
  61   base::TimeDelta GetStartTime() const override;
  62   base::Time GetTimelineOffset() const override;
  63   int64_t GetMemoryUsage() const override;
  64   absl::optional<container_names::MediaContainerName> GetContainerForMetrics()
  65       const override;
  66   void OnEnabledAudioTracksChanged(const std::vector<MediaTrack::Id>& track_ids,
  67                                    base::TimeDelta curr_time,
  68                                    TrackChangeCB change_completed_cb) override;
  69   void OnSelectedVideoTrackChanged(const std::vector<MediaTrack::Id>& track_ids,
  70                                    base::TimeDelta curr_time,
  71                                    TrackChangeCB change_completed_cb) override;
  72   void SetPlaybackRate(double rate) override {}
  73
  74  protected:
  75   // Deletion is only allowed via Destroy().
  76   ~StreamProvider() override;
  77
  78  private:
  79   // An implementation of media::DemuxerStream on Media Remoting receiver.
  80   // Receives data from mojo data pipe, and returns one frame or/and status when
  81   // Read() is called.
  82   class MediaStream final : public DemuxerStream,
  83                             public mojom::RemotingDataStreamReceiver {
  84    public:
  85     using UniquePtr =
  86         std::unique_ptr<MediaStream, std::function<void(MediaStream*)>>;
  87
  88     // MediaStream should be created on the main thread to be able to get unique
  89     // handle ID from |rpc_messenger_|.
  90     static void CreateOnMainThread(
  91         openscreen::cast::RpcMessenger* rpc_messenger,
  92         Type type,
  93         int32_t handle,
  94         const scoped_refptr<base::SequencedTaskRunner>& media_task_runner,
  95         base::OnceCallback<void(MediaStream::UniquePtr)> callback);
  96
  97     // In order to destroy members in the right thread, MediaStream has to use
  98     // DestructionHelper() to destroy itself.
  99     static void DestructionHelper(MediaStream* stream);
 100
 101     MediaStream(
 102         openscreen::cast::RpcMessenger* rpc_messenger,
 103         Type type,
 104         int32_t remote_handle,
 105         const scoped_refptr<base::SequencedTaskRunner>& media_task_runner);
 106
 107     // DemuxerStream implementation.
 108     void Read(uint32_t count, ReadCB read_cb) override;
 109     AudioDecoderConfig audio_decoder_config() override;
 110     VideoDecoderConfig video_decoder_config() override;
 111     DemuxerStream::Type type() const override;
 112     StreamLiveness liveness() const override;
 113     bool SupportsConfigChanges() override;
 114
 115     void Initialize(base::OnceClosure init_done_cb);
 116
 117     mojo::PendingRemote<mojom::RemotingDataStreamReceiver>
 118     BindNewPipeAndPassRemote() {
 119       return receiver_.BindNewPipeAndPassRemote();
 120     }
 121
 122    private:
 123     friend class base::DeleteHelper<MediaStream>;  // For using DeleteSoon().
 124     // For testing.
 125     friend class StreamProviderTest;
 126
 127     // Prevent from unique_ptr using ~MediaStream() to destroy MediaStream
 128     // instances. Use DestructionHelper() as the custom deleter with unique_ptr
 129     // to destroy MediaStream instances.
 130     ~MediaStream() override;
 131
 132     void Destroy();
 133
 134     // Send RPC message on |main_task_runner_|.
 135     void SendRpcMessageOnMainThread(
 136         std::unique_ptr<openscreen::cast::RpcMessage> message);
 137
 138     // mojom::RemotingDataStreamReceiver implementation.
 139     void InitializeDataPipe(
 140         mojo::ScopedDataPipeConsumerHandle data_pipe) override;
 141     void ReceiveFrame(uint32_t count, mojom::DecoderBufferPtr buffer) override;
 142     void FlushUntil(uint32_t count) override;
 143
 144     // RPC messages handlers.
 145     void OnReceivedRpc(std::unique_ptr<openscreen::cast::RpcMessage> message);
 146     void OnInitializeCallback(
 147         std::unique_ptr<openscreen::cast::RpcMessage> message);
 148     void OnReadUntilCallback(
 149         std::unique_ptr<openscreen::cast::RpcMessage> message);
 150
 151     // Issues the ReadUntil RPC message when read is pending and buffer is
 152     // empty.
 153     void SendReadUntil();
 154
 155     // Run |init_done_callback_| when MojoDecoderBufferReader is created and
 156     // received RPC_DS_INITIALIZE_CALLBACK
 157     void CompleteInitialize();
 158
 159     // Append a frame into |buffers_|.
 160     void AppendBuffer(uint32_t count, scoped_refptr<DecoderBuffer> buffer);
 161
 162     // Run and reset the read callback.
 163     void CompleteRead(DemuxerStream::Status status);
 164
 165     // Update the audio/video decoder config. When config changes in the mid
 166     // stream, the new config will be stored in |next_audio_decoder_config_|.
 167     // Old config will be dropped when all associated frames are consumed.
 168     void UpdateAudioConfig(
 169         const openscreen::cast::AudioDecoderConfig& audio_message);
 170     void UpdateVideoConfig(
 171         const openscreen::cast::VideoDecoderConfig& video_message);
 172
 173     // Called when any error occurs.
 174     void OnError(const std::string& error);
 175
 176     scoped_refptr<base::SingleThreadTaskRunner> main_task_runner_;
 177     scoped_refptr<base::SequencedTaskRunner> media_task_runner_;
 178
 179     const raw_ptr<openscreen::cast::RpcMessenger> rpc_messenger_;
 180     const Type type_;
 181     const int remote_handle_;
 182     const int rpc_handle_;
 183
 184     // Set when Initialize() is called.
 185     base::OnceClosure init_done_callback_;
 186
 187     // The frame count of the frame to be returned on the next Read call. It
 188     // will be increased whenever a frame is read. It will be updated when
 189     // FlushUntil() is called.
 190     uint32_t current_frame_count_ = 0;
 191
 192     // One plus the last frame count received over RTP. Used for continuity
 193     // check.
 194     uint32_t buffered_frame_count_ = 0;
 195
 196     // The total number of frames received from the sender side. It will be used
 197     // as the base value for sending ReadUntil() to request more frames and be
 198     // updated in OnReadUntilCallback() which would get the message that
 199     // contains how many frames are sent.
 200     uint32_t total_received_frame_count_ = 0;
 201
 202     // Indicates whether a ReadUntil RPC message was sent without receiving the
 203     // ReadUntilCallback message yet.
 204     bool read_until_sent_ = false;
 205
 206     // Indicates whether RPC_DS_INITIALIZE_CALLBACK received.
 207     bool rpc_initialized_ = false;
 208
 209     // Set when Read() is called. Run only once when read completes.
 210     ReadCB read_complete_callback_;
 211
 212     // The frame data would be sent via Mojo IPC as MojoDecoderBuffer. When a
 213     // frame is sent to |this| from host by calling ReceiveFrame(),
 214     // |decoder_buffer_reader_| is used to read the frame date from data pipe.
 215     std::unique_ptr<MojoDecoderBufferReader> decoder_buffer_reader_;
 216
 217     base::circular_deque<scoped_refptr<DecoderBuffer>> buffers_;
 218
 219     // Current audio/video config.
 220     AudioDecoderConfig audio_decoder_config_;
 221     VideoDecoderConfig video_decoder_config_;
 222
 223     // Stores the new audio/video config when config changes.
 224     AudioDecoderConfig next_audio_decoder_config_;
 225     VideoDecoderConfig next_video_decoder_config_;
 226
 227     mojo::Receiver<mojom::RemotingDataStreamReceiver> receiver_{this};
 228
 229     base::WeakPtr<MediaStream> media_weak_this_;
 230     base::WeakPtrFactory<MediaStream> media_weak_factory_{this};
 231   };
 232
 233   friend std::default_delete<StreamProvider>;
 234   friend class base::DeleteHelper<StreamProvider>;  // For using DeleteSoon().
 235
 236   // For testing.
 237   friend class StreamProviderTest;
 238
 239   void Destroy();
 240
 241   // RPC messages handlers.
 242   void OnReceivedRpc(std::unique_ptr<openscreen::cast::RpcMessage> message);
 243   void OnAcquireDemuxer(std::unique_ptr<openscreen::cast::RpcMessage> message);
 244
 245   // Called when audio/video stream is created and initialized.
 246   void InitializeDataPipe();
 247   void OnAudioStreamCreated(MediaStream::UniquePtr stream);
 248   void OnVideoStreamCreated(MediaStream::UniquePtr stream);
 249   void OnAudioStreamInitialized();
 250   void OnVideoStreamInitialized();
 251   void CompleteInitialize();
 252
 253   scoped_refptr<base::SingleThreadTaskRunner> main_task_runner_;
 254   scoped_refptr<base::SequencedTaskRunner> media_task_runner_;
 255   const raw_ptr<ReceiverController> receiver_controller_;
 256   const raw_ptr<openscreen::cast::RpcMessenger> rpc_messenger_;
 257   MediaStream::UniquePtr audio_stream_;
 258   MediaStream::UniquePtr video_stream_;
 259   bool has_audio_{false};
 260   bool has_video_{false};
 261   bool audio_stream_initialized_{false};
 262   bool video_stream_initialized_{false};
 263
 264   // Set when Initialize() is called, and will run when both video and audio
 265   // streams are initialized or error occurs.
 266   PipelineStatusCallback init_done_callback_;
 267
 268   base::WeakPtr<StreamProvider> media_weak_this_;
 269   base::WeakPtrFactory<StreamProvider> media_weak_factory_{this};
 270 };
 271
 272 }  // namespace remoting
 273 }  // namespace media
 274
 275 namespace std {
 276
 277 // Specialize std::default_delete to call Destroy().
 278 template <>
 279 struct default_delete<media::remoting::StreamProvider> {
 280   constexpr default_delete() = default;
 281
 282   template <typename U,
 283             typename = typename std::enable_if<std::is_convertible<
 284                 U*,
 285                 media::remoting::StreamProvider*>::value>::type>
 286   explicit default_delete(const default_delete<U>& d) {}
 287
 288   void operator()(media::remoting::StreamProvider* ptr) const;
 289 };
 290
 291 }  // namespace std
 292
 293 #endif  // MEDIA_REMOTING_STREAM_PROVIDER_H_