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