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