text: //
  // Copyright (c) 2013-2021 The SRS Authors
  //
  // SPDX-License-Identifier: MIT
  //
  
  #ifndef SRS_APP_SOURCE_HPP
  #define SRS_APP_SOURCE_HPP
  
  #include <srs_core.hpp>
  
  #include <map>
  #include <vector>
  #include <string>
  
  #include <srs_app_st.hpp>
  #include <srs_app_reload.hpp>
  #include <srs_core_performance.hpp>
  #include <srs_service_st.hpp>
  #include <srs_app_hourglass.hpp>
  
  class SrsFormat;
  class SrsRtmpFormat;
  class SrsLiveConsumer;
  class SrsPlayEdge;
  class SrsPublishEdge;
  class SrsLiveSource;
  class SrsCommonMessage;
  class SrsOnMetaDataPacket;
  class SrsSharedPtrMessage;
  class SrsForwarder;
  class SrsRequest;
  class SrsStSocket;
  class SrsRtmpServer;
  class SrsEdgeProxyContext;
  class SrsMessageArray;
  class SrsNgExec;
  class SrsMessageHeader;
  class SrsHls;
  class SrsRtc;
  class SrsDvr;
  class SrsDash;
  class SrsEncoder;
  class SrsBuffer;
  #ifdef SRS_HDS
  class SrsHds;
  #endif
  
  // The time jitter algorithm:
  // 1. full, to ensure stream start at zero, and ensure stream monotonically increasing.
  // 2. zero, only ensure sttream start at zero, ignore timestamp jitter.
  // 3. off, disable the time jitter algorithm, like atc.
  enum SrsRtmpJitterAlgorithm
  {
      SrsRtmpJitterAlgorithmFULL = 0x01,
      SrsRtmpJitterAlgorithmZERO,
      SrsRtmpJitterAlgorithmOFF
  };
  int srs_time_jitter_string2int(std::string time_jitter);
  
  // Time jitter detect and correct, to ensure the rtmp stream is monotonically.
  class SrsRtmpJitter
  {
  private:
      int64_t last_pkt_time;
      int64_t last_pkt_correct_time;
  public:
      SrsRtmpJitter();
      virtual ~SrsRtmpJitter();
  public:
      // detect the time jitter and correct it.
      // @param ag the algorithm to use for time jitter.
      virtual srs_error_t correct(SrsSharedPtrMessage* msg, SrsRtmpJitterAlgorithm ag);
      // Get current client time, the last packet time.
      virtual int64_t get_time();
  };
  
  #ifdef SRS_PERF_QUEUE_FAST_VECTOR
  // To alloc and increase fixed space, fast remove and insert for msgs sender.
  // @see https://github.com/ossrs/srs/issues/251
  class SrsFastVector
  {
  private:
      SrsSharedPtrMessage** msgs;
      int nb_msgs;
      int count;
  public:
      SrsFastVector();
      virtual ~SrsFastVector();
  public:
      virtual int size();
      virtual int begin();
      virtual int end();
      virtual SrsSharedPtrMessage** data();
      virtual SrsSharedPtrMessage* at(int index);
      virtual void clear();
      virtual void erase(int _begin, int _end);
      virtual void push_back(SrsSharedPtrMessage* msg);
      virtual void free();
  };
  #endif
  
  // The message queue for the consumer(client), forwarder.
  // We limit the size in seconds, drop old messages(the whole gop) if full.
  class SrsMessageQueue
  {
  private:
      // The start and end time.
      srs_utime_t av_start_time;
      srs_utime_t av_end_time;
  private:
  	// Whether do logging when shrinking.
      bool _ignore_shrink;
      // The max queue size, shrink if exceed it.
      srs_utime_t max_queue_size;
  #ifdef SRS_PERF_QUEUE_FAST_VECTOR
      SrsFastVector msgs;
  #else
      std::vector<SrsSharedPtrMessage*> msgs;
  #endif
  public:
      SrsMessageQueue(bool ignore_shrink = false);
      virtual ~SrsMessageQueue();
  public:
      // Get the size of queue.
      virtual int size();
      // Get the duration of queue.
      virtual srs_utime_t duration();
      // Set the queue size
      // @param queue_size the queue size in srs_utime_t.
      virtual void set_queue_size(srs_utime_t queue_size);
  public:
      // Enqueue the message, the timestamp always monotonically.
      // @param msg, the msg to enqueue, user never free it whatever the return code.
      // @param is_overflow, whether overflow and shrinked. NULL to ignore.
      virtual srs_error_t enqueue(SrsSharedPtrMessage* msg, bool* is_overflow = NULL);
      // Get packets in consumer queue.
      // @pmsgs SrsSharedPtrMessage*[], used to store the msgs, user must alloc it.
      // @count the count in array, output param.
      // @max_count the max count to dequeue, must be positive.
      virtual srs_error_t dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count);
      // Dumps packets to consumer, use specified args.
      // @remark the atc/tba/tbv/ag are same to SrsLiveConsumer.enqueue().
      virtual srs_error_t dump_packets(SrsLiveConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm ag);
  private:
      // Remove a gop from the front.
      // if no iframe found, clear it.
      virtual void shrink();
  public:
      // clear all messages in queue.
      virtual void clear();
  };
  
  // The wakable used for some object
  // which is waiting on cond.
  class ISrsWakable
  {
  public:
      ISrsWakable();
      virtual ~ISrsWakable();
  public:
      // when the consumer(for player) got msg from recv thread,
      // it must be processed for maybe it's a close msg, so the cond
      // wait must be wakeup.
      virtual void wakeup() = 0;
  };
  
  // The consumer for SrsLiveSource, that is a play client.
  class SrsLiveConsumer : public ISrsWakable
  {
  private:
      SrsRtmpJitter* jitter;
      SrsLiveSource* source;
      SrsMessageQueue* queue;
      bool paused;
      // when source id changed, notice all consumers
      bool should_update_source_id;
  #ifdef SRS_PERF_QUEUE_COND_WAIT
      // The cond wait for mw.
      // @see https://github.com/ossrs/srs/issues/251
      srs_cond_t mw_wait;
      bool mw_waiting;
      int mw_min_msgs;
      srs_utime_t mw_duration;
  #endif
  public:
      SrsLiveConsumer(SrsLiveSource* s);
      virtual ~SrsLiveConsumer();
  public:
      // Set the size of queue.
      virtual void set_queue_size(srs_utime_t queue_size);
      // when source id changed, notice client to print.
      virtual void update_source_id();
  public:
      // Get current client time, the last packet time.
      virtual int64_t get_time();
      // Enqueue an shared ptr message.
      // @param shared_msg, directly ptr, copy it if need to save it.
      // @param whether atc, donot use jitter correct if true.
      // @param ag the algorithm of time jitter.
      virtual srs_error_t enqueue(SrsSharedPtrMessage* shared_msg, bool atc, SrsRtmpJitterAlgorithm ag);
      // Get packets in consumer queue.
      // @param msgs the msgs array to dump packets to send.
      // @param count the count in array, intput and output param.
      // @remark user can specifies the count to get specified msgs; 0 to get all if possible.
      virtual srs_error_t dump_packets(SrsMessageArray* msgs, int& count);
  #ifdef SRS_PERF_QUEUE_COND_WAIT
      // wait for messages incomming, atleast nb_msgs and in duration.
      // @param nb_msgs the messages count to wait.
      // @param msgs_duration the messages duration to wait.
      virtual void wait(int nb_msgs, srs_utime_t msgs_duration);
  #endif
      // when client send the pause message.
      virtual srs_error_t on_play_client_pause(bool is_pause);
  // Interface ISrsWakable
  public:
      // when the consumer(for player) got msg from recv thread,
      // it must be processed for maybe it's a close msg, so the cond
      // wait must be wakeup.
      virtual void wakeup();
  };
  
  // cache a gop of video/audio data,
  // delivery at the connect of flash player,
  // To enable it to fast startup.
  class SrsGopCache
  {
  private:
      // if disabled the gop cache,
      // The client will wait for the next keyframe for h264,
      // and will be black-screen.
      bool enable_gop_cache;
      // The video frame count, avoid cache for pure audio stream.
      int cached_video_count;
      // when user disabled video when publishing, and gop cache enalbed,
      // We will cache the audio/video for we already got video, but we never
      // know when to clear the gop cache, for there is no video in future,
      // so we must guess whether user disabled the video.
      // when we got some audios after laster video, for instance, 600 audio packets,
      // about 3s(26ms per packet) 115 audio packets, clear gop cache.
      //
      // @remark, it is ok for performance, for when we clear the gop cache,
      //       gop cache is disabled for pure audio stream.
      // @see: https://github.com/ossrs/srs/issues/124
      int audio_after_last_video_count;
      // cached gop.
      std::vector<SrsSharedPtrMessage*> gop_cache;
  public:
      SrsGopCache();
      virtual ~SrsGopCache();
  public:
      // cleanup when system quit.
      virtual void dispose();
      // To enable or disable the gop cache.
      virtual void set(bool v);
      virtual bool enabled();
      // only for h264 codec
      // 1. cache the gop when got h264 video packet.
      // 2. clear gop when got keyframe.
      // @param shared_msg, directly ptr, copy it if need to save it.
      virtual srs_error_t cache(SrsSharedPtrMessage* shared_msg);
      // clear the gop cache.
      virtual void clear();
      // dump the cached gop to consumer.
      virtual srs_error_t dump(SrsLiveConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm jitter_algorithm);
      // used for atc to get the time of gop cache,
      // The atc will adjust the sequence header timestamp to gop cache.
      virtual bool empty();
      // Get the start time of gop cache, in srs_utime_t.
      // @return 0 if no packets.
      virtual srs_utime_t start_time();
      // whether current stream is pure audio,
      // when no video in gop cache, the stream is pure audio right now.
      virtual bool pure_audio();
  };
  
  // The handler to handle the event of srs source.
  // For example, the http flv streaming module handle the event and
  // mount http when rtmp start publishing.
  class ISrsLiveSourceHandler
  {
  public:
      ISrsLiveSourceHandler();
      virtual ~ISrsLiveSourceHandler();
  public:
      // when stream start publish, mount stream.
      virtual srs_error_t on_publish(SrsLiveSource* s, SrsRequest* r) = 0;
      // when stream stop publish, unmount stream.
      virtual void on_unpublish(SrsLiveSource* s, SrsRequest* r) = 0;
  };
  
  // The mix queue to correct the timestamp for mix_correct algorithm.
  class SrsMixQueue
  {
  private:
      uint32_t nb_videos;
      uint32_t nb_audios;
      std::multimap<int64_t, SrsSharedPtrMessage*> msgs;
  public:
      SrsMixQueue();
      virtual ~SrsMixQueue();
  public:
      virtual void clear();
      virtual void push(SrsSharedPtrMessage* msg);
      virtual SrsSharedPtrMessage* pop();
  };
  
  // The hub for origin is a collection of utilities for origin only,
  // For example, DVR, HLS, Forward and Transcode are only available for origin,
  // they are meanless for edge server.
  class SrsOriginHub : public ISrsReloadHandler
  {
  private:
      SrsLiveSource* source;
      SrsRequest* req;
      bool is_active;
  private:
      // The format, codec information.
      SrsRtmpFormat* format;
      // hls handler.
      SrsHls* hls;
      // The DASH encoder.
      SrsDash* dash;
      // dvr handler.
      SrsDvr* dvr;
      // transcoding handler.
      SrsEncoder* encoder;
  #ifdef SRS_HDS
      // adobe hds(http dynamic streaming).
      SrsHds *hds;
  #endif
      // nginx-rtmp exec feature.
      SrsNgExec* ng_exec;
      // To forward stream to other servers
      std::vector<SrsForwarder*> forwarders;
  public:
      SrsOriginHub();
      virtual ~SrsOriginHub();
  public:
      // Initialize the hub with source and request.
      // @param r The request object, managed by source.
      virtual srs_error_t initialize(SrsLiveSource* s, SrsRequest* r);
      // Dispose the hub, release utilities resource,
      // For example, delete all HLS pieces.
      virtual void dispose();
      // Cycle the hub, process some regular events,
      // For example, dispose hls in cycle.
      virtual srs_error_t cycle();
      // Whether the stream hub is active, or stream is publishing.
      virtual bool active();
  public:
      // When got a parsed metadata.
      virtual srs_error_t on_meta_data(SrsSharedPtrMessage* shared_metadata, SrsOnMetaDataPacket* packet);
      // When got a parsed audio packet.
      virtual srs_error_t on_audio(SrsSharedPtrMessage* shared_audio);
      // When got a parsed video packet.
      virtual srs_error_t on_video(SrsSharedPtrMessage* shared_video, bool is_sequence_header);
  public:
      // When start publish stream.
      virtual srs_error_t on_publish();
      // When stop publish stream.
      virtual void on_unpublish();
  // Internal callback.
  public:
      // For the SrsForwarder to callback to request the sequence headers.
      virtual srs_error_t on_forwarder_start(SrsForwarder* forwarder);
      // For the SrsDvr to callback to request the sequence headers.
      virtual srs_error_t on_dvr_request_sh();
  // Interface ISrsReloadHandler
  public:
      virtual srs_error_t on_reload_vhost_forward(std::string vhost);
      virtual srs_error_t on_reload_vhost_dash(std::string vhost);
      virtual srs_error_t on_reload_vhost_hls(std::string vhost);
      virtual srs_error_t on_reload_vhost_hds(std::string vhost);
      virtual srs_error_t on_reload_vhost_dvr(std::string vhost);
      virtual srs_error_t on_reload_vhost_transcode(std::string vhost);
      virtual srs_error_t on_reload_vhost_exec(std::string vhost);
  private:
      virtual srs_error_t create_forwarders();
      virtual void destroy_forwarders();
  };
  
  // Each stream have optional meta(sps/pps in sequence header and metadata).
  // This class cache and update the meta.
  class SrsMetaCache
  {
  private:
      // The cached metadata, FLV script data tag.
      SrsSharedPtrMessage* meta;
      // The cached video sequence header, for example, sps/pps for h.264.
      SrsSharedPtrMessage* video;
      SrsSharedPtrMessage* previous_video;
      // The cached audio sequence header, for example, asc for aac.
      SrsSharedPtrMessage* audio;
      SrsSharedPtrMessage* previous_audio;
      // The format for sequence header.
      SrsRtmpFormat* vformat;
      SrsRtmpFormat* aformat;
  public:
      SrsMetaCache();
      virtual ~SrsMetaCache();
  public:
      // Dispose the metadata cache.
      virtual void dispose();
      // For each publishing, clear the metadata cache.
      virtual void clear();
  public:
      // Get the cached metadata.
      virtual SrsSharedPtrMessage* data();
      // Get the cached vsh(video sequence header).
      virtual SrsSharedPtrMessage* vsh();
      virtual SrsFormat* vsh_format();
      // Get the cached ash(audio sequence header).
      virtual SrsSharedPtrMessage* ash();
      virtual SrsFormat* ash_format();
      // Dumps cached metadata to consumer.
      // @param dm Whether dumps the metadata.
      // @param ds Whether dumps the sequence header.
      virtual srs_error_t dumps(SrsLiveConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm ag, bool dm, bool ds);
  public:
      // Previous exists sequence header.
      virtual SrsSharedPtrMessage* previous_vsh();
      virtual SrsSharedPtrMessage* previous_ash();
      // Update previous sequence header, drop old one, set to new sequence header.
      virtual void update_previous_vsh();
      virtual void update_previous_ash();
  public:
      // Update the cached metadata by packet.
      virtual srs_error_t update_data(SrsMessageHeader* header, SrsOnMetaDataPacket* metadata, bool& updated);
      // Update the cached audio sequence header.
      virtual srs_error_t update_ash(SrsSharedPtrMessage* msg);
      // Update the cached video sequence header.
      virtual srs_error_t update_vsh(SrsSharedPtrMessage* msg);
  };
  
  // The source manager to create and refresh all stream sources.
  class SrsLiveSourceManager : public ISrsHourGlass
  {
  private:
      srs_mutex_t lock;
      std::map<std::string, SrsLiveSource*> pool;
      SrsHourGlass* timer_;
  public:
      SrsLiveSourceManager();
      virtual ~SrsLiveSourceManager();
  public:
      virtual srs_error_t initialize();
      //  create source when fetch from cache failed.
      // @param r the client request.
      // @param h the event handler for source.
      // @param pps the matched source, if success never be NULL.
      virtual srs_error_t fetch_or_create(SrsRequest* r, ISrsLiveSourceHandler* h, SrsLiveSource** pps);
  private:
      // Get the exists source, NULL when not exists.
      // update the request and return the exists source.
      virtual SrsLiveSource* fetch(SrsRequest* r);
  public:
      // dispose and cycle all sources.
      virtual void dispose();
  // interface ISrsHourGlass
  private:
      virtual srs_error_t setup_ticks();
      virtual srs_error_t notify(int event, srs_utime_t interval, srs_utime_t tick);
  public:
      // when system exit, destroy th`e sources,
      // For gmc to analysis mem leaks.
      virtual void destroy();
  };
  
  // Global singleton instance.
  extern SrsLiveSourceManager* _srs_sources;
  
  // For RTMP2RTC, bridge SrsLiveSource to SrsRtcSource
  class ISrsLiveSourceBridger
  {
  public:
      ISrsLiveSourceBridger();
      virtual ~ISrsLiveSourceBridger();
  public:
      virtual srs_error_t on_publish() = 0;
      virtual srs_error_t on_audio(SrsSharedPtrMessage* audio) = 0;
      virtual srs_error_t on_video(SrsSharedPtrMessage* video) = 0;
      virtual void on_unpublish() = 0;
  };
  
  // The live streaming source.
  class SrsLiveSource : public ISrsReloadHandler
  {
      friend class SrsOriginHub;
  private:
      // For publish, it's the publish client id.
      // For edge, it's the edge ingest id.
      // when source id changed, for example, the edge reconnect,
      // invoke the on_source_id_changed() to let all clients know.
      SrsContextId _source_id;
      // previous source id.
      SrsContextId _pre_source_id;
      // deep copy of client request.
      SrsRequest* req;
      // To delivery stream to clients.
      std::vector<SrsLiveConsumer*> consumers;
      // The time jitter algorithm for vhost.
      SrsRtmpJitterAlgorithm jitter_algorithm;
      // For play, whether use interlaced/mixed algorithm to correct timestamp.
      bool mix_correct;
      // The mix queue to implements the mix correct algorithm.
      SrsMixQueue* mix_queue;
      // For play, whether enabled atc.
      // The atc(use absolute time and donot adjust time),
      // directly use msg time and donot adjust if atc is true,
      // otherwise, adjust msg time to start from 0 to make flash happy.
      bool atc;
      // whether stream is monotonically increase.
      bool is_monotonically_increase;
      // The time of the packet we just got.
      int64_t last_packet_time;
      // The event handler.
      ISrsLiveSourceHandler* handler;
      // The source bridger for other source.
      ISrsLiveSourceBridger* bridger_;
      // The edge control service
      SrsPlayEdge* play_edge;
      SrsPublishEdge* publish_edge;
      // The gop cache for client fast startup.
      SrsGopCache* gop_cache;
      // The hub for origin server.
      SrsOriginHub* hub;
      // The metadata cache.
      SrsMetaCache* meta;
  private:
      // Whether source is avaiable for publishing.
      bool _can_publish;
      // The last die time, when all consumers quit and no publisher,
      // We will remove the source when source die.
      srs_utime_t die_at;
  public:
      SrsLiveSource();
      virtual ~SrsLiveSource();
  public:
      virtual void dispose();
      virtual srs_error_t cycle();
      // Remove source when expired.
      virtual bool expired();
  public:
      // Initialize the hls with handlers.
      virtual srs_error_t initialize(SrsRequest* r, ISrsLiveSourceHandler* h);
      // Bridge to other source, forward packets to it.
      void set_bridger(ISrsLiveSourceBridger* v);
  // Interface ISrsReloadHandler
  public:
      virtual srs_error_t on_reload_vhost_play(std::string vhost);
  public:
      // The source id changed.
      virtual srs_error_t on_source_id_changed(SrsContextId id);
      // Get current source id.
      virtual SrsContextId source_id();
      virtual SrsContextId pre_source_id();
      // Whether source is inactive, which means there is no publishing stream source.
      // @remark For edge, it's inactive util stream has been pulled from origin.
      virtual bool inactive();
      // Update the authentication information in request.
      virtual void update_auth(SrsRequest* r);
  public:
      virtual bool can_publish(bool is_edge);
      virtual srs_error_t on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
  public:
      // TODO: FIXME: Use SrsSharedPtrMessage instead.
      virtual srs_error_t on_audio(SrsCommonMessage* audio);
  private:
      virtual srs_error_t on_audio_imp(SrsSharedPtrMessage* audio);
  public:
      // TODO: FIXME: Use SrsSharedPtrMessage instead.
      virtual srs_error_t on_video(SrsCommonMessage* video);
  private:
      virtual srs_error_t on_video_imp(SrsSharedPtrMessage* video);
  public:
      virtual srs_error_t on_aggregate(SrsCommonMessage* msg);
      // Publish stream event notify.
      // @param _req the request from client, the source will deep copy it,
      //         for when reload the request of client maybe invalid.
      virtual srs_error_t on_publish();
      virtual void on_unpublish();
  public:
      // Create consumer
      // @param consumer, output the create consumer.
      virtual srs_error_t create_consumer(SrsLiveConsumer*& consumer);
      // Dumps packets in cache to consumer.
      // @param ds, whether dumps the sequence header.
      // @param dm, whether dumps the metadata.
      // @param dg, whether dumps the gop cache.
      virtual srs_error_t consumer_dumps(SrsLiveConsumer* consumer, bool ds = true, bool dm = true, bool dg = true);
      virtual void on_consumer_destroy(SrsLiveConsumer* consumer);
      virtual void set_cache(bool enabled);
      virtual SrsRtmpJitterAlgorithm jitter();
  public:
      // For edge, when publish edge stream, check the state
      virtual srs_error_t on_edge_start_publish();
      // For edge, proxy the publish
      virtual srs_error_t on_edge_proxy_publish(SrsCommonMessage* msg);
      // For edge, proxy stop publish
      virtual void on_edge_proxy_unpublish();
  public:
      virtual std::string get_curr_origin();
  };
  
  #endif
  