hls 0.5.5

hls library.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164

use tokio::sync::oneshot;

use {
    super::{
        errors::{HlsError, HlsErrorValue},
        flv2hls::Flv2HlsRemuxer,
    },
    std::time::Duration,
    streamhub::{
        define::{
            FrameData, FrameDataReceiver, NotifyInfo, StreamHubEvent, StreamHubEventSender,
            SubscribeType, SubscriberInfo,
        },
        stream::StreamIdentifier,
        utils::{RandomDigitCount, Uuid},
    },
    tokio::{sync::mpsc, time::sleep},
    xflv::define::FlvData,
};

///https://www.jianshu.com/p/d6311f03b81f

pub struct FlvDataReceiver {
    app_name: String,
    stream_name: String,
    event_producer: StreamHubEventSender,
    data_consumer: FrameDataReceiver,
    media_processor: Flv2HlsRemuxer,
    subscriber_id: Uuid,
}

impl FlvDataReceiver {
    pub fn new(
        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, need_record),
            subscriber_id,
        }
    }

    pub async fn run(&mut self) -> Result<(), HlsError> {
        self.subscribe_from_stream_hub(self.app_name.clone(), self.stream_name.clone())
            .await?;
        self.receive_flv_data().await?;

        Ok(())
    }

    pub async fn receive_flv_data(&mut self) -> Result<(), HlsError> {
        let mut retry_count = 0;

        loop {
            if let Some(data) = self.data_consumer.recv().await {
                let flv_data: FlvData = match data {
                    FrameData::Audio { timestamp, data } => FlvData::Audio { timestamp, data },
                    FrameData::Video { timestamp, data } => FlvData::Video { timestamp, data },
                    _ => continue,
                };
                retry_count = 0;
                self.media_processor.process_flv_data(flv_data)?;
            } else {
                sleep(Duration::from_millis(100)).await;
                retry_count += 1;
            }
            //When rtmp stream is interupted here we retry 10 times.
            //maybe have a better way to judge the stream status.
            //will do an optimization in the future.
            //todo
            if retry_count > 10 {
                self.media_processor.flush_remaining_data()?;
                break;
            }
        }

        self.media_processor.clear_files()?;
        self.unsubscribe_from_stream_hub().await
    }

    pub fn flush_response_data(&mut self) -> Result<(), HlsError> {
        Ok(())
    }

    pub async fn subscribe_from_stream_hub(
        &mut self,
        app_name: String,
        stream_name: String,
    ) -> Result<(), HlsError> {
        /*the sub info is only used to transfer from RTMP to HLS, but not for client player */
        let sub_info = SubscriberInfo {
            id: self.subscriber_id,
            sub_type: SubscribeType::RtmpRemux2Hls,
            sub_data_type: streamhub::define::SubDataType::Frame,
            notify_info: NotifyInfo {
                request_url: String::from(""),
                remote_addr: String::from(""),
            },
        };

        let identifier = StreamIdentifier::Rtmp {
            app_name,
            stream_name,
        };

        let (event_result_sender, event_result_receiver) = oneshot::channel();

        let subscribe_event = StreamHubEvent::Subscribe {
            identifier,
            info: sub_info,
            result_sender: event_result_sender,
        };

        let rv = self.event_producer.send(subscribe_event);
        if rv.is_err() {
            return Err(HlsError {
                value: HlsErrorValue::StreamHubEventSendErr,
            });
        }

        let receiver = event_result_receiver.await??.0.frame_receiver.unwrap();

        self.data_consumer = receiver;

        Ok(())
    }

    pub async fn unsubscribe_from_stream_hub(&mut self) -> Result<(), HlsError> {
        let sub_info = SubscriberInfo {
            id: self.subscriber_id,
            sub_type: SubscribeType::RtmpRemux2Hls,
            sub_data_type: streamhub::define::SubDataType::Frame,
            notify_info: NotifyInfo {
                request_url: String::from(""),
                remote_addr: String::from(""),
            },
        };

        let identifier = StreamIdentifier::Rtmp {
            app_name: self.app_name.clone(),
            stream_name: self.stream_name.clone(),
        };

        let subscribe_event = StreamHubEvent::UnSubscribe {
            identifier,
            info: sub_info,
        };
        if let Err(err) = self.event_producer.send(subscribe_event) {
            log::error!("unsubscribe_from_stream_hub err {}", err);
        }

        Ok(())
    }
}