xref: /xiu/protocol/rtmp/src/session/common.rs (revision 2f7fa101) 
1 use {
2     super::{
3         define::{SessionSubType, SessionType},
4         errors::{SessionError, SessionErrorValue},
5     },
6     crate::{
7         channels::define::{
8             ChannelData, ChannelDataConsumer, ChannelDataProducer, ChannelEvent,
9             ChannelEventProducer,
10         },
11         chunk::{
12             define::{chunk_type, csid_type},
13             packetizer::ChunkPacketizer,
14             ChunkInfo,
15         },
16         messages::define::msg_type_id,
17     },
18     bytes::BytesMut,
19     bytesio::bytesio::BytesIO,
20     std::{sync::Arc, time::Duration},
21     tokio::{
22         sync::{mpsc, oneshot, Mutex},
23         time::sleep,
24     },
25     uuid::Uuid,
26 };
27 #[derive(Debug)]
28 pub struct SessionInfo {
29     pub subscriber_id: Uuid,
30     pub session_sub_type: SessionSubType,
31 }
32 pub struct Common {
33     packetizer: ChunkPacketizer,
34 
35     data_consumer: ChannelDataConsumer,
36     data_producer: ChannelDataProducer,
37 
38     event_producer: ChannelEventProducer,
39     pub session_type: SessionType,
40 }
41 
42 impl Common {
43     pub fn new(
44         net_io: Arc<Mutex<BytesIO>>,
45         event_producer: ChannelEventProducer,
46         session_type: SessionType,
47     ) -> Self {
48         //only used for init,since I don't found a better way to deal with this.
49         let (init_producer, init_consumer) = mpsc::unbounded_channel();
50 
51         Self {
52             packetizer: ChunkPacketizer::new(Arc::clone(&net_io)),
53 
54             data_producer: init_producer,
55             data_consumer: init_consumer,
56 
57             event_producer,
58             session_type,
59         }
60     }
61     pub async fn send_channel_data(&mut self) -> Result<(), SessionError> {
62         let mut retry_times = 0;
63         loop {
64             if let Some(data) = self.data_consumer.recv().await {
65                 match data {
66                     ChannelData::Audio { timestamp, data } => {
67                         self.send_audio(data, timestamp).await?;
68                     }
69                     ChannelData::Video { timestamp, data } => {
70                         self.send_video(data, timestamp).await?;
71                     }
72                     ChannelData::MetaData { timestamp, data } => {
73                         self.send_metadata(data, timestamp).await?;
74                     }
75                 }
76             } else {
77                 retry_times += 1;
78                 log::debug!(
79                     "send_channel_data: no data receives ,retry {} times!",
80                     retry_times
81                 );
82 
83                 if retry_times > 10 {
84                     return Err(SessionError {
85                         value: SessionErrorValue::NoMediaDataReceived,
86                     });
87                 }
88             }
89         }
90     }
91 
92     pub async fn send_audio(&mut self, data: BytesMut, timestamp: u32) -> Result<(), SessionError> {
93         let mut chunk_info = ChunkInfo::new(
94             csid_type::AUDIO,
95             chunk_type::TYPE_0,
96             timestamp,
97             data.len() as u32,
98             msg_type_id::AUDIO,
99             0,
100             data,
101         );
102 
103         self.packetizer.write_chunk(&mut chunk_info).await?;
104 
105         Ok(())
106     }
107 
108     pub async fn send_video(&mut self, data: BytesMut, timestamp: u32) -> Result<(), SessionError> {
109         let mut chunk_info = ChunkInfo::new(
110             csid_type::VIDEO,
111             chunk_type::TYPE_0,
112             timestamp,
113             data.len() as u32,
114             msg_type_id::VIDEO,
115             0,
116             data,
117         );
118 
119         self.packetizer.write_chunk(&mut chunk_info).await?;
120 
121         Ok(())
122     }
123 
124     pub async fn send_metadata(
125         &mut self,
126         data: BytesMut,
127         timestamp: u32,
128     ) -> Result<(), SessionError> {
129         let mut chunk_info = ChunkInfo::new(
130             csid_type::DATA_AMF0_AMF3,
131             chunk_type::TYPE_0,
132             timestamp,
133             data.len() as u32,
134             msg_type_id::DATA_AMF0,
135             0,
136             data,
137         );
138 
139         self.packetizer.write_chunk(&mut chunk_info).await?;
140         Ok(())
141     }
142 
143     pub fn on_video_data(
144         &mut self,
145         data: &mut BytesMut,
146         timestamp: &u32,
147     ) -> Result<(), SessionError> {
148         let data = ChannelData::Video {
149             timestamp: timestamp.clone(),
150             data: data.clone(),
151         };
152 
153         match self.data_producer.send(data) {
154             Ok(_) => {}
155             Err(err) => {
156                 log::error!("send video err: {}", err);
157                 return Err(SessionError {
158                     value: SessionErrorValue::SendChannelDataErr,
159                 });
160             }
161         }
162 
163         Ok(())
164     }
165 
166     pub fn on_audio_data(
167         &mut self,
168         data: &mut BytesMut,
169         timestamp: &u32,
170     ) -> Result<(), SessionError> {
171         let data = ChannelData::Audio {
172             timestamp: timestamp.clone(),
173             data: data.clone(),
174         };
175 
176         match self.data_producer.send(data) {
177             Ok(_) => {}
178             Err(err) => {
179                 log::error!("receive audio err {}\n", err);
180                 return Err(SessionError {
181                     value: SessionErrorValue::SendChannelDataErr,
182                 });
183             }
184         }
185 
186         Ok(())
187     }
188 
189     pub fn on_meta_data(
190         &mut self,
191         body: &mut BytesMut,
192         timestamp: &u32,
193     ) -> Result<(), SessionError> {
194         let data = ChannelData::MetaData {
195             timestamp: timestamp.clone(),
196             data: body.clone(),
197         };
198 
199         match self.data_producer.send(data) {
200             Ok(_) => {}
201             Err(_) => {
202                 return Err(SessionError {
203                     value: SessionErrorValue::SendChannelDataErr,
204                 })
205             }
206         }
207 
208         Ok(())
209     }
210 
211     fn get_session_info(&mut self, sub_id: Uuid) -> SessionInfo {
212         match self.session_type {
213             SessionType::Client => SessionInfo {
214                 subscriber_id: sub_id,
215                 session_sub_type: SessionSubType::Publisher,
216             },
217             SessionType::Server => SessionInfo {
218                 subscriber_id: sub_id,
219                 session_sub_type: SessionSubType::Player,
220             },
221         }
222     }
223 
224     /*Begin to send data to common player or relay pull client*/
225     pub async fn subscribe_from_channels(
226         &mut self,
227         app_name: String,
228         stream_name: String,
229         sub_id: Uuid,
230     ) -> Result<(), SessionError> {
231         log::info!(
232             "subscribe_from_channels, app_name: {} stream_name: {} subscribe_id: {}",
233             app_name,
234             stream_name.clone(),
235             sub_id
236         );
237 
238         let mut retry_count: u8 = 0;
239 
240         loop {
241             let (sender, receiver) = oneshot::channel();
242 
243             let subscribe_event = ChannelEvent::Subscribe {
244                 app_name: app_name.clone(),
245                 stream_name: stream_name.clone(),
246                 session_info: self.get_session_info(sub_id),
247                 responder: sender,
248             };
249             let rv = self.event_producer.send(subscribe_event);
250             match rv {
251                 Err(_) => {
252                     return Err(SessionError {
253                         value: SessionErrorValue::ChannelEventSendErr,
254                     })
255                 }
256                 _ => {}
257             }
258 
259             match receiver.await {
260                 Ok(consumer) => {
261                     self.data_consumer = consumer;
262                     break;
263                 }
264                 Err(_) => {
265                     if retry_count > 10 {
266                         return Err(SessionError {
267                             value: SessionErrorValue::SubscribeCountLimitReach,
268                         });
269                     }
270                 }
271             }
272 
273             sleep(Duration::from_millis(800)).await;
274             retry_count = retry_count + 1;
275         }
276 
277         Ok(())
278     }
279 
280     pub async fn unsubscribe_from_channels(
281         &mut self,
282         app_name: String,
283         stream_name: String,
284         sub_id: Uuid,
285     ) -> Result<(), SessionError> {
286         let subscribe_event = ChannelEvent::UnSubscribe {
287             app_name,
288             stream_name,
289             session_info: self.get_session_info(sub_id),
290         };
291         if let Err(err) = self.event_producer.send(subscribe_event) {
292             log::error!("unsubscribe_from_channels err {}\n", err);
293         }
294 
295         Ok(())
296     }
297 
298     /*Begin to receive stream data from RTMP push client or RTMP relay push client*/
299     pub async fn publish_to_channels(
300         &mut self,
301         app_name: String,
302         stream_name: String,
303     ) -> Result<(), SessionError> {
304         let (sender, receiver) = oneshot::channel();
305         let publish_event = ChannelEvent::Publish {
306             app_name,
307             stream_name,
308             responder: sender,
309         };
310 
311         let rv = self.event_producer.send(publish_event);
312         match rv {
313             Err(_) => {
314                 return Err(SessionError {
315                     value: SessionErrorValue::ChannelEventSendErr,
316                 })
317             }
318             _ => {}
319         }
320 
321         match receiver.await {
322             Ok(producer) => {
323                 self.data_producer = producer;
324             }
325             Err(err) => {
326                 log::error!("publish_to_channels err{}\n", err);
327             }
328         }
329         Ok(())
330     }
331 
332     pub async fn unpublish_to_channels(
333         &mut self,
334         app_name: String,
335         stream_name: String,
336     ) -> Result<(), SessionError> {
337         let unpublish_event = ChannelEvent::UnPublish {
338             app_name: app_name.clone(),
339             stream_name: stream_name.clone(),
340         };
341 
342         let rv = self.event_producer.send(unpublish_event);
343         match rv {
344             Err(_) => {
345                 log::error!(
346                     "unpublish_to_channels error.app_name: {}, stream_name: {}",
347                     app_name,
348                     stream_name
349                 );
350                 return Err(SessionError {
351                     value: SessionErrorValue::ChannelEventSendErr,
352                 });
353             }
354             _ => {
355                 log::info!(
356                     "unpublish_to_channels successfully.app_name: {}, stream_name: {}",
357                     app_name,
358                     stream_name
359                 );
360             }
361         }
362         Ok(())
363     }
364 }
365