1 use {
2     super::{define, errors::EventMessagesError},
3     byteorder::BigEndian,
4     networkio::bytes_writer::AsyncBytesWriter,
5     crate::messages::define::msg_type_id,
6 };
7 
8 pub struct EventMessagesWriter {
9     writer: AsyncBytesWriter,
10     // amf0_writer: Amf0Writer,
11 }
12 
13 impl EventMessagesWriter {
14     pub fn new(writer: AsyncBytesWriter) -> Self {
15         Self { writer: writer }
16     }
17     fn write_control_message_header(&mut self, len: u32) -> Result<(), EventMessagesError> {
18         //0 1 2 3 4 5 6 7
19         //+-+-+-+-+-+-+-+-+
20         //|fmt|  cs id  |
21         //+-+-+-+-+-+-+-+-+
22         // 0x0     0x02
23         self.writer.write_u8(0x0 << 6 | 0x02)?; //fmt 0 and csid 2
24         self.writer.write_u24::<BigEndian>(0)?; //timestamp 3 bytes and value 0
25         self.writer.write_u24::<BigEndian>(len)?; //msg length
26         self.writer.write_u8(msg_type_id::USER_CONTROL_EVENT)?; //msg type id
27         self.writer.write_u32::<BigEndian>(0)?; //msg stream ID 0
28 
29         Ok(())
30     }
31 
32     pub async fn write_stream_begin(&mut self, stream_id: u32) -> Result<(), EventMessagesError> {
33         self.write_control_message_header(6)?;
34         self.writer
35             .write_u16::<BigEndian>(define::RTMP_EVENT_STREAM_BEGIN)?;
36         self.writer.write_u32::<BigEndian>(stream_id)?;
37 
38         self.writer.flush().await?;
39         Ok(())
40     }
41 
42     pub async fn write_stream_eof(&mut self, stream_id: u32) -> Result<(), EventMessagesError> {
43         self.write_control_message_header(6)?;
44         self.writer
45             .write_u16::<BigEndian>(define::RTMP_EVENT_STREAM_EOF)?;
46         self.writer.write_u32::<BigEndian>(stream_id)?;
47 
48         self.writer.flush().await?;
49 
50         Ok(())
51     }
52 
53     pub async fn write_stream_dry(&mut self, stream_id: u32) -> Result<(), EventMessagesError> {
54         self.write_control_message_header(6)?;
55         self.writer
56             .write_u16::<BigEndian>(define::RTMP_EVENT_STREAM_DRY)?;
57         self.writer.write_u32::<BigEndian>(stream_id)?;
58 
59         self.writer.flush().await?;
60 
61         Ok(())
62     }
63 
64     pub async fn write_set_buffer_length(
65         &mut self,
66         stream_id: u32,
67         ms: u32,
68     ) -> Result<(), EventMessagesError> {
69         self.write_control_message_header(6)?;
70         self.writer
71             .write_u16::<BigEndian>(define::RTMP_EVENT_SET_BUFFER_LENGTH)?;
72         self.writer.write_u32::<BigEndian>(stream_id)?;
73         self.writer.write_u32::<BigEndian>(ms)?;
74 
75         self.writer.flush().await?;
76 
77         Ok(())
78     }
79 
80     pub async fn write_stream_is_record(
81         &mut self,
82         stream_id: u32,
83     ) -> Result<(), EventMessagesError> {
84         self.write_control_message_header(6)?;
85         self.writer
86             .write_u16::<BigEndian>(define::RTMP_EVENT_STREAM_IS_RECORDED)?;
87         self.writer.write_u32::<BigEndian>(stream_id)?;
88 
89         self.writer.flush().await?;
90 
91         Ok(())
92     }
93 
94     pub async fn write_ping_request(&mut self, timestamp: u32) -> Result<(), EventMessagesError> {
95         self.write_control_message_header(6)?;
96         self.writer
97             .write_u16::<BigEndian>(define::RTMP_EVENT_PING)?;
98         self.writer.write_u32::<BigEndian>(timestamp)?;
99 
100         self.writer.flush().await?;
101 
102         Ok(())
103     }
104 
105     pub async fn write_ping_response(&mut self, timestamp: u32) -> Result<(), EventMessagesError> {
106         self.write_control_message_header(6)?;
107         self.writer
108             .write_u16::<BigEndian>(define::RTMP_EVENT_PONG)?;
109         self.writer.write_u32::<BigEndian>(timestamp)?;
110 
111         self.writer.flush().await?;
112 
113         Ok(())
114     }
115 }
116