[feat] support HLS record

harlanc · harlanc · commit b754b6929107 · 2023-08-04T23:15:00.000+08:00
diff --git a/README.md b/README.md
@@ -44,6 +44,7 @@ Xiu is a simple,high performance and secure live media server written in pure Ru
   - [x] Support querying stream information.
   - [x] Support notification of stream status.
 - [x] Support token authentications.
+- [x] Support recording live streams into HLS files(m3u8+ts).
 
 
 ## Preparation
@@ -163,6 +164,8 @@ You can use command line to configure the xiu server easily. You can specify to
     enabled = true
     # listening port
     port = 8080
+    # need record the live stream or not
+    need_record = true
 
 ##### Log
 
diff --git a/README_CN.md b/README_CN.md
@@ -42,6 +42,7 @@ XIU是用纯Rust开发的一款简单和安全的流媒体服务器，目前支
     - [x] 支持查询流信息
     - [x] 支持流事件通知
 - [x] 支持token鉴权
+- [x] 支持把直播流录制成HLS协议(m3u8+ts)文件.
 
 ## 准备工作
 #### 安装 Rust and Cargo
@@ -163,6 +164,8 @@ XIU是用纯Rust开发的一款简单和安全的流媒体服务器，目前支
     enabled = true
     # listening port
     port = 8080
+    # need record the live stream or not
+    need_record = true
 
 ##### Log
 
diff --git a/application/http-server/Cargo.toml b/application/http-server/Cargo.toml
@@ -15,7 +15,7 @@ serde_json = { version = "1", default-features = false, features = [
 ] }
 axum = "0.6.10"
 log = "0.4.0"
-env_logger = "0.9.3"
+env_logger = "0.10.0"
 
 [dependencies.tokio]
 version = "1.4.0"
diff --git a/application/pprtmp/Cargo.toml b/application/pprtmp/Cargo.toml
@@ -8,7 +8,7 @@ edition = "2021"
 [dependencies]
 anyhow = "^1.0"
 log = "0.4.0"
-env_logger = "0.9.3"
+env_logger = "0.10.0"
 clap = "4.1.4"
 
 rtmp = "0.4.0"
diff --git a/application/xiu/Cargo.toml b/application/xiu/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "xiu"
 description = "A powerful live server by Rust ."
-version = "0.7.0"
+version = "0.8.0"
 authors = ["HarlanC <wawacry@qq.com"]
 repository = "https://github.com/harlanc/xiu"
 license = "MIT"
@@ -31,7 +31,7 @@ streamhub = "0.1.0"
 rtmp = "0.4.0"
 xrtsp = "0.1.0"
 httpflv = "0.3.0"
-hls = "0.3.0"
+hls = "0.4.0"
 
 [features]
 default = ["std"]
diff --git a/application/xiu/README.md b/application/xiu/README.md
@@ -322,5 +322,7 @@ Open issues if you have any problems. Star and pull requests are welcomed. Your
 - Fix RTMP examples in README.
 ## v0.7.0
 - Support RTSP.
