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