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 { body } => { 71 self.send_metadata(body).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(&mut self, data: BytesMut) -> Result<(), SessionError> { 111 let mut chunk_info = ChunkInfo::new( 112 csid_type::DATA_AMF0_AMF3, 113 chunk_type::TYPE_0, 114 0, 115 data.len() as u32, 116 msg_type_id::DATA_AMF0, 117 0, 118 data, 119 ); 120 121 self.packetizer.write_chunk(&mut chunk_info).await?; 122 Ok(()) 123 } 124 125 pub fn on_video_data( 126 &mut self, 127 data: &mut BytesMut, 128 timestamp: &u32, 129 ) -> Result<(), SessionError> { 130 let data = ChannelData::Video { 131 timestamp: timestamp.clone(), 132 data: data.clone(), 133 }; 134 135 //print!("receive video data\n"); 136 match self.data_producer.send(data) { 137 Ok(_) => {} 138 Err(err) => { 139 print!("send video err {}\n", err); 140 return Err(SessionError { 141 value: SessionErrorValue::SendChannelDataErr, 142 }); 143 } 144 } 145 146 Ok(()) 147 } 148 149 pub fn on_audio_data( 150 &mut self, 151 data: &mut BytesMut, 152 timestamp: &u32, 153 ) -> Result<(), SessionError> { 154 let data = ChannelData::Audio { 155 timestamp: timestamp.clone(), 156 data: data.clone(), 157 }; 158 159 match self.data_producer.send(data) { 160 Ok(_) => {} 161 Err(err) => { 162 print!("receive audio err {}\n", err); 163 return Err(SessionError { 164 value: SessionErrorValue::SendChannelDataErr, 165 }); 166 } 167 } 168 169 Ok(()) 170 } 171 172 pub fn on_meta_data(&mut self, body: &mut BytesMut) -> Result<(), SessionError> { 173 let data = ChannelData::MetaData { body: body.clone() }; 174 175 match self.data_producer.send(data) { 176 Ok(_) => {} 177 Err(_) => { 178 return Err(SessionError { 179 value: SessionErrorValue::SendChannelDataErr, 180 }) 181 } 182 } 183 184 Ok(()) 185 } 186 187 fn get_session_info(&mut self, session_id: u64) -> SessionInfo { 188 match self.session_type { 189 SessionType::Client => SessionInfo { 190 session_id: session_id, 191 session_sub_type: SessionSubType::Publisher, 192 }, 193 SessionType::Server => SessionInfo { 194 session_id: session_id, 195 session_sub_type: SessionSubType::Player, 196 }, 197 } 198 } 199 200 pub async fn subscribe_from_channels( 201 &mut self, 202 app_name: String, 203 stream_name: String, 204 session_id: u64, 205 ) -> Result<(), SessionError> { 206 print!( 207 "subscribe info............{} {} {}\n", 208 app_name, 209 stream_name.clone(), 210 session_id 211 ); 212 213 let mut retry_count: u8 = 0; 214 215 loop { 216 let (sender, receiver) = oneshot::channel(); 217 218 let subscribe_event = ChannelEvent::Subscribe { 219 app_name: app_name.clone(), 220 stream_name: stream_name.clone(), 221 session_info: self.get_session_info(session_id), 222 responder: sender, 223 }; 224 let rv = self.event_producer.send(subscribe_event); 225 match rv { 226 Err(_) => { 227 return Err(SessionError { 228 value: SessionErrorValue::ChannelEventSendErr, 229 }) 230 } 231 _ => {} 232 } 233 234 match receiver.await { 235 Ok(consumer) => { 236 self.data_consumer = consumer; 237 break; 238 } 239 Err(_) => { 240 if retry_count > 10 { 241 return Err(SessionError { 242 value: SessionErrorValue::SubscribeCountLimitReach, 243 }); 244 } 245 } 246 } 247 248 sleep(Duration::from_millis(800)).await; 249 retry_count = retry_count + 1; 250 } 251 252 Ok(()) 253 } 254 255 pub async fn unsubscribe_from_channels( 256 &mut self, 257 app_name: String, 258 stream_name: String, 259 session_id: u64, 260 ) -> Result<(), SessionError> { 261 let subscribe_event = ChannelEvent::UnSubscribe { 262 app_name, 263 stream_name, 264 session_info: self.get_session_info(session_id), 265 }; 266 if let Err(err) = self.event_producer.send(subscribe_event) { 267 print!("unsubscribe_from_channels err {}\n", err) 268 } 269 270 Ok(()) 271 } 272 273 pub async fn publish_to_channels( 274 &mut self, 275 app_name: String, 276 stream_name: String, 277 ) -> Result<(), SessionError> { 278 let (sender, receiver) = oneshot::channel(); 279 let publish_event = ChannelEvent::Publish { 280 app_name, 281 stream_name, 282 responder: sender, 283 }; 284 285 let rv = self.event_producer.send(publish_event); 286 match rv { 287 Err(_) => { 288 return Err(SessionError { 289 value: SessionErrorValue::ChannelEventSendErr, 290 }) 291 } 292 _ => {} 293 } 294 295 match receiver.await { 296 Ok(producer) => { 297 //print!("set producer before\n"); 298 self.data_producer = producer; 299 //print!("set producer after\n"); 300 } 301 Err(err) => { 302 print!("publish_to_channels err{}\n", err) 303 } 304 } 305 Ok(()) 306 } 307 308 pub async fn unpublish_to_channels( 309 &mut self, 310 app_name: String, 311 stream_name: String, 312 ) -> Result<(), SessionError> { 313 let unpublish_event = ChannelEvent::UnPublish { 314 app_name, 315 stream_name, 316 }; 317 318 let rv = self.event_producer.send(unpublish_event); 319 match rv { 320 Err(_) => { 321 println!("unpublish_to_channels error."); 322 return Err(SessionError { 323 value: SessionErrorValue::ChannelEventSendErr, 324 }); 325 } 326 _ => { 327 println!("unpublish_to_channels successfully.") 328 } 329 } 330 Ok(()) 331 } 332 } 333