+## v0.8.0
+- Support HLS record.
 
 
diff --git a/application/xiu/src/api.rs b/application/xiu/src/api.rs
@@ -36,11 +36,11 @@ impl ApiService {
     async fn get_stream_status(&self) -> Result<String> {
         let (data_sender, mut data_receiver) = mpsc::unbounded_channel();
         let (size_sender, size_receiver) = oneshot::channel();
-        let channel_event = define::StreamHubEvent::ApiStatistic {
+        let hub_event = define::StreamHubEvent::ApiStatistic {
             data_sender,
             size_sender,
         };
-        if let Err(err) = self.channel_event_producer.send(channel_event) {
+        if let Err(err) = self.channel_event_producer.send(hub_event) {
             log::error!("send api event error: {}", err);
         }
         let mut data = Vec::new();
@@ -74,9 +74,9 @@ impl ApiService {
         let id_result = Uuid::from_str2(&id.id);
 
         if let Some(id) = id_result {
-            let channel_event = define::StreamHubEvent::ApiKickClient { id };
+            let hub_event = define::StreamHubEvent::ApiKickClient { id };
 
-            if let Err(err) = self.channel_event_producer.send(channel_event) {
+            if let Err(err) = self.channel_event_producer.send(hub_event) {
                 log::error!("send api kick_off_client event error: {}", err);
             }
         }
diff --git a/application/xiu/src/config/config_rtmp_hls.toml b/application/xiu/src/config/config_rtmp_hls.toml
@@ -12,9 +12,11 @@ port = 1935
 [hls]
 enabled = true
 port = 8080
+#defalut false
+need_record = true
 
 #######################################
 #   LOG configurations                #
 #######################################
 [log]
-level = "info"
+level = "info"
diff --git a/application/xiu/src/config/mod.rs b/application/xiu/src/config/mod.rs
@@ -56,6 +56,7 @@ impl Config {
             hls_config = Some(HlsConfig {
                 enabled: true,
                 port: hls_port,
+                need_record: false,
             });
         }
 
@@ -113,6 +114,8 @@ pub struct HttpFlvConfig {
 pub struct HlsConfig {
     pub enabled: bool,
     pub port: usize,
+    //record or not
+    pub need_record: bool,
 }
 
 pub enum LogLevel {
@@ -164,7 +167,9 @@ fn test_toml_parse() {
     //     Err(err) => print!("{}\n", err),
     // }
 
-    let str = fs::read_to_string("/Users/zexu/github/xiu_live_rust/application/xiu/src/config/config.toml");
+    let str = fs::read_to_string(
+        "/Users/zexu/github/xiu_live_rust/application/xiu/src/config/config.toml",
+    );
 
     match str {
         Ok(val) => {
diff --git a/application/xiu/src/main.rs b/application/xiu/src/main.rs
@@ -14,7 +14,7 @@ async fn main() -> Result<()> {
 
     let mut cmd = Command::new("XIU")
         .bin_name("xiu")
-        .version("0.7.0")
+        .version("0.8.0")
         .author("HarlanC <harlanc@foxmail.com>")
         .about("A secure and easy to use live media server, hope you love it!!!")
         .arg(
diff --git a/application/xiu/src/service.rs b/application/xiu/src/service.rs
@@ -5,7 +5,7 @@ use {
     super::config::Config,
     //https://rustcc.cn/article?id=6dcbf032-0483-4980-8bfe-c64a7dfb33c7
     anyhow::Result,
-    hls::rtmp_event_processor::RtmpEventProcessor,
+    hls::remuxer::HlsRemuxer,
     hls::server as hls_server,
     httpflv::server as httpflv_server,
     rtmp::{
@@ -49,6 +49,7 @@ impl Service {
         self.start_rtmp(&mut stream_hub).await?;
         self.start_rtsp(&mut stream_hub).await?;
         self.start_http_api_server(&mut stream_hub).await?;
+        self.start_rtmp_remuxer(&mut stream_hub).await?;
 
         tokio::spawn(async move {
             stream_hub.run().await;
@@ -149,14 +150,33 @@ impl Service {
                     log::error!("rtmp server error: {}\n", err);
                 }
             });
-
-            self.start_rtmp_remuxer(stream_hub).await?;
         }
 
         Ok(())
     }
 
     async fn start_rtmp_remuxer(&mut self, stream_hub: &mut StreamsHub) -> Result<()> {
+        //The remuxer now is used for rtsp2rtmp, so both rtsp/rtmp cfg need to be enabled.
+        let mut rtsp_enabled = false;
+        if let Some(rtsp_cfg_value) = &self.cfg.rtsp {
+            if rtsp_cfg_value.enabled {
+                rtsp_enabled = true;
+            }
+        }
+        if !rtsp_enabled {
+            return Ok(());
+        }
+
+        let mut rtmp_enabled: bool = false;
+        if let Some(rtmp_cfg_value) = &self.cfg.rtmp {
+            if rtmp_cfg_value.enabled {
+                rtmp_enabled = true;
+            }
+        }
+        if !rtmp_enabled {
+            return Ok(());
+        }
+
         let event_producer = stream_hub.get_hub_event_sender();
         let broadcast_event_receiver = stream_hub.get_client_event_consumer();
         let mut remuxer = RtmpRemuxer::new(broadcast_event_receiver, event_producer);
@@ -224,11 +244,14 @@ impl Service {
 
             let event_producer = stream_hub.get_hub_event_sender();
             let cient_event_consumer = stream_hub.get_client_event_consumer();
-            let mut rtmp_event_processor =
-                RtmpEventProcessor::new(cient_event_consumer, event_producer);
+            let mut hls_remuxer = HlsRemuxer::new(
+                cient_event_consumer,
+                event_producer,
+                hls_cfg_value.need_record,
+            );
 
             tokio::spawn(async move {
-                if let Err(err) = rtmp_event_processor.run().await {
+                if let Err(err) = hls_remuxer.run().await {
                     log::error!("rtmp event processor error: {}\n", err);
                 }
             });
diff --git a/confs/local/pprtmp.Cargo.toml b/confs/local/pprtmp.Cargo.toml
@@ -10,7 +10,7 @@ anyhow = "^1.0"
 rtmp = { path = "../../protocol/rtmp/" }
 streamhub = { path = "../../library/streamhub/" }
 log = "0.4.0"
-env_logger = "0.9.3"
+env_logger = "0.10.0"
 clap = "4.1.4"
 
 [dependencies.tokio]
diff --git a/confs/online/hls.Cargo.toml b/confs/online/hls.Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "hls"
 description = "hls library."
-version = "0.3.0"
+version = "0.4.0"
 authors = ["HarlanC <wawacry@qq.com"]
 repository = "https://github.com/harlanc/xiu"
 license = "MIT"
diff --git a/confs/online/pprtmp.Cargo.toml b/confs/online/pprtmp.Cargo.toml
@@ -8,7 +8,7 @@ edition = "2021"
 [dependencies]
 anyhow = "^1.0"
 log = "0.4.0"
-env_logger = "0.9.3"
+env_logger = "0.10.0"
 clap = "4.1.4"
 
 rtmp = "0.4.0"
diff --git a/confs/online/xiu.Cargo.toml b/confs/online/xiu.Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "xiu"
 description = "A powerful live server by Rust ."
-version = "0.7.0"
+version = "0.8.0"
 authors = ["HarlanC <wawacry@qq.com"]
 repository = "https://github.com/harlanc/xiu"
 license = "MIT"
@@ -31,7 +31,7 @@ streamhub = "0.1.0"
 rtmp = "0.4.0"
 xrtsp = "0.1.0"
 httpflv = "0.3.0"
-hls = "0.3.0"
+hls = "0.4.0"
 
 [features]
 default = ["std"]
diff --git a/library/logger/Cargo.toml b/library/logger/Cargo.toml
@@ -11,10 +11,8 @@ repository = "https://github.com/harlanc/xiu"
 
 [dependencies]
 anyhow = "^1.0"
-env_logger = "0.9.3"
+env_logger = "0.10.0"
 job_scheduler = "1.2.1"
 chrono = "0.4"
 failure = "0.1.1"
 log = "0.4.0"
-
-
diff --git a/protocol/hls/Cargo.toml b/protocol/hls/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "hls"
 description = "hls library."
-version = "0.3.0"
+version = "0.4.0"
 authors = ["HarlanC <wawacry@qq.com"]
 repository = "https://github.com/harlanc/xiu"
 license = "MIT"
diff --git a/protocol/hls/README.md b/protocol/hls/README.md
@@ -25,5 +25,8 @@ Change the listening port type.
 ## 0.3.0
 - Do some refactoring.
 - Reference bytesio v0.3.0.
+## 0.4.0
+- Support record.
+
 
 
diff --git a/protocol/hls/src/flv2hls.rs b/protocol/hls/src/flv2hls.rs
@@ -33,7 +33,7 @@ pub struct Flv2HlsRemuxer {
 }
 
 impl Flv2HlsRemuxer {
-    pub fn new(duration: i64, app_name: String, stream_name: String) -> Self {
+    pub fn new(duration: i64, app_name: String, stream_name: String, need_record: bool) -> Self {
         let mut ts_muxer = TsMuxer::new();
         let audio_pid = ts_muxer
             .add_stream(epsi_stream_type::PSI_STREAM_AAC, BytesMut::new())
@@ -42,8 +42,6 @@ impl Flv2HlsRemuxer {
             .add_stream(epsi_stream_type::PSI_STREAM_H264, BytesMut::new())
             .unwrap();
 
-        let m3u8_name = format!("{stream_name}.m3u8");
-
         Self {
             video_demuxer: FlvVideoTagDemuxer::new(),
             audio_demuxer: FlvAudioTagDemuxer::new(),
@@ -62,7 +60,7 @@ impl Flv2HlsRemuxer {
             video_pid,
             audio_pid,
 
-            m3u8_handler: M3u8::new(duration, 6, m3u8_name, app_name, stream_name),
+            m3u8_handler: M3u8::new(duration, 6, app_name, stream_name, need_record),
         }
     }
 
@@ -173,3 +171,38 @@ impl Flv2HlsRemuxer {
         self.m3u8_handler.clear()
     }
 }
+#[cfg(test)]
+mod tests {
+    // use std::{
+    //     env,
+    //     fs::{self},
+    // };
+
+    // #[test]
+    // fn test_new_path() {
+    //     if let Ok(current_dir) = env::current_dir() {
+    //         println!("Current directory: {:?}", current_dir);
+    //     } else {
+    //         eprintln!("Failed to get the current directory");
+    //     }
+    //     let directory = "test";
+
+    //     if !fs::metadata(directory).is_ok() {
+    //         match fs::create_dir(directory) {
+    //             Ok(_) => println!("目录已创建"),
+    //             Err(err) => println!("创建目录时出错：{:?}", err),
+    //         }
+    //     } else {
+    //         println!("目录已存在");
+    //     }
+    // }
+    // #[test]
+    // fn test_copy() {
+    //     let path = "./aa.txt";
+    //     if let Err(err) = fs::copy(path, "./test/") {
+    //         println!("copy err: {err}");
+    //     } else {
+    //         println!("copy success");
+    //     }
+    // }
+}
diff --git a/protocol/hls/src/flv_data_receiver.rs b/protocol/hls/src/flv_data_receiver.rs
@@ -22,7 +22,6 @@ use {
 pub struct FlvDataReceiver {
     app_name: String,
     stream_name: String,
-
     event_producer: StreamHubEventSender,
     data_consumer: FrameDataReceiver,
     media_processor: Flv2HlsRemuxer,
@@ -34,19 +33,18 @@ impl FlvDataReceiver {
         app_name: String,
         stream_name: String,
         event_producer: StreamHubEventSender,
-
         duration: i64,
+        need_record: bool,
     ) -> Self {
         let (_, data_consumer) = mpsc::unbounded_channel();
         let subscriber_id = Uuid::new(RandomDigitCount::Four);
 
         Self {
             app_name: app_name.clone(),
             stream_name: stream_name.clone(),
-
             data_consumer,
             event_producer,
-            media_processor: Flv2HlsRemuxer::new(duration, app_name, stream_name),
+            media_processor: Flv2HlsRemuxer::new(duration, app_name, stream_name, need_record),
             subscriber_id,
         }
     }
diff --git a/protocol/hls/src/lib.rs b/protocol/hls/src/lib.rs
diff --git a/protocol/hls/src/m3u8.rs b/protocol/hls/src/m3u8.rs
diff --git a/protocol/hls/src/remuxer.rs b/protocol/hls/src/remuxer.rs
diff --git a/protocol/hls/src/test_flv2hls.rs b/protocol/hls/src/test_flv2hls.rs
diff --git a/protocol/hls/src/ts.rs b/protocol/hls/src/ts.rs
