1 use {
2     super::{
3         common::Common,
4         define,
5         errors::{SessionError, SessionErrorValue},
6     },
7     crate::utils::print::print,
8     crate::{
9         amf0::Amf0ValueType,
10         channels::define::ChannelEventProducer,
11         chunk::{
12             define::{chunk_type, csid_type, CHUNK_SIZE},
13             packetizer::ChunkPacketizer,
14             unpacketizer::{ChunkUnpacketizer, UnpackResult},
15             ChunkInfo,
16         },
17         handshake::handshake::{ClientHandshakeState, SimpleHandshakeClient},
18         messages::{
19             define::{msg_type_id, RtmpMessageData},
20             parser::MessageParser,
21         },
22         netconnection::commands::{ConnectProperties, NetConnection},
23         netstream::writer::NetStreamWriter,
24         protocol_control_messages::writer::ProtocolControlMessagesWriter,
25         user_control_messages::writer::EventMessagesWriter,
26     },
27     bytes::BytesMut,
28     networkio::{
29         bytes_writer::{AsyncBytesWriter, BytesWriter},
30         networkio::NetworkIO,
31     },
32     std::{collections::HashMap, sync::Arc},
33     tokio::{net::TcpStream, sync::Mutex},
34 };
35 
36 #[allow(dead_code)]
37 enum ClientSessionState {
38     Handshake,
39     Connect,
40     CreateStream,
41     Play,
42     PublishingContent,
43     StartPublish,
44     WaitStateChange,
45 }
46 
47 #[allow(dead_code)]
48 enum ClientSessionPlayState {
49     Handshake,
50     Connect,
51     CreateStream,
52     Play,
53 }
54 
55 #[allow(dead_code)]
56 enum ClientSessionPublishState {
57     Handshake,
58     Connect,
59     CreateStream,
60     PublishingContent,
61 }
62 #[allow(dead_code)]
63 pub enum ClientType {
64     Play,
65     Publish,
66 }
67 pub struct ClientSession {
68     io: Arc<Mutex<NetworkIO>>,
69     common: Common,
70 
71     handshaker: SimpleHandshakeClient,
72 
73     packetizer: ChunkPacketizer,
74     unpacketizer: ChunkUnpacketizer,
75 
76     app_name: String,
77     stream_name: String,
78     session_type: u8,
79     session_id: u64,
80 
81     state: ClientSessionState,
82     client_type: ClientType,
83     connect_command_object: Option<HashMap<String, Amf0ValueType>>,
84 }
85 
86 impl ClientSession {
87     #[allow(dead_code)]
88     pub fn new(
89         stream: TcpStream,
90         client_type: ClientType,
91         app_name: String,
92         stream_name: String,
93         event_producer: ChannelEventProducer,
94         session_id: u64,
95     ) -> Self {
96         let net_io = Arc::new(Mutex::new(NetworkIO::new(stream)));
97 
98         Self {
99             io: Arc::clone(&net_io),
100             common: Common::new(Arc::clone(&net_io), event_producer),
101 
102             handshaker: SimpleHandshakeClient::new(Arc::clone(&net_io)),
103 
104             packetizer: ChunkPacketizer::new(Arc::clone(&net_io)),
105             unpacketizer: ChunkUnpacketizer::new(),
106 
107             app_name: app_name,
108             stream_name: stream_name,
109             client_type: client_type,
110 
111             state: ClientSessionState::Handshake,
112             session_type: 0,
113             session_id: session_id,
114             connect_command_object: None,
115         }
116     }
117 
118     pub fn set_connect_command_object(&mut self, object: HashMap<String, Amf0ValueType>) {
119         self.connect_command_object = Some(object);
120     }
121 
122     pub async fn run(&mut self) -> Result<(), SessionError> {
123         loop {
124             match self.state {
125                 ClientSessionState::Handshake => {
126                     println!("handshake");
127                     self.handshake().await?;
128                     continue;
129                 }
130                 ClientSessionState::Connect => {
131                     println!("connect");
132                     self.send_connect(&(define::TRANSACTION_ID_CONNECT as f64))
133                         .await?;
134                     self.state = ClientSessionState::WaitStateChange;
135                 }
136                 ClientSessionState::CreateStream => {
137                     println!("CreateStream");
138                     self.send_create_stream(&(define::TRANSACTION_ID_CREATE_STREAM as f64))
139                         .await?;
140                     self.state = ClientSessionState::WaitStateChange;
141                 }
142                 ClientSessionState::Play => {
143                     self.send_play(&0.0, &self.stream_name.clone(), &0.0, &0.0, &false)
144                         .await?;
145                     self.state = ClientSessionState::WaitStateChange;
146                 }
147                 ClientSessionState::PublishingContent => {
148                     self.send_publish(&0.0, &self.stream_name.clone(), &"live".to_string())
149                         .await?;
150                     self.state = ClientSessionState::WaitStateChange;
151                 }
152                 ClientSessionState::StartPublish => {
153                     self.common.send_channel_data().await?;
154                 }
155                 ClientSessionState::WaitStateChange => {}
156             }
157 
158             let data = self.io.lock().await.read().await?;
159             self.unpacketizer.extend_data(&data[..]);
160             let result = self.unpacketizer.read_chunk()?;
161 
162             match result {
163                 UnpackResult::ChunkInfo(chunk_info) => {
164                     let mut message_parser = MessageParser::new(chunk_info, self.session_type);
165                     let mut msg = message_parser.parse()?;
166 
167                     self.process_messages(&mut msg).await?;
168                 }
169                 _ => {}
170             }
171         }
172 
173         // Ok(())
174     }
175 
176     async fn handshake(&mut self) -> Result<(), SessionError> {
177         loop {
178             self.handshaker.handshake().await?;
179             if self.handshaker.state == ClientHandshakeState::Finish {
180                 println!("handshake finish");
181                 break;
182             }
183 
184             let data = self.io.lock().await.read().await?;
185             print(data.clone());
186             self.handshaker.extend_data(&data[..]);
187         }
188 
189         self.state = ClientSessionState::Connect;
190 
191         Ok(())
192     }
193 
194     pub async fn process_messages(
195         &mut self,
196         msg: &mut RtmpMessageData,
197     ) -> Result<(), SessionError> {
198         match msg {
199             RtmpMessageData::Amf0Command {
200                 command_name,
201                 transaction_id,
202                 command_object,
203                 others,
204             } => {
205                 self.process_amf0_command_message(
206                     command_name,
207                     transaction_id,
208                     command_object,
209                     others,
210                 )
211                 .await?
212             }
213             RtmpMessageData::SetPeerBandwidth { properties } => {
214                 print!("{}", properties.window_size);
215                 self.on_set_peer_bandwidth().await?
216             }
217             RtmpMessageData::SetChunkSize { chunk_size } => self.on_set_chunk_size(chunk_size)?,
218             RtmpMessageData::AudioData { data } => {
219                 let _ = data.len();
220             }
221             RtmpMessageData::VideoData { data } => {
222                 let _ = data.len();
223             }
224 
225             _ => {}
226         }
227         Ok(())
228     }
229 
230     pub async fn process_amf0_command_message(
231         &mut self,
232         command_name: &Amf0ValueType,
233         transaction_id: &Amf0ValueType,
234         command_object: &Amf0ValueType,
235         others: &mut Vec<Amf0ValueType>,
236     ) -> Result<(), SessionError> {
237         let empty_cmd_name = &String::new();
238         let cmd_name = match command_name {
239             Amf0ValueType::UTF8String(str) => str,
240             _ => empty_cmd_name,
241         };
242 
243         let transaction_id = match transaction_id {
244             Amf0ValueType::Number(number) => number.clone() as u8,
245             _ => 0,
246         };
247 
248         let empty_cmd_obj: HashMap<String, Amf0ValueType> = HashMap::new();
249         let _ = match command_object {
250             Amf0ValueType::Object(obj) => obj,
251             // Amf0ValueType::Null =>
252             _ => &empty_cmd_obj,
253         };
254 
255         match cmd_name.as_str() {
256             "_reslut" => match transaction_id {
257                 define::TRANSACTION_ID_CONNECT => {
258                     self.on_result_connect().await?;
259                 }
260                 define::TRANSACTION_ID_CREATE_STREAM => {
261                     self.on_result_create_stream()?;
262                 }
263                 _ => {}
264             },
265             "_error" => {
266                 self.on_error()?;
267             }
268             "onStatus" => {
269                 match others.remove(0) {
270                     Amf0ValueType::Object(obj) => self.on_status(&obj).await?,
271                     _ => {
272                         return Err(SessionError {
273                             value: SessionErrorValue::Amf0ValueCountNotCorrect,
274                         })
275                     }
276                 };
277             }
278 
279             _ => {}
280         }
281 
282         Ok(())
283     }
284 
285     pub async fn send_connect(&mut self, transaction_id: &f64) -> Result<(), SessionError> {
286         self.send_set_chunk_size().await?;
287 
288         // let properties = ConnectProperties::new(self.app_name.clone());
289         let mut netconnection = NetConnection::new(BytesWriter::new());
290         let data = netconnection
291             .connect_with_value(transaction_id, self.connect_command_object.clone().unwrap())?;
292 
293         let mut chunk_info = ChunkInfo::new(
294             csid_type::COMMAND_AMF0_AMF3,
295             chunk_type::TYPE_0,
296             0,
297             data.len() as u32,
298             msg_type_id::COMMAND_AMF0,
299             0,
300             data,
301         );
302 
303         self.packetizer.write_chunk(&mut chunk_info).await?;
304         Ok(())
305     }
306 
307     pub async fn send_create_stream(&mut self, transaction_id: &f64) -> Result<(), SessionError> {
308         let mut netconnection = NetConnection::new(BytesWriter::new());
309         let data = netconnection.create_stream(transaction_id)?;
310 
311         let mut chunk_info = ChunkInfo::new(
312             csid_type::COMMAND_AMF0_AMF3,
313             chunk_type::TYPE_0,
314             0,
315             data.len() as u32,
316             msg_type_id::COMMAND_AMF0,
317             0,
318             data,
319         );
320 
321         self.packetizer.write_chunk(&mut chunk_info).await?;
322 
323         Ok(())
324     }
325 
326     pub async fn send_delete_stream(
327         &mut self,
328         transaction_id: &f64,
329         stream_id: &f64,
330     ) -> Result<(), SessionError> {
331         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
332         netstream.delete_stream(transaction_id, stream_id).await?;
333 
334         Ok(())
335     }
336 
337     pub async fn send_publish(
338         &mut self,
339         transaction_id: &f64,
340         stream_name: &String,
341         stream_type: &String,
342     ) -> Result<(), SessionError> {
343         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
344         netstream
345             .publish(transaction_id, stream_name, stream_type)
346             .await?;
347 
348         Ok(())
349     }
350 
351     pub async fn send_play(
352         &mut self,
353         transaction_id: &f64,
354         stream_name: &String,
355         start: &f64,
356         duration: &f64,
357         reset: &bool,
358     ) -> Result<(), SessionError> {
359         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
360         netstream
361             .play(transaction_id, stream_name, start, duration, reset)
362             .await?;
363 
364         Ok(())
365     }
366 
367     pub async fn send_set_chunk_size(&mut self) -> Result<(), SessionError> {
368         let mut controlmessage =
369             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
370         controlmessage.write_set_chunk_size(CHUNK_SIZE).await?;
371         Ok(())
372     }
373 
374     pub async fn send_window_acknowledgement_size(
375         &mut self,
376         window_size: u32,
377     ) -> Result<(), SessionError> {
378         let mut controlmessage =
379             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
380         controlmessage
381             .write_window_acknowledgement_size(window_size)
382             .await?;
383         Ok(())
384     }
385 
386     pub async fn send_set_buffer_length(
387         &mut self,
388         stream_id: u32,
389         ms: u32,
390     ) -> Result<(), SessionError> {
391         let mut eventmessages = EventMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
392         eventmessages.write_set_buffer_length(stream_id, ms).await?;
393 
394         Ok(())
395     }
396 
397     // pub async fn send_audio(&mut self, data: BytesMut) -> Result<(), SessionError> {
398     //     let mut chunk_info = ChunkInfo::new(
399     //         csid_type::AUDIO,
400     //         chunk_type::TYPE_0,
401     //         0,
402     //         data.len() as u32,
403     //         msg_type_id::AUDIO,
404     //         0,
405     //         data,
406     //     );
407 
408     //     self.packetizer.write_chunk(&mut chunk_info).await?;
409 
410     //     Ok(())
411     // }
412 
413     // pub async fn send_video(&mut self, data: BytesMut) -> Result<(), SessionError> {
414     //     let mut chunk_info = ChunkInfo::new(
415     //         csid_type::VIDEO,
416     //         chunk_type::TYPE_0,
417     //         0,
418     //         data.len() as u32,
419     //         msg_type_id::VIDEO,
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_result_connect(&mut self) -> Result<(), SessionError> {
430         let mut controlmessage =
431             ProtocolControlMessagesWriter::new(AsyncBytesWriter::new(self.io.clone()));
432         controlmessage.write_acknowledgement(3107).await?;
433 
434         let mut netstream = NetStreamWriter::new(BytesWriter::new(), Arc::clone(&self.io));
435         netstream
436             .release_stream(&(define::TRANSACTION_ID_CONNECT as f64), &self.stream_name)
437             .await?;
438         netstream
439             .fcpublish(&(define::TRANSACTION_ID_CONNECT as f64), &self.stream_name)
440             .await?;
441 
442         self.state = ClientSessionState::CreateStream;
443 
444         Ok(())
445     }
446 
447     pub fn on_result_create_stream(&mut self) -> Result<(), SessionError> {
448         match self.client_type {
449             ClientType::Play => {
450                 self.state = ClientSessionState::Play;
451             }
452             ClientType::Publish => {
453                 self.state = ClientSessionState::PublishingContent;
454             }
455         }
456         Ok(())
457     }
458 
459     pub fn on_set_chunk_size(&mut self, chunk_size: &mut u32) -> Result<(), SessionError> {
460         self.unpacketizer
461             .update_max_chunk_size(chunk_size.clone() as usize);
462         Ok(())
463     }
464 
465     pub async fn on_set_peer_bandwidth(&mut self) -> Result<(), SessionError> {
466         self.send_window_acknowledgement_size(250000).await?;
467         Ok(())
468     }
469 
470     pub fn on_error(&mut self) -> Result<(), SessionError> {
471         Ok(())
472     }
473 
474     pub async fn on_status(
475         &mut self,
476         obj: &HashMap<String, Amf0ValueType>,
477     ) -> Result<(), SessionError> {
478         if let Some(Amf0ValueType::UTF8String(code_info)) = obj.get("code") {
479             match &code_info[..] == "NetStream.Publish.Start" {
480                 true => {
481                     self.state = ClientSessionState::StartPublish;
482                     self.common
483                         .subscribe_from_channels(
484                             self.app_name.clone(),
485                             self.stream_name.clone(),
486                             self.session_id,
487                         )
488                         .await?;
489                 }
490                 false => {}
491             }
492         }
493         println!("{}", obj.len());
494         Ok(())
495     }
496 }
497