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