1 use {
2     super::{
3         define,
4         errors::{SessionError, SessionErrorValue},
5     },
6     crate::{
7         amf0::Amf0ValueType,
8         channels::define::{
9             ChannelData, ChannelDataConsumer, ChannelDataPublisher, ChannelEvent,
10             ChannelEventPublisher,
11         },
12         chunk::{
13             define::{chunk_type, csid_type, CHUNK_SIZE},
14             packetizer::ChunkPacketizer,
15             unpacketizer::{ChunkUnpacketizer, UnpackResult},
16             ChunkInfo,
17         },
18         config,
19         handshake::handshake::{ServerHandshakeState, SimpleHandshakeServer},
20         messages::{
21             define::{msg_type_id, RtmpMessageData},
22             parser::MessageParser,
23         },
24         netconnection::commands::NetConnection,
25         netstream::writer::NetStreamWriter,
26         protocol_control_messages::writer::ProtocolControlMessagesWriter,
27         user_control_messages::writer::EventMessagesWriter,
28     },
29     bytes::BytesMut,
30     netio::{
31         bytes_writer::{AsyncBytesWriter, BytesWriter},
32         netio::NetworkIO,
33     },
34     std::{collections::HashMap, sync::Arc},
35     tokio::{
36         net::TcpStream,
37         sync::{mpsc, oneshot, Mutex},
38     },
39 };
40 
41 enum ServerSessionState {
42     Handshake,
43     ReadChunk,
44     // OnConnect,
45     // OnCreateStream,
46     //Publish,
47     Play,
48 }
49 
50 pub struct ServerSession {
51     app_name: String,
52     stream_name: String,
53 
54     io: Arc<Mutex<NetworkIO>>,
55     simple_handshaker: SimpleHandshakeServer,
56     //complex_handshaker: ComplexHandshakeServer,
57     packetizer: ChunkPacketizer,
58     unpacketizer: ChunkUnpacketizer,
59 
60     state: ServerSessionState,
61 
62     event_producer: ChannelEventPublisher,
63 
64     //send video, audio or metadata from publish server session to player server sessions
65     data_producer: ChannelDataPublisher,
66     //receive video, audio or metadata from publish server session and send out to player
67     data_consumer: ChannelDataConsumer,
68 
69     netio_data: BytesMut,
70     need_process: bool,
71 
72     pub session_id: u64,
73     pub session_type: u8,
74 }
75 
76 impl ServerSession {
77     pub fn new(stream: TcpStream, event_producer: ChannelEventPublisher, session_id: u64) -> Self {
78         let net_io = Arc::new(Mutex::new(NetworkIO::new(stream)));
79         //only used for init,since I don't found a better way to deal with this.
80         let (init_producer, init_consumer) = mpsc::unbounded_channel();
81 
82         Self {
83             app_name: String::from(""),
84             stream_name: String::from(""),
85 
86             io: Arc::clone(&net_io),
87             simple_handshaker: SimpleHandshakeServer::new(Arc::clone(&net_io)),
88             //complex_handshaker: ComplexHandshakeServer::new(Arc::clone(&net_io)),
89             packetizer: ChunkPacketizer::new(Arc::clone(&net_io)),
90             unpacketizer: ChunkUnpacketizer::new(),
91 
92             state: ServerSessionState::Handshake,
93 
94             event_producer,
95             data_producer: init_producer,
96             data_consumer: init_consumer,
97             session_id: session_id,
98             netio_data: BytesMut::new(),
99             need_process: false,
100             session_type: 0,
101         }
102     }
103 
104     pub async fn run(&mut self) -> Result<(), SessionError> {
105         loop {
106             match self.state {
107                 ServerSessionState::Handshake => {
108                     self.handshake().await?;
109                 }
110                 ServerSessionState::ReadChunk => {
111                     self.read_parse_chunks().await?;
112                 }
113                 ServerSessionState::Play => {
114                     self.play().await?;
115                 }
116             }
117         }
118 
119         //Ok(())
120     }
121 
122     async fn handshake(&mut self) -> Result<(), SessionError> {
123         self.netio_data = self.io.lock().await.read().await?;
124         self.simple_handshaker.extend_data(&self.netio_data[..]);
125         self.simple_handshaker.handshake().await?;
126 
127         match self.simple_handshaker.state {
128             ServerHandshakeState::Finish => {
129                 self.state = ServerSessionState::ReadChunk;
130 
131                 let left_bytes = self.simple_handshaker.get_remaining_bytes();
132                 if left_bytes.len() > 0 {
133                     self.unpacketizer.extend_data(&left_bytes[..]);
134                     self.need_process = true;
135                 }
136 
137                 return Ok(());
138             }
139             _ => {}
140         }
141 
142         Ok(())
143     }
144 
145     async fn read_parse_chunks(&mut self) -> Result<(), SessionError> {
146         if !self.need_process {
147             self.netio_data = self.io.lock().await.read().await?;
148             self.unpacketizer.extend_data(&self.netio_data[..]);
149         }
150 
151         self.need_process = false;
152 
153         loop {
154             let result = self.unpacketizer.read_chunks();
155 
156             if let Ok(rv) = result {
157                 match rv {
158                     UnpackResult::Chunks(chunks) => {
159                         for chunk_info in chunks.iter() {
160                             let mut msg = MessageParser::new(chunk_info.clone(), self.session_type)
161                                 .parse()?;
162 
163                             let msg_stream_id = chunk_info.message_header.msg_streamd_id;
164                             let timestamp = chunk_info.message_header.timestamp;
165                             self.process_messages(&mut msg, &msg_stream_id, &timestamp)
166                                 .await?;
167                         }
168                     }
169                     _ => {}
170                 }
171             } else {
172                 break;
173             }
174         }
175         Ok(())
176     }
177 
178     async fn play(&mut self) -> Result<(), SessionError> {
179         match self.send_media_data().await {
180             Ok(_) => {}
181             Err(err) => {
182                 // let len = self.unpacketizer.reader.get_remaining_bytes().len();
183                 // print!("send meidi data err len:{}\n", len);
184 
185                 // utils::print::print(self.unpacketizer.reader.get_remaining_bytes());
186 
187                 // if len > 0 {
188                 //     self.need_process = true;
189                 // }
190 
191                 // self.state = ServerSessionState::ReadChunk;
192 
193                 self.unsubscribe_from_channels().await?;
194 
195                 return Err(err);
196             }
197         }
198 
199         Ok(())
200     }
201 
202     async fn send_media_data(&mut self) -> Result<(), SessionError> {
203         loop {
204             if let Some(data) = self.data_consumer.recv().await {
205                 match data {
206                     ChannelData::Audio { timestamp, data } => {
207                         //print!("send audio data\n");
208                         self.send_audio(data, timestamp).await?;
209                     }
210                     ChannelData::Video { timestamp, data } => {
211                         //print!("send video data\n");
212                         self.send_video(data, timestamp).await?;
213                     }
214                     ChannelData::MetaData { body } => {
215                         print!("send meta data\n");
216                         self.send_metadata(body).await?;
217                     }
218                 }
219             }
220         }
221     }
222 
223     pub async fn send_set_chunk_size(&mut self) -> Result<(), SessionError> {
224         let mut controlmessage =
225             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
226         controlmessage.write_set_chunk_size(CHUNK_SIZE).await?;
227 
228         Ok(())
229     }
230     pub async fn send_audio(&mut self, data: BytesMut, timestamp: u32) -> Result<(), SessionError> {
231         let mut chunk_info = ChunkInfo::new(
232             csid_type::AUDIO,
233             chunk_type::TYPE_0,
234             timestamp,
235             data.len() as u32,
236             msg_type_id::AUDIO,
237             0,
238             data,
239         );
240 
241         self.packetizer.write_chunk(&mut chunk_info).await?;
242 
243         Ok(())
244     }
245 
246     pub async fn send_video(&mut self, data: BytesMut, timestamp: u32) -> Result<(), SessionError> {
247         let mut chunk_info = ChunkInfo::new(
248             csid_type::VIDEO,
249             chunk_type::TYPE_0,
250             timestamp,
251             data.len() as u32,
252             msg_type_id::VIDEO,
253             0,
254             data,
255         );
256 
257         self.packetizer.write_chunk(&mut chunk_info).await?;
258 
259         Ok(())
260     }
261 
262     async fn send_metadata(&mut self, data: BytesMut) -> Result<(), SessionError> {
263         let mut chunk_info = ChunkInfo::new(
264             csid_type::DATA_AMF0_AMF3,
265             chunk_type::TYPE_0,
266             0,
267             data.len() as u32,
268             msg_type_id::DATA_AMF0,
269             0,
270             data,
271         );
272 
273         self.packetizer.write_chunk(&mut chunk_info).await?;
274         Ok(())
275     }
276     pub async fn process_messages(
277         &mut self,
278         rtmp_msg: &mut RtmpMessageData,
279         msg_stream_id: &u32,
280         timestamp: &u32,
281     ) -> Result<(), SessionError> {
282         match rtmp_msg {
283             RtmpMessageData::Amf0Command {
284                 command_name,
285                 transaction_id,
286                 command_object,
287                 others,
288             } => {
289                 self.on_amf0_command_message(
290                     msg_stream_id,
291                     command_name,
292                     transaction_id,
293                     command_object,
294                     others,
295                 )
296                 .await?
297             }
298             RtmpMessageData::SetChunkSize { chunk_size } => {
299                 self.on_set_chunk_size(chunk_size.clone() as usize)?;
300             }
301             RtmpMessageData::AudioData { data } => {
302                 self.on_audio_data(data, timestamp)?;
303             }
304             RtmpMessageData::VideoData { data } => {
305                 self.on_video_data(data, timestamp)?;
306             }
307             RtmpMessageData::AmfData { raw_data } => {
308                 self.on_amf_data(raw_data)?;
309             }
310 
311             _ => {}
312         }
313         Ok(())
314     }
315 
316     pub async fn on_amf0_command_message(
317         &mut self,
318         stream_id: &u32,
319         command_name: &Amf0ValueType,
320         transaction_id: &Amf0ValueType,
321         command_object: &Amf0ValueType,
322         others: &mut Vec<Amf0ValueType>,
323     ) -> Result<(), SessionError> {
324         let empty_cmd_name = &String::new();
325         let cmd_name = match command_name {
326             Amf0ValueType::UTF8String(str) => str,
327             _ => empty_cmd_name,
328         };
329 
330         let transaction_id = match transaction_id {
331             Amf0ValueType::Number(number) => number,
332             _ => &0.0,
333         };
334 
335         let empty_cmd_obj: HashMap<String, Amf0ValueType> = HashMap::new();
336         let obj = match command_object {
337             Amf0ValueType::Object(obj) => obj,
338             _ => &empty_cmd_obj,
339         };
340 
341         match cmd_name.as_str() {
342             "connect" => {
343                 print!("connect .......");
344                 self.on_connect(&transaction_id, &obj).await?;
345             }
346             "createStream" => {
347                 self.on_create_stream(transaction_id).await?;
348             }
349             "deleteStream" => {
350                 print!("deletestream....\n");
351                 if others.len() > 0 {
352                     let stream_id = match others.pop() {
353                         Some(val) => match val {
354                             Amf0ValueType::Number(streamid) => streamid,
355                             _ => 0.0,
356                         },
357                         _ => 0.0,
358                     };
359 
360                     print!("deletestream....{}\n", stream_id);
361 
362                     self.on_delete_stream(transaction_id, &stream_id).await?;
363                 }
364             }
365             "play" => {
366                 self.session_type = config::SERVER_PULL;
367                 self.unpacketizer.session_type = config::SERVER_PULL;
368                 self.on_play(transaction_id, stream_id, others).await?;
369             }
370             "publish" => {
371                 self.session_type = config::SERVER_PUSH;
372                 self.unpacketizer.session_type = config::SERVER_PUSH;
373                 self.on_publish(transaction_id, stream_id, others).await?;
374             }
375             _ => {}
376         }
377 
378         Ok(())
379     }
380 
381     fn on_set_chunk_size(&mut self, chunk_size: usize) -> Result<(), SessionError> {
382         self.unpacketizer.update_max_chunk_size(chunk_size);
383         Ok(())
384     }
385 
386     async fn on_connect(
387         &mut self,
388         transaction_id: &f64,
389         command_obj: &HashMap<String, Amf0ValueType>,
390     ) -> Result<(), SessionError> {
391         let mut control_message =
392             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
393         control_message
394             .write_window_acknowledgement_size(define::WINDOW_ACKNOWLEDGEMENT_SIZE)
395             .await?;
396         control_message
397             .write_set_peer_bandwidth(
398                 define::PEER_BANDWIDTH,
399                 define::peer_bandwidth_limit_type::DYNAMIC,
400             )
401             .await?;
402         control_message.write_set_chunk_size(CHUNK_SIZE).await?;
403 
404         let obj_encoding = command_obj.get("objectEncoding");
405         let encoding = match obj_encoding {
406             Some(Amf0ValueType::Number(encoding)) => encoding,
407             _ => &define::OBJENCODING_AMF0,
408         };
409 
410         let app_name = command_obj.get("app");
411         self.app_name = match app_name {
412             Some(Amf0ValueType::UTF8String(app)) => app.clone(),
413             _ => {
414                 return Err(SessionError {
415                     value: SessionErrorValue::NoAppName,
416                 });
417             }
418         };
419 
420         let mut netconnection = NetConnection::new(BytesWriter::new());
421         let data = netconnection.connect_response(
422             &transaction_id,
423             &define::FMSVER.to_string(),
424             &define::CAPABILITIES,
425             &String::from("NetConnection.Connect.Success"),
426             &define::LEVEL.to_string(),
427             &String::from("Connection Succeeded."),
428             encoding,
429         )?;
430 
431         let mut chunk_info = ChunkInfo::new(
432             csid_type::COMMAND_AMF0_AMF3,
433             chunk_type::TYPE_0,
434             0,
435             data.len() as u32,
436             msg_type_id::COMMAND_AMF0,
437             0,
438             data,
439         );
440 
441         self.packetizer.write_chunk(&mut chunk_info).await?;
442 
443         Ok(())
444     }
445 
446     pub async fn on_create_stream(&mut self, transaction_id: &f64) -> Result<(), SessionError> {
447         let mut netconnection = NetConnection::new(BytesWriter::new());
448         let data = netconnection.create_stream_response(transaction_id, &define::STREAM_ID)?;
449 
450         let mut chunk_info = ChunkInfo::new(
451             csid_type::COMMAND_AMF0_AMF3,
452             chunk_type::TYPE_0,
453             0,
454             data.len() as u32,
455             msg_type_id::COMMAND_AMF0,
456             0,
457             data,
458         );
459 
460         self.packetizer.write_chunk(&mut chunk_info).await?;
461 
462         Ok(())
463     }
464 
465     pub async fn on_delete_stream(
466         &mut self,
467         transaction_id: &f64,
468         stream_id: &f64,
469     ) -> Result<(), SessionError> {
470         self.unpublish_to_channels().await?;
471 
472         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
473         netstream
474             .on_status(
475                 transaction_id,
476                 &"status".to_string(),
477                 &"NetStream.DeleteStream.Suceess".to_string(),
478                 &"".to_string(),
479             )
480             .await?;
481 
482         print!("stream id{}", stream_id);
483 
484         //self.unsubscribe_from_channels().await?;
485 
486         Ok(())
487     }
488     pub async fn on_play(
489         &mut self,
490         transaction_id: &f64,
491         stream_id: &u32,
492         other_values: &mut Vec<Amf0ValueType>,
493     ) -> Result<(), SessionError> {
494         let length = other_values.len() as u8;
495         let mut index: u8 = 0;
496 
497         let mut stream_name: Option<String> = None;
498         let mut start: Option<f64> = None;
499         let mut duration: Option<f64> = None;
500         let mut reset: Option<bool> = None;
501 
502         loop {
503             if index >= length {
504                 break;
505             }
506             index = index + 1;
507             stream_name = match other_values.remove(0) {
508                 Amf0ValueType::UTF8String(val) => Some(val),
509                 _ => None,
510             };
511 
512             if index >= length {
513                 break;
514             }
515             index = index + 1;
516             start = match other_values.remove(0) {
517                 Amf0ValueType::Number(val) => Some(val),
518                 _ => None,
519             };
520 
521             if index >= length {
522                 break;
523             }
524             index = index + 1;
525             duration = match other_values.remove(0) {
526                 Amf0ValueType::Number(val) => Some(val),
527                 _ => None,
528             };
529 
530             if index >= length {
531                 break;
532             }
533             //index = index + 1;
534             reset = match other_values.remove(0) {
535                 Amf0ValueType::Boolean(val) => Some(val),
536                 _ => None,
537             };
538             break;
539         }
540         print!("start {}", start.is_some());
541         print!("druation {}", duration.is_some());
542         print!("reset {}", reset.is_some());
543 
544         let mut event_messages = EventMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
545         event_messages.write_stream_begin(stream_id.clone()).await?;
546 
547         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
548         netstream
549             .on_status(
550                 transaction_id,
551                 &"status".to_string(),
552                 &"NetStream.Play.Reset".to_string(),
553                 &"reset".to_string(),
554             )
555             .await?;
556 
557         netstream
558             .on_status(
559                 transaction_id,
560                 &"status".to_string(),
561                 &"NetStream.Play.Start".to_string(),
562                 &"play start".to_string(),
563             )
564             .await?;
565 
566         netstream
567             .on_status(
568                 transaction_id,
569                 &"status".to_string(),
570                 &"NetStream.Data.Start".to_string(),
571                 &"data start.".to_string(),
572             )
573             .await?;
574 
575         netstream
576             .on_status(
577                 transaction_id,
578                 &"status".to_string(),
579                 &"NetStream.Play.PublishNotify".to_string(),
580                 &"play publish notify.".to_string(),
581             )
582             .await?;
583 
584         event_messages
585             .write_stream_is_record(stream_id.clone())
586             .await?;
587 
588         self.subscribe_from_channels(stream_name.unwrap()).await?;
589         self.state = ServerSessionState::Play;
590 
591         Ok(())
592     }
593 
594     async fn unsubscribe_from_channels(&mut self) -> Result<(), SessionError> {
595         let subscribe_event = ChannelEvent::UnSubscribe {
596             app_name: self.app_name.clone(),
597             stream_name: self.stream_name.clone(),
598             session_id: self.session_id,
599         };
600 
601         let _ = self.event_producer.send(subscribe_event);
602 
603         Ok(())
604     }
605 
606     async fn subscribe_from_channels(&mut self, stream_name: String) -> Result<(), SessionError> {
607         self.stream_name = stream_name.clone();
608 
609         print!(
610             "subscribe info............{} {} {}\n",
611             self.app_name,
612             stream_name.clone(),
613             self.session_id
614         );
615 
616         let (sender, receiver) = oneshot::channel();
617         let subscribe_event = ChannelEvent::Subscribe {
618             app_name: self.app_name.clone(),
619             stream_name,
620             session_id: self.session_id,
621             responder: sender,
622         };
623 
624         let rv = self.event_producer.send(subscribe_event);
625         match rv {
626             Err(_) => {
627                 return Err(SessionError {
628                     value: SessionErrorValue::ChannelEventSendErr,
629                 })
630             }
631             _ => {}
632         }
633 
634         match receiver.await {
635             Ok(consumer) => {
636                 self.data_consumer = consumer;
637             }
638             Err(_) => {}
639         }
640         Ok(())
641     }
642 
643     pub async fn on_publish(
644         &mut self,
645         transaction_id: &f64,
646         stream_id: &u32,
647         other_values: &mut Vec<Amf0ValueType>,
648     ) -> Result<(), SessionError> {
649         let length = other_values.len();
650 
651         if length < 2 {
652             return Err(SessionError {
653                 value: SessionErrorValue::Amf0ValueCountNotCorrect,
654             });
655         }
656 
657         let stream_name = match other_values.remove(0) {
658             Amf0ValueType::UTF8String(val) => val,
659             _ => {
660                 return Err(SessionError {
661                     value: SessionErrorValue::Amf0ValueCountNotCorrect,
662                 });
663             }
664         };
665 
666         self.stream_name = stream_name;
667 
668         let _ = match other_values.remove(0) {
669             Amf0ValueType::UTF8String(val) => val,
670             _ => {
671                 return Err(SessionError {
672                     value: SessionErrorValue::Amf0ValueCountNotCorrect,
673                 });
674             }
675         };
676 
677         let mut event_messages = EventMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
678         event_messages.write_stream_begin(stream_id.clone()).await?;
679 
680         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
681         netstream
682             .on_status(
683                 transaction_id,
684                 &"status".to_string(),
685                 &"NetStream.Publish.Start".to_string(),
686                 &"".to_string(),
687             )
688             .await?;
689 
690         print!("before publish_to_channels\n");
691         self.publish_to_channels().await?;
692         print!("after publish_to_channels\n");
693 
694         Ok(())
695     }
696 
697     async fn unpublish_to_channels(&mut self) -> Result<(), SessionError> {
698         let unpublish_event = ChannelEvent::UnPublish {
699             app_name: self.app_name.clone(),
700             stream_name: self.stream_name.clone(),
701         };
702 
703         let rv = self.event_producer.send(unpublish_event);
704         match rv {
705             Err(_) => {
706                 println!("unpublish_to_channels error.");
707                 return Err(SessionError {
708                     value: SessionErrorValue::ChannelEventSendErr,
709                 });
710             }
711             _ => {
712                 println!("unpublish_to_channels successfully.")
713             }
714         }
715         Ok(())
716     }
717 
718     async fn publish_to_channels(&mut self) -> Result<(), SessionError> {
719         let (sender, receiver) = oneshot::channel();
720         let publish_event = ChannelEvent::Publish {
721             app_name: self.app_name.clone(),
722             stream_name: self.stream_name.clone(),
723             responder: sender,
724         };
725 
726         let rv = self.event_producer.send(publish_event);
727         match rv {
728             Err(_) => {
729                 return Err(SessionError {
730                     value: SessionErrorValue::ChannelEventSendErr,
731                 })
732             }
733             _ => {}
734         }
735 
736         match receiver.await {
737             Ok(producer) => {
738                 print!("set producer before\n");
739                 self.data_producer = producer;
740                 print!("set producer after\n");
741             }
742             Err(err) => {
743                 print!("publish_to_channels err{}\n", err)
744             }
745         }
746         Ok(())
747     }
748 
749     pub fn on_video_data(
750         &mut self,
751         data: &mut BytesMut,
752         timestamp: &u32,
753     ) -> Result<(), SessionError> {
754         let data = ChannelData::Video {
755             timestamp: timestamp.clone(),
756             data: data.clone(),
757         };
758 
759         //print!("receive video data\n");
760         match self.data_producer.send(data) {
761             Ok(_) => {}
762             Err(err) => {
763                 print!("send video err {}\n", err);
764                 return Err(SessionError {
765                     value: SessionErrorValue::SendChannelDataErr,
766                 });
767             }
768         }
769 
770         Ok(())
771     }
772 
773     pub fn on_audio_data(
774         &mut self,
775         data: &mut BytesMut,
776         timestamp: &u32,
777     ) -> Result<(), SessionError> {
778         let data = ChannelData::Audio {
779             timestamp: timestamp.clone(),
780             data: data.clone(),
781         };
782 
783         match self.data_producer.send(data) {
784             Ok(_) => {}
785             Err(err) => {
786                 print!("receive audio err {}\n", err);
787                 return Err(SessionError {
788                     value: SessionErrorValue::SendChannelDataErr,
789                 });
790             }
791         }
792 
793         Ok(())
794     }
795 
796     pub fn on_amf_data(&mut self, body: &mut BytesMut) -> Result<(), SessionError> {
797         let data = ChannelData::MetaData { body: body.clone() };
798 
799         match self.data_producer.send(data) {
800             Ok(_) => {}
801             Err(_) => {
802                 return Err(SessionError {
803                     value: SessionErrorValue::SendChannelDataErr,
804                 })
805             }
806         }
807 
808         Ok(())
809     }
810 }
811