1 use super::errors::EventMessagesError;
2 
3 use super::define;
4 use crate::messages::define::msg_type_id;
5 use byteorder::BigEndian;
6 use netio::bytes_writer::AsyncBytesWriter;
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(&mut self, stream_id: u32) -> Result<(), EventMessagesError> {
81         self.write_control_message_header(6)?;
82         self.writer
83             .write_u16::<BigEndian>(define::RTMP_EVENT_STREAM_IS_RECORD)?;
84         self.writer.write_u32::<BigEndian>(stream_id)?;
85 
86         self.writer.flush().await?;
87 
88         Ok(())
89     }
90 
91     pub async fn write_ping_request(&mut self, timestamp: u32) -> Result<(), EventMessagesError> {
92         self.write_control_message_header(6)?;
93         self.writer
94             .write_u16::<BigEndian>(define::RTMP_EVENT_PING)?;
95         self.writer.write_u32::<BigEndian>(timestamp)?;
96 
97         self.writer.flush().await?;
98 
99         Ok(())
100     }
101 
102     pub async fn write_ping_response(&mut self, timestamp: u32) -> Result<(), EventMessagesError> {
103         self.write_control_message_header(6)?;
104         self.writer
105             .write_u16::<BigEndian>(define::RTMP_EVENT_PONG)?;
106         self.writer.write_u32::<BigEndian>(timestamp)?;
107 
108         self.writer.flush().await?;
109 
110         Ok(())
111     }
112 }
113