1 use super::define;
2 use super::errors::SessionError;
3 use super::errors::SessionErrorValue;
4 use crate::chunk::packetizer::ChunkPacketizer;
5 use crate::chunk::{unpacketizer::ChunkUnpacketizer, ChunkInfo};
6 use crate::handshake::handshake::SimpleHandshakeServer;
7 use crate::{amf0::Amf0ValueType, chunk::unpacketizer::UnpackResult};
8 use crate::{
9     chunk::define::CHUNK_SIZE,
10     chunk::define::{chunk_type, csid_type},
11 };
12 
13 use crate::messages::define::msg_type_id;
14 use crate::messages::define::MessageTypes;
15 
16 use crate::messages::parser::MessageParser;
17 use bytes::BytesMut;
18 
19 use netio::bytes_writer::AsyncBytesWriter;
20 use netio::bytes_writer::BytesWriter;
21 use netio::netio::NetworkIO;
22 use std::time::Duration;
23 
24 use crate::netconnection::commands::NetConnection;
25 use crate::netstream::commands::NetStream;
26 use crate::protocol_control_messages::control_messages::ControlMessages;
27 
28 use crate::user_control_messages::event_messages::EventMessages;
29 
30 use std::collections::HashMap;
31 
32 use tokio::prelude::*;
33 
34 // use std::cell::{RefCell, RefMut};
35 // use std::rc::Rc;
36 
37 use std::sync::Arc;
38 use tokio::sync::Mutex;
39 
40 enum ServerSessionState {
41     Handshake,
42     ReadChunk,
43     // OnConnect,
44     // OnCreateStream,
45     // OnPlay,
46     // OnPublish,
47 }
48 
49 pub struct ServerSession<S>
50 where
51     S: AsyncRead + AsyncWrite + Unpin + Send + Sync,
52 {
53     packetizer: ChunkPacketizer<S>,
54     unpacketizer: ChunkUnpacketizer,
55     handshaker: SimpleHandshakeServer<S>,
56 
57     io: Arc<Mutex<NetworkIO<S>>>,
58     state: ServerSessionState,
59 }
60 
61 impl<S> ServerSession<S>
62 where
63     S: AsyncRead + AsyncWrite + Unpin + Send + Sync,
64 {
65     pub fn new(stream: S, timeout: Duration) -> Self {
66         let net_io = Arc::new(Mutex::new(NetworkIO::new(stream, timeout)));
67         let bytes_writer = AsyncBytesWriter::new(Arc::clone(&net_io));
68 
69         Self {
70             packetizer: ChunkPacketizer::new(bytes_writer),
71             unpacketizer: ChunkUnpacketizer::new(),
72             handshaker: SimpleHandshakeServer::new(Arc::clone(&net_io)),
73 
74             io: Arc::clone(&net_io),
75             state: ServerSessionState::Handshake,
76         }
77     }
78 
79     pub async fn run(&mut self) -> Result<(), SessionError> {
80         let duration = Duration::new(10, 10);
81 
82         loop {
83             let data = self.io.lock().await.read().await?;
84 
85             match self.state {
86                 ServerSessionState::Handshake => {
87                     self.handshaker.extend_data(&data[..]);
88                     let result = self.handshaker.handshake().await;
89 
90                     match result {
91                         Ok(v) => {
92                             self.state = ServerSessionState::ReadChunk;
93                         }
94                         Err(e) => {}
95                     }
96                 }
97                 ServerSessionState::ReadChunk => {
98                     self.unpacketizer.extend_data(&data[..]);
99                     let result = self.unpacketizer.read_chunk()?;
100 
101                     match result {
102                         UnpackResult::ChunkInfo(chunk_info) => {
103                             let msg_stream_id = chunk_info.message_header.msg_streamd_id;
104                             let mut message_parser = MessageParser::new(chunk_info);
105                             let mut msg = message_parser.parse()?;
106 
107                             self.process_messages(&mut msg, &msg_stream_id).await?;
108                         }
109                         _ => {}
110                     }
111                 }
112             }
113         }
114 
115         //Ok(())
116     }
117     pub fn send_set_chunk_size(&mut self) -> Result<(), SessionError> {
118         let mut controlmessage = ControlMessages::new(AsyncBytesWriter::new(self.io.clone()));
119         controlmessage.write_set_chunk_size(CHUNK_SIZE)?;
120 
121         Ok(())
122     }
123     pub async fn send_audio(&mut self, data: BytesMut) -> Result<(), SessionError> {
124         let mut chunk_info = ChunkInfo::new(
125             csid_type::AUDIO,
126             chunk_type::TYPE_0,
127             0,
128             data.len() as u32,
129             msg_type_id::AUDIO,
130             0,
131             data,
132         );
133 
134         self.packetizer.write_chunk(&mut chunk_info).await?;
135 
136         Ok(())
137     }
138 
139     pub async fn send_video(&mut self, data: BytesMut) -> Result<(), SessionError> {
140         let mut chunk_info = ChunkInfo::new(
141             csid_type::VIDEO,
142             chunk_type::TYPE_0,
143             0,
144             data.len() as u32,
145             msg_type_id::VIDEO,
146             0,
147             data,
148         );
149 
150         self.packetizer.write_chunk(&mut chunk_info).await?;
151 
152         Ok(())
153     }
154     pub async fn process_messages(
155         &mut self,
156         rtmp_msg: &mut MessageTypes,
157         msg_stream_id: &u32,
158     ) -> Result<(), SessionError> {
159         match rtmp_msg {
160             MessageTypes::Amf0Command {
161                 command_name,
162                 transaction_id,
163                 command_object,
164                 others,
165             } => {
166                 self.process_amf0_command_message(
167                     msg_stream_id,
168                     command_name,
169                     transaction_id,
170                     command_object,
171                     others,
172                 )
173                 .await?
174             }
175 
176             _ => {}
177         }
178         Ok(())
179     }
180 
181     pub async fn process_amf0_command_message(
182         &mut self,
183         stream_id: &u32,
184         command_name: &Amf0ValueType,
185         transaction_id: &Amf0ValueType,
186         command_object: &Amf0ValueType,
187         others: &mut Vec<Amf0ValueType>,
188     ) -> Result<(), SessionError> {
189         let empty_cmd_name = &String::new();
190         let cmd_name = match command_name {
191             Amf0ValueType::UTF8String(str) => str,
192             _ => empty_cmd_name,
193         };
194 
195         let transaction_id = match transaction_id {
196             Amf0ValueType::Number(number) => number,
197             _ => &0.0,
198         };
199 
200         let empty_cmd_obj: HashMap<String, Amf0ValueType> = HashMap::new();
201         let obj = match command_object {
202             Amf0ValueType::Object(obj) => obj,
203             _ => &empty_cmd_obj,
204         };
205 
206         match cmd_name.as_str() {
207             "connect" => {
208                 self.on_connect(&transaction_id, &obj).await?;
209             }
210             "createStream" => {
211                 self.on_create_stream(transaction_id)?;
212             }
213             "deleteStream" => {
214                 if others.len() > 1 {
215                     let stream_id = match others.pop() {
216                         Some(val) => match val {
217                             Amf0ValueType::Number(streamid) => streamid,
218                             _ => 0.0,
219                         },
220                         _ => 0.0,
221                     };
222 
223                     self.on_delete_stream(transaction_id, &stream_id).await?;
224                 }
225             }
226             "play" => {
227                 self.on_play(transaction_id, stream_id, others).await?;
228             }
229             "publish" => {
230                 self.on_publish(transaction_id, stream_id, others).await?;
231             }
232             _ => {}
233         }
234 
235         Ok(())
236     }
237 
238     async fn on_connect(
239         &mut self,
240         transaction_id: &f64,
241         command_obj: &HashMap<String, Amf0ValueType>,
242     ) -> Result<(), SessionError> {
243         let mut control_message = ControlMessages::new(AsyncBytesWriter::new(self.io.clone()));
244         control_message.write_window_acknowledgement_size(define::WINDOW_ACKNOWLEDGEMENT_SIZE)?;
245         control_message.write_set_peer_bandwidth(
246             define::PEER_BANDWIDTH,
247             define::PeerBandWidthLimitType::DYNAMIC,
248         )?;
249         control_message.write_set_chunk_size(CHUNK_SIZE)?;
250 
251         let obj_encoding = command_obj.get("objectEncoding");
252         let encoding = match obj_encoding {
253             Some(Amf0ValueType::Number(encoding)) => encoding,
254             _ => &define::OBJENCODING_AMF0,
255         };
256 
257         let mut netconnection = NetConnection::new(BytesWriter::new());
258         let data = netconnection.connect_response(
259             &transaction_id,
260             &define::FMSVER.to_string(),
261             &define::CAPABILITIES,
262             &String::from("NetConnection.Connect.Success"),
263             &define::LEVEL.to_string(),
264             &String::from("Connection Succeeded."),
265             encoding,
266         )?;
267 
268         let mut chunk_info = ChunkInfo::new(
269             csid_type::COMMAND_AMF0_AMF3,
270             chunk_type::TYPE_0,
271             0,
272             data.len() as u32,
273             msg_type_id::COMMAND_AMF0,
274             0,
275             data,
276         );
277 
278         self.packetizer.write_chunk(&mut chunk_info).await?;
279 
280         Ok(())
281     }
282 
283     pub fn on_create_stream(&mut self, transaction_id: &f64) -> Result<(), SessionError> {
284         let mut netconnection = NetConnection::new(BytesWriter::new());
285         netconnection.create_stream_response(transaction_id, &define::STREAM_ID)?;
286 
287         Ok(())
288     }
289 
290     pub async fn on_delete_stream(
291         &mut self,
292         transaction_id: &f64,
293         stream_id: &f64,
294     ) -> Result<(), SessionError> {
295         let mut netstream = NetStream::new(BytesWriter::new());
296         let data = netstream.on_status(
297             transaction_id,
298             &"status".to_string(),
299             &"NetStream.DeleteStream.Suceess".to_string(),
300             &"".to_string(),
301         )?;
302 
303         let mut chunk_info = ChunkInfo::new(
304             csid_type::COMMAND_AMF0_AMF3,
305             chunk_type::TYPE_0,
306             0,
307             data.len() as u32,
308             msg_type_id::COMMAND_AMF0,
309             0,
310             data,
311         );
312 
313         self.packetizer.write_chunk(&mut chunk_info).await?;
314         Ok(())
315     }
316     pub async fn on_play(
317         &mut self,
318         transaction_id: &f64,
319         stream_id: &u32,
320         other_values: &mut Vec<Amf0ValueType>,
321     ) -> Result<(), SessionError> {
322         let length = other_values.len() as u8;
323         let mut index: u8 = 0;
324 
325         let mut stream_name: Option<String> = None;
326         let mut start: Option<f64> = None;
327         let mut duration: Option<f64> = None;
328         let mut reset: Option<bool> = None;
329 
330         loop {
331             if index >= length {
332                 break;
333             }
334             index = index + 1;
335             stream_name = match other_values.remove(0) {
336                 Amf0ValueType::UTF8String(val) => Some(val),
337                 _ => None,
338             };
339 
340             if index >= length {
341                 break;
342             }
343             index = index + 1;
344             start = match other_values.remove(0) {
345                 Amf0ValueType::Number(val) => Some(val),
346                 _ => None,
347             };
348 
349             if index >= length {
350                 break;
351             }
352             index = index + 1;
353             duration = match other_values.remove(0) {
354                 Amf0ValueType::Number(val) => Some(val),
355                 _ => None,
356             };
357 
358             if index >= length {
359                 break;
360             }
361             index = index + 1;
362             reset = match other_values.remove(0) {
363                 Amf0ValueType::Boolean(val) => Some(val),
364                 _ => None,
365             };
366             break;
367         }
368 
369         let mut event_messages = EventMessages::new(AsyncBytesWriter::new(self.io.clone()));
370         event_messages.stream_begin(stream_id.clone()).await?;
371 
372         let mut netstream = NetStream::new(BytesWriter::new());
373         match reset {
374             Some(val) => {
375                 if val {
376                     netstream.on_status(
377                         transaction_id,
378                         &"status".to_string(),
379                         &"NetStream.Play.Reset".to_string(),
380                         &"".to_string(),
381                     )?;
382                 }
383             }
384             _ => {}
385         }
386 
387         netstream.on_status(
388             transaction_id,
389             &"status".to_string(),
390             &"NetStream.Play.Start".to_string(),
391             &"".to_string(),
392         )?;
393 
394         event_messages.stream_is_record(stream_id.clone()).await?;
395 
396         Ok(())
397     }
398 
399     pub async fn on_publish(
400         &mut self,
401         transaction_id: &f64,
402         stream_id: &u32,
403         other_values: &mut Vec<Amf0ValueType>,
404     ) -> Result<(), SessionError> {
405         let length = other_values.len();
406 
407         if length < 2 {
408             return Err(SessionError {
409                 value: SessionErrorValue::Amf0ValueCountNotCorrect,
410             });
411         }
412 
413         let stream_name = match other_values.remove(0) {
414             Amf0ValueType::UTF8String(val) => val,
415             _ => {
416                 return Err(SessionError {
417                     value: SessionErrorValue::Amf0ValueCountNotCorrect,
418                 });
419             }
420         };
421 
422         let stream_type = match other_values.remove(0) {
423             Amf0ValueType::UTF8String(val) => val,
424             _ => {
425                 return Err(SessionError {
426                     value: SessionErrorValue::Amf0ValueCountNotCorrect,
427                 });
428             }
429         };
430 
431         let mut event_messages = EventMessages::new(AsyncBytesWriter::new(self.io.clone()));
432         event_messages.stream_begin(stream_id.clone()).await?;
433 
434         let mut netstream = NetStream::new(BytesWriter::new());
435         let data = netstream.on_status(
436             transaction_id,
437             &"status".to_string(),
438             &"NetStream.Publish.Start".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 
454         Ok(())
455     }
456 }
457