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