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::writer::NetStreamWriter;
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 =
216             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
217         controlmessage.write_set_chunk_size(CHUNK_SIZE).await?;
218 
219         Ok(())
220     }
221     pub async fn send_audio(&mut self, data: BytesMut, timestamp: u32) -> Result<(), SessionError> {
222         let mut chunk_info = ChunkInfo::new(
223             csid_type::AUDIO,
224             chunk_type::TYPE_0,
225             timestamp,
226             data.len() as u32,
227             msg_type_id::AUDIO,
228             0,
229             data,
230         );
231 
232         self.packetizer.write_chunk(&mut chunk_info).await?;
233 
234         Ok(())
235     }
236 
237     pub async fn send_video(&mut self, data: BytesMut, timestamp: u32) -> Result<(), SessionError> {
238         let mut chunk_info = ChunkInfo::new(
239             csid_type::VIDEO,
240             chunk_type::TYPE_0,
241             timestamp,
242             data.len() as u32,
243             msg_type_id::VIDEO,
244             0,
245             data,
246         );
247 
248         self.packetizer.write_chunk(&mut chunk_info).await?;
249 
250         Ok(())
251     }
252     pub async fn process_messages(
253         &mut self,
254         rtmp_msg: &mut RtmpMessageData,
255         msg_stream_id: &u32,
256         timestamp: &u32,
257     ) -> Result<(), SessionError> {
258         match rtmp_msg {
259             RtmpMessageData::Amf0Command {
260                 command_name,
261                 transaction_id,
262                 command_object,
263                 others,
264             } => {
265                 self.process_amf0_command_message(
266                     msg_stream_id,
267                     command_name,
268                     transaction_id,
269                     command_object,
270                     others,
271                 )
272                 .await?
273             }
274             RtmpMessageData::SetChunkSize { chunk_size } => {
275                 self.on_set_chunk_size(chunk_size.clone() as usize)?;
276             }
277             RtmpMessageData::AudioData { data } => {
278                 self.on_audio_data(data, timestamp)?;
279             }
280             RtmpMessageData::VideoData { data } => {
281                 self.on_video_data(data, timestamp)?;
282             }
283 
284             _ => {}
285         }
286         Ok(())
287     }
288 
289     pub async fn process_amf0_command_message(
290         &mut self,
291         stream_id: &u32,
292         command_name: &Amf0ValueType,
293         transaction_id: &Amf0ValueType,
294         command_object: &Amf0ValueType,
295         others: &mut Vec<Amf0ValueType>,
296     ) -> Result<(), SessionError> {
297         let empty_cmd_name = &String::new();
298         let cmd_name = match command_name {
299             Amf0ValueType::UTF8String(str) => str,
300             _ => empty_cmd_name,
301         };
302 
303         let transaction_id = match transaction_id {
304             Amf0ValueType::Number(number) => number,
305             _ => &0.0,
306         };
307 
308         let empty_cmd_obj: HashMap<String, Amf0ValueType> = HashMap::new();
309         let obj = match command_object {
310             Amf0ValueType::Object(obj) => obj,
311             _ => &empty_cmd_obj,
312         };
313 
314         match cmd_name.as_str() {
315             "connect" => {
316                 print!("connect .......");
317                 self.on_connect(&transaction_id, &obj).await?;
318             }
319             "createStream" => {
320                 self.on_create_stream(transaction_id).await?;
321             }
322             "deleteStream" => {
323                 if others.len() > 1 {
324                     let stream_id = match others.pop() {
325                         Some(val) => match val {
326                             Amf0ValueType::Number(streamid) => streamid,
327                             _ => 0.0,
328                         },
329                         _ => 0.0,
330                     };
331 
332                     self.on_delete_stream(transaction_id, &stream_id).await?;
333                 }
334             }
335             "play" => {
336                 self.on_play(transaction_id, stream_id, others).await?;
337             }
338             "publish" => {
339                 self.on_publish(transaction_id, stream_id, others).await?;
340             }
341             _ => {}
342         }
343 
344         Ok(())
345     }
346 
347     fn on_set_chunk_size(&mut self, chunk_size: usize) -> Result<(), SessionError> {
348         self.unpacketizer.update_max_chunk_size(chunk_size);
349         Ok(())
350     }
351 
352     async fn on_connect(
353         &mut self,
354         transaction_id: &f64,
355         command_obj: &HashMap<String, Amf0ValueType>,
356     ) -> Result<(), SessionError> {
357         let mut control_message =
358             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
359         control_message
360             .write_window_acknowledgement_size(define::WINDOW_ACKNOWLEDGEMENT_SIZE)
361             .await?;
362         control_message
363             .write_set_peer_bandwidth(
364                 define::PEER_BANDWIDTH,
365                 define::PeerBandWidthLimitType::DYNAMIC,
366             )
367             .await?;
368         control_message.write_set_chunk_size(CHUNK_SIZE).await?;
369 
370         let obj_encoding = command_obj.get("objectEncoding");
371         let encoding = match obj_encoding {
372             Some(Amf0ValueType::Number(encoding)) => encoding,
373             _ => &define::OBJENCODING_AMF0,
374         };
375 
376         let app_name = command_obj.get("app");
377         self.app_name = match app_name {
378             Some(Amf0ValueType::UTF8String(app)) => app.clone(),
379             _ => {
380                 return Err(SessionError {
381                     value: SessionErrorValue::NoAppName,
382                 });
383             }
384         };
385 
386         let mut netconnection = NetConnection::new(BytesWriter::new());
387         let data = netconnection.connect_response(
388             &transaction_id,
389             &define::FMSVER.to_string(),
390             &define::CAPABILITIES,
391             &String::from("NetConnection.Connect.Success"),
392             &define::LEVEL.to_string(),
393             &String::from("Connection Succeeded."),
394             encoding,
395         )?;
396 
397         let mut chunk_info = ChunkInfo::new(
398             csid_type::COMMAND_AMF0_AMF3,
399             chunk_type::TYPE_0,
400             0,
401             data.len() as u32,
402             msg_type_id::COMMAND_AMF0,
403             0,
404             data,
405         );
406 
407         self.packetizer.write_chunk(&mut chunk_info).await?;
408 
409         Ok(())
410     }
411 
412     pub async fn on_create_stream(&mut self, transaction_id: &f64) -> Result<(), SessionError> {
413         let mut netconnection = NetConnection::new(BytesWriter::new());
414         let data = netconnection.create_stream_response(transaction_id, &define::STREAM_ID)?;
415 
416         let mut chunk_info = ChunkInfo::new(
417             csid_type::COMMAND_AMF0_AMF3,
418             chunk_type::TYPE_0,
419             0,
420             data.len() as u32,
421             msg_type_id::COMMAND_AMF0,
422             0,
423             data,
424         );
425 
426         self.packetizer.write_chunk(&mut chunk_info).await?;
427 
428         Ok(())
429     }
430 
431     pub async fn on_delete_stream(
432         &mut self,
433         transaction_id: &f64,
434         stream_id: &f64,
435     ) -> Result<(), SessionError> {
436         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
437         netstream
438             .on_status(
439                 transaction_id,
440                 &"status".to_string(),
441                 &"NetStream.DeleteStream.Suceess".to_string(),
442                 &"".to_string(),
443             )
444             .await?;
445 
446         Ok(())
447     }
448     pub async fn on_play(
449         &mut self,
450         transaction_id: &f64,
451         stream_id: &u32,
452         other_values: &mut Vec<Amf0ValueType>,
453     ) -> Result<(), SessionError> {
454         let length = other_values.len() as u8;
455         let mut index: u8 = 0;
456 
457         let mut stream_name: Option<String> = None;
458         let mut start: Option<f64> = None;
459         let mut duration: Option<f64> = None;
460         let mut reset: Option<bool> = None;
461 
462         loop {
463             if index >= length {
464                 break;
465             }
466             index = index + 1;
467             stream_name = match other_values.remove(0) {
468                 Amf0ValueType::UTF8String(val) => Some(val),
469                 _ => None,
470             };
471 
472             if index >= length {
473                 break;
474             }
475             index = index + 1;
476             start = match other_values.remove(0) {
477                 Amf0ValueType::Number(val) => Some(val),
478                 _ => None,
479             };
480 
481             if index >= length {
482                 break;
483             }
484             index = index + 1;
485             duration = match other_values.remove(0) {
486                 Amf0ValueType::Number(val) => Some(val),
487                 _ => None,
488             };
489 
490             if index >= length {
491                 break;
492             }
493             index = index + 1;
494             reset = match other_values.remove(0) {
495                 Amf0ValueType::Boolean(val) => Some(val),
496                 _ => None,
497             };
498             break;
499         }
500 
501         let mut event_messages = EventMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
502         event_messages.write_stream_begin(stream_id.clone()).await?;
503 
504         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
505         match reset {
506             Some(val) => {
507                 if val {
508                     netstream
509                         .on_status(
510                             transaction_id,
511                             &"status".to_string(),
512                             &"NetStream.Play.Reset".to_string(),
513                             &"".to_string(),
514                         )
515                         .await?;
516                 }
517             }
518             _ => {}
519         }
520 
521         netstream
522             .on_status(
523                 transaction_id,
524                 &"status".to_string(),
525                 &"NetStream.Play.Start".to_string(),
526                 &"".to_string(),
527             )
528             .await?;
529 
530         event_messages
531             .write_stream_is_record(stream_id.clone())
532             .await?;
533 
534         self.subscribe_from_channels(stream_name.unwrap()).await?;
535         self.state = ServerSessionState::Play;
536 
537         Ok(())
538     }
539 
540     async fn subscribe_from_channels(&mut self, stream_name: String) -> Result<(), SessionError> {
541         let (sender, receiver) = oneshot::channel();
542         let subscribe_event = ChannelEvent::Subscribe {
543             app_name: self.app_name.clone(),
544             stream_name,
545             responder: sender,
546         };
547 
548         let rv = self.event_producer.send(subscribe_event);
549         match rv {
550             Err(_) => {
551                 return Err(SessionError {
552                     value: SessionErrorValue::ChannelEventSendErr,
553                 })
554             }
555             _ => {}
556         }
557 
558         match receiver.await {
559             Ok(consumer) => {
560                 self.data_consumer = consumer;
561             }
562             Err(_) => {}
563         }
564         Ok(())
565     }
566 
567     pub async fn on_publish(
568         &mut self,
569         transaction_id: &f64,
570         stream_id: &u32,
571         other_values: &mut Vec<Amf0ValueType>,
572     ) -> Result<(), SessionError> {
573         let length = other_values.len();
574 
575         if length < 2 {
576             return Err(SessionError {
577                 value: SessionErrorValue::Amf0ValueCountNotCorrect,
578             });
579         }
580 
581         let stream_name = match other_values.remove(0) {
582             Amf0ValueType::UTF8String(val) => val,
583             _ => {
584                 return Err(SessionError {
585                     value: SessionErrorValue::Amf0ValueCountNotCorrect,
586                 });
587             }
588         };
589 
590         let stream_type = match other_values.remove(0) {
591             Amf0ValueType::UTF8String(val) => val,
592             _ => {
593                 return Err(SessionError {
594                     value: SessionErrorValue::Amf0ValueCountNotCorrect,
595                 });
596             }
597         };
598 
599         let mut event_messages = EventMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
600         event_messages.write_stream_begin(stream_id.clone()).await?;
601 
602         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
603         netstream
604             .on_status(
605                 transaction_id,
606                 &"status".to_string(),
607                 &"NetStream.Publish.Start".to_string(),
608                 &"".to_string(),
609             )
610             .await?;
611 
612         print!("before publish_to_channels\n");
613         self.publish_to_channels(stream_name).await?;
614         print!("after publish_to_channels\n");
615 
616         Ok(())
617     }
618 
619     async fn publish_to_channels(&mut self, stream_name: String) -> Result<(), SessionError> {
620         let (sender, receiver) = oneshot::channel();
621         let publish_event = ChannelEvent::Publish {
622             app_name: self.app_name.clone(),
623             stream_name,
624             responder: sender,
625         };
626 
627         let rv = self.event_producer.send(publish_event);
628         match rv {
629             Err(_) => {
630                 return Err(SessionError {
631                     value: SessionErrorValue::ChannelEventSendErr,
632                 })
633             }
634             _ => {}
635         }
636 
637         match receiver.await {
638             Ok(producer) => {
639                 print!("set producer before\n");
640                 self.data_producer = producer;
641                 print!("set producer after\n");
642             }
643             Err(_) => {}
644         }
645         Ok(())
646     }
647 
648     pub fn on_video_data(
649         &mut self,
650         data: &mut BytesMut,
651         timestamp: &u32,
652     ) -> Result<(), SessionError> {
653         let data = ChannelData::Video {
654             timestamp: timestamp.clone(),
655             data: data.clone(),
656         };
657 
658         print!("receive video data\n");
659         match self.data_producer.send(data) {
660             Ok(size) => {}
661             Err(err) => {
662                 format!("receive video err {}", err);
663                 return Err(SessionError {
664                     value: SessionErrorValue::SendChannelDataErr,
665                 });
666             }
667         }
668 
669         Ok(())
670     }
671 
672     pub fn on_audio_data(
673         &mut self,
674         data: &mut BytesMut,
675         timestamp: &u32,
676     ) -> Result<(), SessionError> {
677         // let data = ChannelData::Audio {
678         //     timestamp: timestamp.clone(),
679         //     data: data.clone(),
680         // };
681 
682         // match self.data_producer.send(data) {
683         //     Ok(size) => {}
684         //     Err(_) => {
685         //         return Err(SessionError {
686         //             value: SessionErrorValue::SendChannelDataErr,
687         //         })
688         //     }
689         // }
690 
691         Ok(())
692     }
693 }
694