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