media/filters/manifest_demuxer.h

   1 // Copyright 2023 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_FILTERS_MANIFEST_DEMUXER_H_
   6 #define MEDIA_FILTERS_MANIFEST_DEMUXER_H_
   7
   8 #include <vector>
   9
  10 #include "base/cancelable_callback.h"
  11 #include "base/containers/flat_map.h"
  12 #include "base/memory/scoped_refptr.h"
  13 #include "base/task/sequenced_task_runner.h"
  14 #include "base/threading/sequence_bound.h"
  15 #include "base/time/time.h"
  16 #include "media/base/container_names.h"
  17 #include "media/base/demuxer.h"
  18 #include "media/base/demuxer_stream.h"
  19 #include "media/base/media_export.h"
  20 #include "media/base/media_log.h"
  21 #include "media/base/media_track.h"
  22 #include "media/base/pipeline_status.h"
  23 #include "media/filters/chunk_demuxer.h"
  24 #include "media/filters/hls_data_source_provider.h"
  25 #include "third_party/abseil-cpp/absl/types/optional.h"
  26
  27 namespace media {
  28
  29 // Declared and defined in manifest_demuxer.cc.
  30 class ManifestDemuxerStream;
  31
  32 // This class provides an interface for ManifestDemuxer::Engine to talk to
  33 // ManifestDemuxer about the internal ChunkDemuxer, without allowing it to call
  34 // any of the methods that the pipeline uses, like Seek or Stop.
  35 class MEDIA_EXPORT ManifestDemuxerEngineHost {
  36  public:
  37   virtual ~ManifestDemuxerEngineHost() {}
  38
  39   // Adds a new role to the chunk demuxer, and returns true if it succeeded.
  40   virtual bool AddRole(base::StringPiece role,
  41                        std::string container,
  42                        std::string codec) = 0;
  43
  44   // Removes a role (on the media thread) to ensure that there are no
  45   // media-thread-bound weak references.
  46   virtual void RemoveRole(base::StringPiece role) = 0;
  47
  48   // Sets the sequence mode flag for a |role| which has been created with
  49   // `AddRole`
  50   virtual void SetSequenceMode(base::StringPiece role, bool sequence_mode) = 0;
  51
  52   // Sets the chunk demuxer duration.
  53   virtual void SetDuration(double duration) = 0;
  54
  55   // Get the ranges that chunk demuxer has loaded, which allow seeking to avoid
  56   // fetching new data, if the seek is into a loaded range already.
  57   virtual Ranges<base::TimeDelta> GetBufferedRanges(base::StringPiece role) = 0;
  58
  59   // Removes all data from the chunk demuxer between `start` and `end`.
  60   virtual void Remove(base::StringPiece role,
  61                       base::TimeDelta start,
  62                       base::TimeDelta end) = 0;
  63
  64   // Removes all data from the chunk demuxer between |start| and |end| for a
  65   // given role, and resets the parser state while updating the parse offset.
  66   virtual void RemoveAndReset(base::StringPiece role,
  67                               base::TimeDelta start,
  68                               base::TimeDelta end,
  69                               base::TimeDelta* offset) = 0;
  70
  71   // Checks to see if we're parsing a media segment and if it is the case, then
  72   // resets the group start timestamp.
  73   virtual void SetGroupStartIfParsingAndSequenceMode(base::StringPiece role,
  74                                                      base::TimeDelta start) = 0;
  75
  76   // Evicts frames from chunk demuxer.
  77   virtual void EvictCodedFrames(base::StringPiece role,
  78                                 base::TimeDelta time,
  79                                 size_t data_size) = 0;
  80
  81   // Appends data to the chunk demuxer, parses it, and returns true if the new
  82   // data was parsed successfully.
  83   virtual bool AppendAndParseData(base::StringPiece role,
  84                                   base::TimeDelta start,
  85                                   base::TimeDelta end,
  86                                   base::TimeDelta* offset,
  87                                   const uint8_t* data,
  88                                   size_t data_size) = 0;
  89
  90   // Allow seeking from within an implementation.
  91   virtual void RequestSeek(base::TimeDelta time) = 0;
  92
  93   // Handle errors.
  94   virtual void OnError(PipelineStatus error) = 0;
  95
  96   virtual void SetGroupStartTimestamp(base::StringPiece role,
  97                                       base::TimeDelta time) = 0;
  98
  99   virtual void SetEndOfStream() = 0;
 100   virtual void UnsetEndOfStream() = 0;
 101 };
 102
 103 // A Demuxer designed to allow implementation of media demuxers which don't
 104 // rely on raw media data alone, such as HLS or DASH. This demuxer owns an
 105 // implementation of an engine and handles the seeking and event dispatching
 106 // for the engine so that it can focus on keeping internal manifest states up
 107 // to date.
 108 class MEDIA_EXPORT ManifestDemuxer : public Demuxer, ManifestDemuxerEngineHost {
 109  public:
 110   using DelayCallback = base::OnceCallback<void(base::TimeDelta)>;
 111
 112   // Seeks respond with either:
 113   //  - an error
 114   //  - kIsReady: buffers are full and chunk demuxer can seek normally.
 115   //  - kNeedsData: buffers are empty and need more data before chunk demuxer
 116   //                would otherwise finish seeking.
 117   enum class SeekState {
 118     kIsReady,
 119     kNeedsData,
 120   };
 121   using SeekResponse = PipelineStatus::Or<SeekState>;
 122   using SeekCallback = base::OnceCallback<void(SeekResponse)>;
 123
 124   class Engine {
 125    public:
 126     virtual ~Engine() {}
 127
 128     // Set for the engine, such as fetching manifests or content.
 129     virtual void Initialize(ManifestDemuxerEngineHost* demuxer,
 130                             PipelineStatusCallback status_cb) = 0;
 131
 132     // Get the name of the engine impl.
 133     virtual std::string GetName() const = 0;
 134
 135     // A tick signal indicating that the state of the engine should be
 136     // checked. `time` is the current player time. `playback_rate` is the
 137     // current playback rate. `loaded_ranges` is the current set of loaded
 138     // ranges in the chunk demuxer. `cb` should be called with the amount of
 139     // time to delay until the next event is requested.
 140     virtual void OnTimeUpdate(base::TimeDelta time,
 141                               double playback_rate,
 142                               DelayCallback cb) = 0;
 143
 144     // A synchronous seek, mostly intended to reset parts of the chunk
 145     // demuxer. returns whether the chunk demuxer needs more data.
 146     virtual void Seek(base::TimeDelta time, SeekCallback cb) = 0;
 147
 148     // Start waiting for seek, usually means canceling outstanding events
 149     // and network fetches.
 150     virtual void StartWaitingForSeek() = 0;
 151
 152     // Abort any pending reads, parses, or network requests.
 153     virtual void AbortPendingReads() = 0;
 154
 155     // Returns whether this engine supports seeking. Some live stream content
 156     // can't be seeked.
 157     virtual bool IsSeekable() const = 0;
 158
 159     // Gets the memory usage of the engine.
 160     virtual int64_t GetMemoryUsage() const = 0;
 161
 162     // Stop demuxing and clean up pending CBs.
 163     virtual void Stop() = 0;
 164   };
 165
 166   // ManifestDemuxer takes and keeps ownership of `impl` for the lifetime of
 167   // both.
 168   ManifestDemuxer(scoped_refptr<base::SequencedTaskRunner> media_task_runner,
 169                   base::RepeatingCallback<void(base::TimeDelta)> request_seek,
 170                   std::unique_ptr<Engine> impl,
 171                   MediaLog* media_log);
 172
 173   ~ManifestDemuxer() override;
 174
 175   // `media::Demuxer` implementation
 176   std::vector<DemuxerStream*> GetAllStreams() override;
 177   std::string GetDisplayName() const override;
 178   DemuxerType GetDemuxerType() const override;
 179   void Initialize(DemuxerHost* host, PipelineStatusCallback status_cb) override;
 180   void AbortPendingReads() override;
 181   void StartWaitingForSeek(base::TimeDelta seek_time) override;
 182   void CancelPendingSeek(base::TimeDelta seek_time) override;
 183   void Seek(base::TimeDelta time, PipelineStatusCallback status_cb) override;
 184   bool IsSeekable() const override;
 185   void Stop() override;
 186   base::TimeDelta GetStartTime() const override;
 187   base::Time GetTimelineOffset() const override;
 188   int64_t GetMemoryUsage() const override;
 189   void SetPlaybackRate(double rate) override;
 190   absl::optional<container_names::MediaContainerName> GetContainerForMetrics()
 191       const override;
 192
 193   void OnEnabledAudioTracksChanged(const std::vector<MediaTrack::Id>& track_ids,
 194                                    base::TimeDelta curr_time,
 195                                    TrackChangeCB change_completed_cb) override;
 196   void OnSelectedVideoTrackChanged(const std::vector<MediaTrack::Id>& track_ids,
 197                                    base::TimeDelta curr_time,
 198                                    TrackChangeCB change_completed_cb) override;
 199
 200   // `ManifestDemuxerEngineHost` implementation
 201   bool AddRole(base::StringPiece role,
 202                std::string container,
 203                std::string codec) override;
 204   void RemoveRole(base::StringPiece role) override;
 205   void SetSequenceMode(base::StringPiece role, bool sequence_mode) override;
 206   void SetDuration(double duration) override;
 207   Ranges<base::TimeDelta> GetBufferedRanges(base::StringPiece role) override;
 208   void Remove(base::StringPiece role,
 209               base::TimeDelta start,
 210               base::TimeDelta end) override;
 211   void RemoveAndReset(base::StringPiece role,
 212                       base::TimeDelta start,
 213                       base::TimeDelta end,
 214                       base::TimeDelta* offset) override;
 215   void SetGroupStartIfParsingAndSequenceMode(base::StringPiece role,
 216                                              base::TimeDelta start) override;
 217   void EvictCodedFrames(base::StringPiece role,
 218                         base::TimeDelta time,
 219                         size_t data_size) override;
 220   bool AppendAndParseData(base::StringPiece role,
 221                           base::TimeDelta start,
 222                           base::TimeDelta end,
 223                           base::TimeDelta* offset,
 224                           const uint8_t* data,
 225                           size_t data_size) override;
 226   void OnError(PipelineStatus status) override;
 227   void RequestSeek(base::TimeDelta time) override;
 228   void SetGroupStartTimestamp(base::StringPiece role,
 229                               base::TimeDelta time) override;
 230   void SetEndOfStream() override;
 231   void UnsetEndOfStream() override;
 232
 233   // Allow unit tests to grab the chunk demuxer.
 234   ChunkDemuxer* GetChunkDemuxerForTesting();
 235   bool has_pending_seek_for_testing() const { return !pending_seek_.is_null(); }
 236   base::TimeDelta get_media_time_for_testing() const { return media_time_; }
 237   bool has_pending_event_for_testing() const { return has_pending_event_; }
 238   bool has_next_task_for_testing() const {
 239     return !cancelable_next_event_.IsCancelled();
 240   }
 241
 242  private:
 243   // This wrapper class allows us to capture the results of Read() and use
 244   // DecoderBuffer timestamps to update the current media time within the
 245   // loaded buffer, without having to make modifications to ChunkDemuxer.
 246   class ManifestDemuxerStream : public DemuxerStream {
 247    public:
 248     ~ManifestDemuxerStream() override;
 249     using WrapperReadCb =
 250         base::RepeatingCallback<void(DemuxerStream::ReadCB,
 251                                      DemuxerStream::Status,
 252                                      DemuxerStream::DecoderBufferVector)>;
 253     ManifestDemuxerStream(DemuxerStream* stream, WrapperReadCb cb);
 254     void Read(uint32_t count, DemuxerStream::ReadCB cb) override;
 255     AudioDecoderConfig audio_decoder_config() override;
 256     VideoDecoderConfig video_decoder_config() override;
 257     DemuxerStream::Type type() const override;
 258     StreamLiveness liveness() const override;
 259     void EnableBitstreamConverter() override;
 260     bool SupportsConfigChanges() override;
 261
 262    private:
 263     WrapperReadCb read_cb_;
 264     raw_ptr<DemuxerStream> stream_;
 265   };
 266
 267   void OnChunkDemuxerInitialized(PipelineStatus init_status);
 268   void OnChunkDemuxerOpened();
 269   void OnProgress();
 270   void OnEncryptedMediaData(EmeInitDataType type,
 271                             const std::vector<uint8_t>& data);
 272   void OnChunkDemuxerParseWarning(std::string role,
 273                                   SourceBufferParseWarning warning);
 274   void OnChunkDemuxerTracksChanged(std::string role,
 275                                    std::unique_ptr<MediaTracks> tracks);
 276
 277   void OnDemuxerStreamRead(DemuxerStream::ReadCB wrapped_read_cb,
 278                            DemuxerStream::Status status,
 279                            DemuxerStream::DecoderBufferVector buffers);
 280
 281   // Helper for the `Seek` call, so that returning from an event when a seek
 282   // is pending can continue the seek process.
 283   void SeekInternal();
 284   void OnEngineSeeked(SeekResponse seek_status);
 285   void OnChunkDemuxerSeeked(PipelineStatus seek_status);
 286   void OnSeekBuffered(base::TimeDelta delay_time);
 287
 288   // Allows for both the chunk demuxer and the engine to be required for
 289   // initialization.
 290   void OnEngineInitialized(PipelineStatus status);
 291   void MaybeCompleteInitialize();
 292
 293   // Trigger the next event, and based on it's expected delay, post a
 294   // cancellable callback to TriggerEvent again.
 295   void TriggerEvent();
 296   void TriggerEventWithTime(DelayCallback cb, base::TimeDelta current_time);
 297   void OnEngineEventFinished(base::TimeDelta delay_time);
 298
 299   base::RepeatingCallback<void(base::TimeDelta)> request_seek_;
 300
 301   std::unique_ptr<MediaLog> media_log_;
 302   scoped_refptr<base::SequencedTaskRunner> media_task_runner_;
 303
 304   // Pending callbacks.
 305   PipelineStatusCallback pending_seek_;
 306   PipelineStatusCallback pending_init_;
 307
 308   // Engine implementation
 309   std::unique_ptr<Engine> impl_;
 310
 311   // Wrapped chunk demuxer that actually does the parsing and demuxing of the
 312   // raw data we feed it.
 313   std::unique_ptr<ChunkDemuxer> chunk_demuxer_;
 314   raw_ptr<DemuxerHost> host_;
 315
 316   // Updated by seek, and by updates from outgoing frames.
 317   base::TimeDelta media_time_ = base::Seconds(0);
 318
 319   // Playback rate helps calculate how often we should check for new data.
 320   double current_playback_rate_ = 0.0;
 321
 322   // Keeps a map of demuxer streams to their wrapper implementations which
 323   // can be used to set the current media time. ChunkDemuxer's streams live
 324   // forever due to the use of raw pointers in the pipeline, so these must
 325   // also live for the duration of `this` lifetime.
 326   base::flat_map<DemuxerStream*, std::unique_ptr<ManifestDemuxerStream>>
 327       streams_;
 328
 329   // Flags for the two part asynchronous initialization process.
 330   bool demuxer_opened_ = false;
 331   bool engine_impl_ready_ = false;
 332
 333   bool can_complete_seek_ = true;
 334
 335   // Pending an event. Don't trigger a new event chain while one is in
 336   // progress.
 337   bool has_pending_event_ = false;
 338
 339   // A pending "next event" callback, which can be canceled in the case of a
 340   // seek or a playback rate change.
 341   base::CancelableOnceClosure cancelable_next_event_;
 342
 343   base::WeakPtrFactory<ManifestDemuxer> weak_factory_{this};
 344 };
 345
 346 }  // namespace media
 347
 348 #endif  // MEDIA_FILTERS_MANIFEST_DEMUXER_H_