Support LL-HLS

application/xiu/Cargo.toml

@@ -22,7 +22,7 @@ failure = "0.1.1"
 
 rtmp = "0.0.14"#{path = "../../protocol/rtmp/"}#"0.0.4"
 httpflv = "0.0.7"#{path = "../../protocol/httpflv/"}
-hls = "0.0.10"#{path = "../../protocol/hls/"}
+hls = {path = "../../protocol/hls/"}
 
 # rtmp = {path = "../../protocol/rtmp/"}#"0.0.4"
 # httpflv = {path = "../../protocol/httpflv/"}

application/xiu/src/main.rs

@@ -1,3 +1,4 @@
+use std::sync::Arc;
 use {
     //https://rustcc.cn/article?id=6dcbf032-0483-4980-8bfe-c64a7dfb33c7
     anyhow::Result,
@@ -17,6 +18,7 @@ use {
 };
 
 //use application::logger::logger;
+use hls::hls_event_manager::HlsEventManager;
 use hls::rtmp_event_processor::RtmpEventProcessor;
 
 #[tokio::main]
@@ -39,7 +41,7 @@ async fn main() -> Result<()> {
             if let Some(log_config_value) = &val.log {
                 env::set_var("RUST_LOG", log_config_value.level.clone());
             } else {
-            env::set_var("RUST_LOG", "info");
+                env::set_var("RUST_LOG", "info");
             }
 
             // let mut builder = Builder::from_default_env();
@@ -196,10 +198,16 @@ impl Service {
                 return Ok(());
             }
 
+            let hls_manager = HlsEventManager::new();
+            let hls_dispatch = hls_manager.setup_dispatch_channel();
+
             let event_producer = channel.get_session_event_producer().clone();
-            let cient_event_consumer = channel.get_client_event_consumer();
-            let mut rtmp_event_processor =
-                RtmpEventProcessor::new(cient_event_consumer, event_producer);
+            let client_event_consumer = channel.get_client_event_consumer();
+            let mut rtmp_event_processor = RtmpEventProcessor::new(
+                client_event_consumer,
+                event_producer,
+                hls_dispatch.clone(),
+            );
 
             tokio::spawn(async move {
                 if let Err(err) = rtmp_event_processor.run().await {
@@ -211,7 +219,7 @@ impl Service {
             let port = hls_cfg_value.port;
 
             tokio::spawn(async move {
-                if let Err(err) = hls_server::run(port).await {
+                if let Err(err) = hls_server::run(port, hls_manager).await {
                     //print!("push client error {}\n", err);
                     log::error!("hls server error: {}\n", err);
                 }

protocol/hls/Cargo.toml

@@ -23,11 +23,11 @@ rtmp = "0.0.14"#{path = "../rtmp/"}#"0.0.4"
 # rtmp = {path = "../rtmp/"}#"0.0.4"
 xmpegts = "0.0.3"#{path = "../../library/container/mpegts/"}
 
-
+url = "2.2.2"
 hyper = { version = "0.14", features = ["full"] }
 tokio-util = { version = "0.6.5", features = ["codec"] }
 
 [dependencies.tokio]
-version = "1.4.0"
+version = "1.20.0"
 default-features = false
 features = ["full"]

protocol/hls/src/flv2hls.rs

@@ -1,5 +1,9 @@
+use crate::hls_event_manager::M3u8Consumer;
+
 use {
-    super::{define::FlvDemuxerData, errors::MediaError, m3u8::M3u8},
+    super::{
+        define::FlvDemuxerData, errors::MediaError, hls_event_manager::HlsEventProducer, m3u8::M3u8,
+    },
     bytes::BytesMut,
     xflv::{
         define::{frame_type, FlvData},
@@ -15,25 +19,40 @@ pub struct Flv2HlsRemuxer {
     video_demuxer: FlvVideoTagDemuxer,
     audio_demuxer: FlvAudioTagDemuxer,
 
+    partial_ts_muxer: TsMuxer,
     ts_muxer: TsMuxer,
 
     last_ts_dts: i64,
+    last_partial_ts_dts: i64,
     last_ts_pts: i64,
 
     last_dts: i64,
     last_pts: i64,
 
     duration: i64,
+    partial_seg_duration: i64,
     need_new_segment: bool,
+    need_new_partial_segment: bool,
+    segment_has_idr: bool,
 
     video_pid: u16,
     audio_pid: u16,
 
+    p_video_pid: u16,
+    p_audio_pid: u16,
+
     m3u8_handler: M3u8,
 }
 
 impl Flv2HlsRemuxer {
-    pub fn new(duration: i64, app_name: String, stream_name: String) -> Self {
+    pub fn new(
+        hls_event_tx: HlsEventProducer,
+        m3u8_consumer: M3u8Consumer,
+        duration: i64,
+        partial_seg_duration: i64,
+        app_name: String,
+        stream_name: String,
+    ) -> Self {
         let mut ts_muxer = TsMuxer::new();
         let audio_pid = ts_muxer
             .add_stream(epsi_stream_type::PSI_STREAM_AAC, BytesMut::new())
@@ -42,33 +61,54 @@ impl Flv2HlsRemuxer {
             .add_stream(epsi_stream_type::PSI_STREAM_H264, BytesMut::new())
             .unwrap();
 
+        let mut partial_ts_muxer = TsMuxer::new();
+        let p_audio_pid = partial_ts_muxer
+            .add_stream(epsi_stream_type::PSI_STREAM_AAC, BytesMut::new())
+            .unwrap();
+        let p_video_pid = partial_ts_muxer
+            .add_stream(epsi_stream_type::PSI_STREAM_H264, BytesMut::new())
+            .unwrap();
+
         let m3u8_name = format!("{}.m3u8", stream_name);
 
+        let m3u8_handler = M3u8::new(
+            hls_event_tx,
+            duration,
+            6,
+            m3u8_name,
+            app_name.clone(),
+            stream_name.clone(),
+        );
+
+        m3u8_handler.setup_m3u8_listener(m3u8_consumer);
+
         Self {
             video_demuxer: FlvVideoTagDemuxer::new(),
             audio_demuxer: FlvAudioTagDemuxer::new(),
 
             ts_muxer,
+            partial_ts_muxer,
 
             last_ts_dts: 0,
+            last_partial_ts_dts: 0,
             last_ts_pts: 0,
 
             last_dts: 0,
             last_pts: 0,
 
             duration,
+            partial_seg_duration,
             need_new_segment: false,
+            need_new_partial_segment: false,
+            segment_has_idr: false,
 
             video_pid,
             audio_pid,
 
-            m3u8_handler: M3u8::new(
-                duration,
-                6,
-                m3u8_name,
-                app_name.clone(),
-                stream_name.clone(),
-            ),
+            p_video_pid,
+            p_audio_pid,
+
+            m3u8_handler: m3u8_handler,
         }
     }
 
@@ -104,7 +144,7 @@ impl Flv2HlsRemuxer {
             true,
             data,
         )?;
-        self.m3u8_handler.refresh_playlist()?;
+        self.m3u8_handler.refresh_playlist(false)?;
 
         Ok(())
     }
@@ -114,8 +154,10 @@ impl Flv2HlsRemuxer {
         flv_demux_data: &FlvDemuxerData,
     ) -> Result<(), MediaError> {
         self.need_new_segment = false;
+        self.need_new_partial_segment = false;
 
         let pid: u16;
+        let p_pid: u16;
         let pts: i64;
         let dts: i64;
         let mut flags: u16 = 0;
@@ -130,14 +172,22 @@ impl Flv2HlsRemuxer {
                 pts = data.pts;
                 dts = data.dts;
                 pid = self.video_pid;
+                p_pid = self.video_pid;
+
                 payload.extend_from_slice(&data.data[..]);
 
                 if data.frame_type == frame_type::KEY_FRAME {
                     flags = MPEG_FLAG_IDR_FRAME;
+                    self.segment_has_idr = true;
+
                     if dts - self.last_ts_dts >= self.duration * 1000 {
                         self.need_new_segment = true;
                     }
                 }
+
+                if dts - self.last_partial_ts_dts >= self.partial_seg_duration {
+                    self.need_new_partial_segment = true;
+                }
             }
             FlvDemuxerData::Audio { data } => {
                 if !data.has_data {
@@ -147,6 +197,7 @@ impl Flv2HlsRemuxer {
                 pts = data.pts;
                 dts = data.dts;
                 pid = self.audio_pid;
+                p_pid = self.audio_pid;
                 payload.extend_from_slice(&data.data[..]);
             }
             _ => return Ok(()),
@@ -161,19 +212,37 @@ impl Flv2HlsRemuxer {
 
             self.m3u8_handler
                 .add_segment(dts - self.last_ts_dts, discontinuity, false, data)?;
-            self.m3u8_handler.refresh_playlist()?;
+            self.m3u8_handler.refresh_playlist(true)?;
 
             self.ts_muxer.reset();
             self.last_ts_dts = dts;
             self.last_ts_pts = pts;
             self.need_new_segment = false;
+            self.segment_has_idr = false;
+        } else if self.need_new_partial_segment {
+            let d = self.partial_ts_muxer.get_data();
+
+            self.m3u8_handler.add_partial_segment(
+                dts - self.last_partial_ts_dts,
+                d,
+                self.segment_has_idr,
+            )?;
+            self.m3u8_handler.refresh_playlist(false)?;
+
+            self.partial_ts_muxer.reset();
+            self.last_partial_ts_dts = dts;
+            self.need_new_partial_segment = false;
+            self.segment_has_idr = false;
         }
 
         self.last_dts = dts;
         self.last_pts = pts;
 
         self.ts_muxer
-            .write(pid, pts * 90, dts * 90, flags, payload)?;
+            .write(pid, pts * 90, dts * 90, flags, payload.clone())?;
+
+        self.partial_ts_muxer
+            .write(p_pid, pts * 90, dts * 90, flags, payload)?;
 
         Ok(())
     }

protocol/hls/src/flv_data_receiver.rs

@@ -1,7 +1,10 @@
+use crate::hls_event_manager::M3u8Consumer;
+
 use {
     super::{
         errors::{HlsError, HlsErrorValue},
         flv2hls::Flv2HlsRemuxer,
+        hls_event_manager::HlsEventProducer,
     },
     rtmp::channels::define::{
         ChannelData, ChannelDataConsumer, ChannelEvent, ChannelEventProducer,
@@ -29,7 +32,6 @@ use {
 pub struct FlvDataReceiver {
     app_name: String,
     stream_name: String,
-
     event_producer: ChannelEventProducer,
     data_consumer: ChannelDataConsumer,
     media_processor: Flv2HlsRemuxer,
@@ -41,6 +43,8 @@ impl FlvDataReceiver {
         app_name: String,
         stream_name: String,
         event_producer: ChannelEventProducer,
+        hls_event_tx: HlsEventProducer,
+        m3u8_consumer: M3u8Consumer,
 
         duration: i64,
     ) -> Self {
@@ -53,7 +57,14 @@ impl FlvDataReceiver {
 
             data_consumer,
             event_producer,
-            media_processor: Flv2HlsRemuxer::new(duration, app_name, stream_name),
+            media_processor: Flv2HlsRemuxer::new(
+                hls_event_tx,
+                m3u8_consumer,
+                duration,
+                1000,
+                app_name,
+                stream_name,
+            ),
             subscriber_id,
         }
     }