Merge pull request #175 from nikitos/hls_notify

add on_hls notification hook

Author: harlanc

application/xiu/src/config/mod.rs

@@ -189,6 +189,7 @@ pub struct HttpNotifierConfig {
     pub on_unpublish: Option<String>,
     pub on_play: Option<String>,
     pub on_stop: Option<String>,
+    pub on_hls: Option<String>,
 }
 
 #[derive(Debug, Deserialize, Clone, Default)]

application/xiu/src/service.rs

@@ -72,6 +72,7 @@ impl Service {
                     httpnotifier.on_unpublish.clone(),
                     httpnotifier.on_play.clone(),
                     httpnotifier.on_stop.clone(),
+                    httpnotifier.on_hls.clone(), 
                 )))
             }
         } else {

library/streamhub/src/define.rs

@@ -318,6 +318,10 @@ pub enum StreamHubEvent {
         identifier: StreamIdentifier,
         sender: InformationSender,
     },
+    OnHls {
+        identifier: StreamIdentifier,
+        segment: Segment,
+    }
 }
 
 impl StreamHubEvent {
@@ -445,3 +449,33 @@ pub enum StatisticData {
         start_time: DateTime<Local>,
     },
 }
+
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Segment {
+    /*ts duration*/
+    pub duration: i64,
+    pub discontinuity: bool,
+    /*ts name*/
+    pub name: String,
+    pub path: String,
+    pub is_eof: bool,
+}
+
+impl Segment {
+    pub fn new(
+        duration: i64,
+        discontinuity: bool,
+        name: String,
+        path: String,
+        is_eof: bool,
+    ) -> Self {
+        Self {
+            duration,
+            discontinuity,
+            name,
+            path,
+            is_eof,
+        }
+    }
+}

library/streamhub/src/lib.rs

@@ -774,6 +774,11 @@ impl StreamsHub {
                         log::error!("event_loop request error: {}", err);
                     }
                 }
+                StreamHubEvent::OnHls { identifier: _ , segment: _ } => {
+                    if let Some(notifier) = &self.notifier {
+                        notifier.on_hls_notify(&message).await;
+                    } 
+                }
             }
         }
     }

library/streamhub/src/notify/http.rs

@@ -23,6 +23,7 @@ pub struct HttpNotifier {
     on_unpublish_url: Option<String>,
     on_play_url: Option<String>,
     on_stop_url: Option<String>,
+    on_hls_url: Option<String>,
 }
 
 impl HttpNotifier {
@@ -31,13 +32,16 @@ impl HttpNotifier {
         on_unpublish_url: Option<String>,
         on_play_url: Option<String>,
         on_stop_url: Option<String>,
+        on_hls_url: Option<String>,
+
     ) -> Self {
         Self {
             request_client: reqwest::Client::new(),
             on_publish_url,
             on_unpublish_url,
             on_play_url,
             on_stop_url,
+            on_hls_url,
         }
     }
 }
@@ -119,4 +123,23 @@ impl Notifier for HttpNotifier {
             }
         }
     }
+
+    async fn on_hls_notify(&self, event: &StreamHubEventMessage) {
+        if let Some(on_hls_url) = &self.on_hls_url {
+            match self
+                .request_client
+                .post(on_hls_url)
+                .body(serialize_event!(event))
+                .send()
+                .await
+            {
+                Err(err) => {
+                    log::error!("on_hls error: {}", err);
+                }
+                Ok(response) => {
+                    log::info!("on_hls success: {:?}", response);
+                }
+            }
+        }
+    }
 }

library/streamhub/src/notify/mod.rs

@@ -9,4 +9,5 @@ pub trait Notifier: Sync + Send {
     async fn on_unpublish_notify(&self, event: &StreamHubEventMessage);
     async fn on_play_notify(&self, event: &StreamHubEventMessage);
     async fn on_stop_notify(&self, event: &StreamHubEventMessage);
+    async fn on_hls_notify(&self, event: &StreamHubEventMessage);
 }

protocol/hls/src/flv2hls.rs

@@ -5,6 +5,7 @@ use {
         define::{frame_type, FlvData},
         demuxer::{FlvAudioTagDemuxer, FlvVideoTagDemuxer},
     },
