1 use {
2     super::{
3         common::Common,
4         define,
5         define::SessionType,
6         errors::{SessionError, SessionErrorValue},
7     },
8     crate::{
9         amf0::Amf0ValueType,
10         channels::define::ChannelEventProducer,
11         chunk::{
12             define::{chunk_type, csid_type, CHUNK_SIZE},
13             packetizer::ChunkPacketizer,
14             unpacketizer::{ChunkUnpacketizer, UnpackResult},
15             ChunkInfo,
16         },
17         config,
18         handshake::handshake::{HandshakeServer, ServerHandshakeState},
19         messages::{
20             define::{msg_type_id, RtmpMessageData},
21             parser::MessageParser,
22         },
23         netconnection::commands::NetConnection,
24         netstream::writer::NetStreamWriter,
25         protocol_control_messages::writer::ProtocolControlMessagesWriter,
26         user_control_messages::writer::EventMessagesWriter,
27     },
28     bytes::BytesMut,
29     bytesio::{
30         bytes_writer::{AsyncBytesWriter, BytesWriter},
31         bytesio::BytesIO,
32     },
33     std::{collections::HashMap, sync::Arc},
34     tokio::{net::TcpStream, sync::Mutex},
35 };
36 
37 enum ServerSessionState {
38     Handshake,
39     ReadChunk,
40     // OnConnect,
41     // OnCreateStream,
42     //Publish,
43     Play,
44 }
45 
46 pub struct ServerSession {
47     pub app_name: String,
48     pub stream_name: String,
49 
50     io: Arc<Mutex<BytesIO>>,
51     handshaker: HandshakeServer,
52 
53     packetizer: ChunkPacketizer,
54     unpacketizer: ChunkUnpacketizer,
55 
56     state: ServerSessionState,
57 
58     pub common: Common,
59 
60     bytesio_data: BytesMut,
61     need_process: bool,
62 
63     pub session_id: u64,
64 
65     connect_command_object: Option<HashMap<String, Amf0ValueType>>,
66 }
67 
68 impl ServerSession {
69     pub fn new(stream: TcpStream, event_producer: ChannelEventProducer, session_id: u64) -> Self {
70         let net_io = Arc::new(Mutex::new(BytesIO::new(stream)));
71 
72         Self {
73             app_name: String::from(""),
74             stream_name: String::from(""),
75 
76             io: Arc::clone(&net_io),
77             handshaker: HandshakeServer::new(Arc::clone(&net_io)),
78 
79             packetizer: ChunkPacketizer::new(Arc::clone(&net_io)),
80             unpacketizer: ChunkUnpacketizer::new(),
81 
82             state: ServerSessionState::Handshake,
83 
84             common: Common::new(Arc::clone(&net_io), event_producer, SessionType::Server),
85 
86             session_id: session_id,
87             bytesio_data: BytesMut::new(),
88             need_process: false,
89 
90             connect_command_object: None,
91         }
92     }
93 
94     pub async fn run(&mut self) -> Result<(), SessionError> {
95         loop {
96             match self.state {
97                 ServerSessionState::Handshake => {
98                     self.handshake().await?;
99                 }
100                 ServerSessionState::ReadChunk => {
101                     self.read_parse_chunks().await?;
102                 }
103                 ServerSessionState::Play => {
104                     self.play().await?;
105                 }
106             }
107         }
108 
109         //Ok(())
110     }
111 
112     async fn handshake(&mut self) -> Result<(), SessionError> {
113         self.bytesio_data = self.io.lock().await.read().await?;
114         self.handshaker.extend_data(&self.bytesio_data[..]);
115         self.handshaker.handshake().await?;
116 
117         match self.handshaker.state() {
118             ServerHandshakeState::Finish => {
119                 self.state = ServerSessionState::ReadChunk;
120 
121                 let left_bytes = self.handshaker.get_remaining_bytes();
122                 if left_bytes.len() > 0 {
123                     self.unpacketizer.extend_data(&left_bytes[..]);
124                     self.need_process = true;
125                 }
126                 self.send_set_chunk_size().await?;
127                 return Ok(());
128             }
129             _ => {}
130         }
131 
132         Ok(())
133     }
134 
135     async fn read_parse_chunks(&mut self) -> Result<(), SessionError> {
136         if !self.need_process {
137             self.bytesio_data = self.io.lock().await.read().await?;
138             self.unpacketizer.extend_data(&self.bytesio_data[..]);
139         }
140 
141         self.need_process = false;
142 
143         loop {
144             let result = self.unpacketizer.read_chunks();
145 
146             if let Ok(rv) = result {
147                 match rv {
148                     UnpackResult::Chunks(chunks) => {
149                         for chunk_info in chunks.iter() {
150                             let mut msg = MessageParser::new(chunk_info.clone()).parse()?;
151 
152                             let msg_stream_id = chunk_info.message_header.msg_streamd_id;
153                             let timestamp = chunk_info.message_header.timestamp;
154                             self.process_messages(&mut msg, &msg_stream_id, &timestamp)
155                                 .await?;
156                         }
157                     }
158                     _ => {}
159                 }
160             } else {
161                 break;
162             }
163         }
164         Ok(())
165     }
166 
167     async fn play(&mut self) -> Result<(), SessionError> {
168         match self.common.send_channel_data().await {
169             Ok(_) => {}
170 
171             Err(err) => {
172                 self.common
173                     .unsubscribe_from_channels(
174                         self.app_name.clone(),
175                         self.stream_name.clone(),
176                         self.session_id,
177                     )
178                     .await?;
179                 return Err(err);
180             }
181         }
182 
183         Ok(())
184     }
185 
186     pub async fn send_set_chunk_size(&mut self) -> Result<(), SessionError> {
187         let mut controlmessage =
188             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
189         controlmessage.write_set_chunk_size(CHUNK_SIZE).await?;
190 
191         Ok(())
192     }
193 
194     pub async fn process_messages(
195         &mut self,
196         rtmp_msg: &mut RtmpMessageData,
197         msg_stream_id: &u32,
198         timestamp: &u32,
199     ) -> Result<(), SessionError> {
200         match rtmp_msg {
201             RtmpMessageData::Amf0Command {
202                 command_name,
203                 transaction_id,
204                 command_object,
205                 others,
206             } => {
207                 self.on_amf0_command_message(
208                     msg_stream_id,
209                     command_name,
210                     transaction_id,
211                     command_object,
212                     others,
213                 )
214                 .await?
215             }
216             RtmpMessageData::SetChunkSize { chunk_size } => {
217                 self.on_set_chunk_size(chunk_size.clone() as usize)?;
218             }
219             RtmpMessageData::AudioData { data } => {
220                 self.common.on_audio_data(data, timestamp)?;
221             }
222             RtmpMessageData::VideoData { data } => {
223                 self.common.on_video_data(data, timestamp)?;
224             }
225             RtmpMessageData::AmfData { raw_data } => {
226                 self.common.on_meta_data(raw_data, timestamp)?;
227             }
228 
229             _ => {}
230         }
231         Ok(())
232     }
233 
234     pub async fn on_amf0_command_message(
235         &mut self,
236         stream_id: &u32,
237         command_name: &Amf0ValueType,
238         transaction_id: &Amf0ValueType,
239         command_object: &Amf0ValueType,
240         others: &mut Vec<Amf0ValueType>,
241     ) -> Result<(), SessionError> {
242         let empty_cmd_name = &String::new();
243         let cmd_name = match command_name {
244             Amf0ValueType::UTF8String(str) => str,
245             _ => empty_cmd_name,
246         };
247 
248         let transaction_id = match transaction_id {
249             Amf0ValueType::Number(number) => number,
250             _ => &0.0,
251         };
252 
253         let empty_cmd_obj: HashMap<String, Amf0ValueType> = HashMap::new();
254         let obj = match command_object {
255             Amf0ValueType::Object(obj) => obj,
256             _ => &empty_cmd_obj,
257         };
258 
259         match cmd_name.as_str() {
260             "connect" => {
261                 self.on_connect(&transaction_id, &obj).await?;
262             }
263             "createStream" => {
264                 self.on_create_stream(transaction_id).await?;
265             }
266             "deleteStream" => {
267                 if others.len() > 0 {
268                     let stream_id = match others.pop() {
269                         Some(val) => match val {
270                             Amf0ValueType::Number(streamid) => streamid,
271                             _ => 0.0,
272                         },
273                         _ => 0.0,
274                     };
275                     self.on_delete_stream(transaction_id, &stream_id).await?;
276                 }
277             }
278             "play" => {
279                 self.unpacketizer.session_type = config::SERVER_PULL;
280                 self.on_play(transaction_id, stream_id, others).await?;
281             }
282             "publish" => {
283                 self.unpacketizer.session_type = config::SERVER_PUSH;
284                 self.on_publish(transaction_id, stream_id, others).await?;
285             }
286             _ => {}
287         }
288 
289         Ok(())
290     }
291 
292     fn on_set_chunk_size(&mut self, chunk_size: usize) -> Result<(), SessionError> {
293         self.unpacketizer.update_max_chunk_size(chunk_size);
294         Ok(())
295     }
296 
297     async fn on_connect(
298         &mut self,
299         transaction_id: &f64,
300         command_obj: &HashMap<String, Amf0ValueType>,
301     ) -> Result<(), SessionError> {
302         log::info!(
303             "[ S<-C ] connect and the transaction id: {}",
304             transaction_id
305         );
306 
307         self.connect_command_object = Some(command_obj.clone());
308         let mut control_message =
309             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
310         control_message
311             .write_window_acknowledgement_size(define::WINDOW_ACKNOWLEDGEMENT_SIZE)
312             .await?;
313         control_message
314             .write_set_peer_bandwidth(
315                 define::PEER_BANDWIDTH,
316                 define::peer_bandwidth_limit_type::DYNAMIC,
317             )
318             .await?;
319         //control_message.write_set_chunk_size(CHUNK_SIZE).await?;
320 
321         let obj_encoding = command_obj.get("objectEncoding");
322         let encoding = match obj_encoding {
323             Some(Amf0ValueType::Number(encoding)) => encoding,
324             _ => &define::OBJENCODING_AMF0,
325         };
326 
327         let app_name = command_obj.get("app");
328         self.app_name = match app_name {
329             Some(Amf0ValueType::UTF8String(app)) => app.clone(),
330             _ => {
331                 return Err(SessionError {
332                     value: SessionErrorValue::NoAppName,
333                 });
334             }
335         };
336 
337         let mut netconnection = NetConnection::new(BytesWriter::new());
338         let data = netconnection.connect_response(
339             &transaction_id,
340             &define::FMSVER.to_string(),
341             &define::CAPABILITIES,
342             &String::from("NetConnection.Connect.Success"),
343             &define::LEVEL.to_string(),
344             &String::from("Connection Succeeded."),
345             encoding,
346         )?;
347 
348         let mut chunk_info = ChunkInfo::new(
349             csid_type::COMMAND_AMF0_AMF3,
350             chunk_type::TYPE_0,
351             0,
352             data.len() as u32,
353             msg_type_id::COMMAND_AMF0,
354             0,
355             data,
356         );
357 
358         self.packetizer.write_chunk(&mut chunk_info).await?;
359 
360         Ok(())
361     }
362 
363     pub async fn on_create_stream(&mut self, transaction_id: &f64) -> Result<(), SessionError> {
364         log::info!(
365             "[ S<-C ] create stream and the transaction id: {}",
366             transaction_id
367         );
368 
369         let mut netconnection = NetConnection::new(BytesWriter::new());
370         let data = netconnection.create_stream_response(transaction_id, &define::STREAM_ID)?;
371 
372         let mut chunk_info = ChunkInfo::new(
373             csid_type::COMMAND_AMF0_AMF3,
374             chunk_type::TYPE_0,
375             0,
376             data.len() as u32,
377             msg_type_id::COMMAND_AMF0,
378             0,
379             data,
380         );
381 
382         self.packetizer.write_chunk(&mut chunk_info).await?;
383 
384         Ok(())
385     }
386 
387     pub async fn on_delete_stream(
388         &mut self,
389         transaction_id: &f64,
390         stream_id: &f64,
391     ) -> Result<(), SessionError> {
392         log::info!(
393             "[ S<-C ] delete stream and the transaction id :{}, the stream id : {}",
394             transaction_id,
395             stream_id
396         );
397 
398         self.common
399             .unpublish_to_channels(self.app_name.clone(), self.stream_name.clone())
400             .await?;
401 
402         let mut netstream = NetStreamWriter::new(Arc::clone(&self.io));
403         netstream
404             .on_status(
405                 transaction_id,
406                 &"status".to_string(),
407                 &"NetStream.DeleteStream.Suceess".to_string(),
408                 &"".to_string(),
409             )
410             .await?;
411 
412         //self.unsubscribe_from_channels().await?;
413 
414         Ok(())
415     }
416     pub async fn on_play(
417         &mut self,
418         transaction_id: &f64,
419         stream_id: &u32,
420         other_values: &mut Vec<Amf0ValueType>,
421     ) -> Result<(), SessionError> {
422         let length = other_values.len() as u8;
423         let mut index: u8 = 0;
424 
425         let mut stream_name: Option<String> = None;
426         let mut start: Option<f64> = None;
427         let mut duration: Option<f64> = None;
428         let mut reset: Option<bool> = None;
429 
430         loop {
431             if index >= length {
432                 break;
433             }
434             index = index + 1;
435             stream_name = match other_values.remove(0) {
436                 Amf0ValueType::UTF8String(val) => Some(val),
437                 _ => None,
438             };
439 
440             if index >= length {
441                 break;
442             }
443             index = index + 1;
444             start = match other_values.remove(0) {
445                 Amf0ValueType::Number(val) => Some(val),
446                 _ => None,
447             };
448 
449             if index >= length {
450                 break;
451             }
452             index = index + 1;
453             duration = match other_values.remove(0) {
454                 Amf0ValueType::Number(val) => Some(val),
455                 _ => None,
456             };
457 
458             if index >= length {
459                 break;
460             }
461             //index = index + 1;
462             reset = match other_values.remove(0) {
463                 Amf0ValueType::Boolean(val) => Some(val),
464                 _ => None,
465             };
466             break;
467         }
468         log::info!(
469             "on_play, start: {}, duration: {}, reset: {}",
470             start.is_some(),
471             duration.is_some(),
472             reset.is_some()
473         );
474 
475         let mut event_messages = EventMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
476         event_messages.write_stream_begin(stream_id.clone()).await?;
477 
478         let mut netstream = NetStreamWriter::new(Arc::clone(&self.io));
479         netstream
480             .on_status(
481                 transaction_id,
482                 &"status".to_string(),
483                 &"NetStream.Play.Reset".to_string(),
484                 &"reset".to_string(),
485             )
486             .await?;
487 
488         netstream
489             .on_status(
490                 transaction_id,
491                 &"status".to_string(),
492                 &"NetStream.Play.Start".to_string(),
493                 &"play start".to_string(),
494             )
495             .await?;
496 
497         netstream
498             .on_status(
499                 transaction_id,
500                 &"status".to_string(),
501                 &"NetStream.Data.Start".to_string(),
502                 &"data start.".to_string(),
503             )
504             .await?;
505 
506         netstream
507             .on_status(
508                 transaction_id,
509                 &"status".to_string(),
510                 &"NetStream.Play.PublishNotify".to_string(),
511                 &"play publish notify.".to_string(),
512             )
513             .await?;
514 
515         event_messages
516             .write_stream_is_record(stream_id.clone())
517             .await?;
518 
519         self.stream_name = stream_name.clone().unwrap();
520         self.common
521             .subscribe_from_channels(self.app_name.clone(), stream_name.unwrap(), self.session_id)
522             .await?;
523 
524         self.state = ServerSessionState::Play;
525 
526         Ok(())
527     }
528 
529     pub async fn on_publish(
530         &mut self,
531         transaction_id: &f64,
532         stream_id: &u32,
533         other_values: &mut Vec<Amf0ValueType>,
534     ) -> Result<(), SessionError> {
535         let length = other_values.len();
536 
537         if length < 2 {
538             return Err(SessionError {
539                 value: SessionErrorValue::Amf0ValueCountNotCorrect,
540             });
541         }
542 
543         let stream_name = match other_values.remove(0) {
544             Amf0ValueType::UTF8String(val) => val,
545             _ => {
546                 return Err(SessionError {
547                     value: SessionErrorValue::Amf0ValueCountNotCorrect,
548                 });
549             }
550         };
551 
552         self.stream_name = stream_name;
553 
554         let _ = match other_values.remove(0) {
555             Amf0ValueType::UTF8String(val) => val,
556             _ => {
557                 return Err(SessionError {
558                     value: SessionErrorValue::Amf0ValueCountNotCorrect,
559                 });
560             }
561         };
562 
563         let mut event_messages = EventMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
564         event_messages.write_stream_begin(stream_id.clone()).await?;
565 
566         let mut netstream = NetStreamWriter::new(Arc::clone(&self.io));
567         netstream
568             .on_status(
569                 transaction_id,
570                 &"status".to_string(),
571                 &"NetStream.Publish.Start".to_string(),
572                 &"".to_string(),
573             )
574             .await?;
575 
576         self.common
577             .publish_to_channels(self.app_name.clone(), self.stream_name.clone())
578             .await?;
579 
580         Ok(())
581     }
582 }
583