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