+    streamhub::{define::{StreamHubEventSender, StreamHubEvent}, stream::StreamIdentifier},
     xmpegts::{
         define::{epsi_stream_type, MPEG_FLAG_IDR_FRAME},
         ts::TsMuxer,
@@ -30,7 +31,11 @@ pub struct Flv2HlsRemuxer {
     audio_pid: u16,
 
     m3u8_handler: M3u8,
+    event_producer: Option<StreamHubEventSender>,
+    app_name: String,
+    stream_name: String,
     aof_ratio: i64,
+
 }
 
 impl Flv2HlsRemuxer {
@@ -39,6 +44,7 @@ impl Flv2HlsRemuxer {
         app_name: String,
         stream_name: String,
         need_record: bool,
+        event_producer: Option<StreamHubEventSender>,
         path: String,
         aof_ratio: i64,
     ) -> Self {
@@ -68,8 +74,10 @@ impl Flv2HlsRemuxer {
             video_pid,
             audio_pid,
 
-            m3u8_handler: M3u8::new(duration, 6, app_name, stream_name, need_record, path),
-
+            m3u8_handler: M3u8::new(duration, 6, app_name.clone(), stream_name.clone(), need_record, path),
+            event_producer,
+            app_name,
+            stream_name,
             aof_ratio,
         }
     }
@@ -162,6 +170,18 @@ impl Flv2HlsRemuxer {
             }
             let data = self.ts_muxer.get_data();
 
+            if let Some(segment) = self.m3u8_handler.segments.back() {
+                let identifier = StreamIdentifier::Rtmp { app_name: self.app_name.clone(), stream_name: self.stream_name.clone() };
+                let hub_event = StreamHubEvent::OnHls { identifier: identifier.clone(), segment: segment.clone() };
+                if let Some(producer) = self.event_producer.clone() {
+                    if let Err(err) = producer.send(hub_event) {
+                        log::error!("send notify on_hls event error: {}", err);
+                    }
+                } else {
+                    log::warn!("event_producer is None, cannot send on_hls event");
+                }
+                log::info!("on_hls success: {:?}", identifier);
+            }
             self.m3u8_handler
                 .add_segment(dts - self.last_ts_dts, discontinuity, false, data)?;
             self.m3u8_handler.refresh_playlist()?;

protocol/hls/src/flv_data_receiver.rs

@@ -46,8 +46,8 @@ impl FlvDataReceiver {
             app_name: app_name.clone(),
             stream_name: stream_name.clone(),
             data_consumer,
-            event_producer,
-            media_processor: Flv2HlsRemuxer::new(duration, app_name, stream_name, need_record, path, aof_ratio),
+            event_producer: event_producer.clone(),
+            media_processor: Flv2HlsRemuxer::new(duration, app_name, stream_name, need_record, Some(event_producer), path, aof_ratio),
             subscriber_id,
         }
     }

protocol/hls/src/m3u8.rs

@@ -2,35 +2,9 @@ use {
     super::{errors::MediaError, ts::Ts},
     bytes::BytesMut,
     std::{collections::VecDeque, fs, fs::File, io::Write},
+    streamhub::define::Segment,
 };
 
-pub struct Segment {
-    /*ts duration*/
-    pub duration: i64,
-    pub discontinuity: bool,
-    /*ts name*/
-    pub name: String,
-    path: String,
-    pub is_eof: bool,
-}
-
-impl Segment {
-    pub fn new(
-        duration: i64,
-        discontinuity: bool,
-        name: String,
-        path: String,
-        is_eof: bool,
-    ) -> Self {
-        Self {
-            duration,
-            discontinuity,
-            name,
-            path,
-            is_eof,
-        }
-    }
-}
 
 pub struct M3u8 {
     version: u16,
@@ -43,7 +17,7 @@ pub struct M3u8 {
     The normal recommendation is 3, but the optimum number may be larger.*/
     live_ts_count: usize,
 
-    segments: VecDeque<Segment>,
+    pub segments: VecDeque<Segment>,
 
     m3u8_folder: String,
     live_m3u8_name: String,

protocol/hls/src/test_flv2hls.rs

@@ -65,7 +65,7 @@ mod tests {
 
         let start = Instant::now();
         let mut media_demuxer =
-            Flv2HlsRemuxer::new(5, String::from("live"), String::from("test"), false, String::from("./"), 1);
+            Flv2HlsRemuxer::new(5, String::from("live"), String::from("test"), false, None, String::from("./"), 1);
 
         loop {
             let data_ = demuxer.read_flv_tag();