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