1 use { 2 super::errors::NetStreamError, 3 crate::{ 4 amf0::{amf0_writer::Amf0Writer, define::Amf0ValueType}, 5 chunk::{chunk::ChunkInfo, define as chunk_define, packetizer::ChunkPacketizer}, 6 messages::define as messages_define, 7 }, 8 bytes::BytesMut, 9 netio::{bytes_writer::BytesWriter, netio::NetworkIO}, 10 std::{collections::HashMap, sync::Arc}, 11 tokio::sync::Mutex, 12 }; 13 14 pub struct NetStreamWriter { 15 amf0_writer: Amf0Writer, 16 packetizer: ChunkPacketizer, 17 } 18 19 impl NetStreamWriter { 20 pub fn new(writer: BytesWriter, io: Arc<Mutex<NetworkIO>>) -> Self { 21 Self { 22 amf0_writer: Amf0Writer::new(writer), 23 packetizer: ChunkPacketizer::new(io), 24 } 25 } 26 pub async fn write_chunk(&mut self, data: BytesMut) -> Result<(), NetStreamError> { 27 let mut chunk_info = ChunkInfo::new( 28 chunk_define::csid_type::COMMAND_AMF0_AMF3, 29 chunk_define::chunk_type::TYPE_0, 30 0, 31 data.len() as u32, 32 messages_define::msg_type_id::COMMAND_AMF0, 33 0, 34 data, 35 ); 36 37 self.packetizer.write_chunk(&mut chunk_info).await?; 38 Ok(()) 39 } 40 pub async fn play( 41 &mut self, 42 transaction_id: &f64, 43 stream_name: &String, 44 start: &f64, 45 duration: &f64, 46 reset: &bool, 47 ) -> Result<(), NetStreamError> { 48 self.amf0_writer.write_string(&String::from("play"))?; 49 self.amf0_writer.write_number(transaction_id)?; 50 self.amf0_writer.write_null()?; 51 self.amf0_writer.write_string(stream_name)?; 52 self.amf0_writer.write_number(start)?; 53 self.amf0_writer.write_number(duration)?; 54 self.amf0_writer.write_bool(reset)?; 55 56 let data = self.amf0_writer.extract_current_bytes(); 57 self.write_chunk(data).await?; 58 59 return Ok(()); 60 } 61 pub async fn delete_stream( 62 &mut self, 63 transaction_id: &f64, 64 stream_id: &f64, 65 ) -> Result<(), NetStreamError> { 66 self.amf0_writer 67 .write_string(&String::from("deleteStream"))?; 68 self.amf0_writer.write_number(transaction_id)?; 69 self.amf0_writer.write_null()?; 70 self.amf0_writer.write_number(stream_id)?; 71 72 let data = self.amf0_writer.extract_current_bytes(); 73 self.write_chunk(data).await?; 74 75 return Ok(()); 76 } 77 78 pub async fn close_stream( 79 &mut self, 80 transaction_id: &f64, 81 stream_id: &f64, 82 ) -> Result<(), NetStreamError> { 83 self.amf0_writer 84 .write_string(&String::from("closeStream"))?; 85 self.amf0_writer.write_number(transaction_id)?; 86 self.amf0_writer.write_null()?; 87 self.amf0_writer.write_number(stream_id)?; 88 89 let data = self.amf0_writer.extract_current_bytes(); 90 self.write_chunk(data).await?; 91 92 return Ok(()); 93 } 94 95 async fn receive_audio( 96 &mut self, 97 transaction_id: &f64, 98 enable: &bool, 99 ) -> Result<(), NetStreamError> { 100 self.amf0_writer 101 .write_string(&String::from("receiveAudio"))?; 102 self.amf0_writer.write_number(transaction_id)?; 103 self.amf0_writer.write_null()?; 104 self.amf0_writer.write_bool(enable)?; 105 106 let data = self.amf0_writer.extract_current_bytes(); 107 self.write_chunk(data).await?; 108 109 return Ok(()); 110 } 111 112 async fn receive_video( 113 &mut self, 114 transaction_id: &f64, 115 enable: &bool, 116 ) -> Result<(), NetStreamError> { 117 self.amf0_writer 118 .write_string(&String::from("receiveVideo"))?; 119 self.amf0_writer.write_number(transaction_id)?; 120 self.amf0_writer.write_null()?; 121 self.amf0_writer.write_bool(enable)?; 122 123 let data = self.amf0_writer.extract_current_bytes(); 124 self.write_chunk(data).await?; 125 126 return Ok(()); 127 } 128 pub async fn publish( 129 &mut self, 130 transaction_id: &f64, 131 stream_name: &String, 132 stream_type: &String, 133 ) -> Result<(), NetStreamError> { 134 self.amf0_writer.write_string(&String::from("publish"))?; 135 self.amf0_writer.write_number(transaction_id)?; 136 self.amf0_writer.write_null()?; 137 self.amf0_writer.write_string(stream_name)?; 138 self.amf0_writer.write_string(stream_type)?; 139 140 let data = self.amf0_writer.extract_current_bytes(); 141 self.write_chunk(data).await?; 142 143 return Ok(()); 144 } 145 async fn seek(&mut self, transaction_id: &f64, ms: &f64) -> Result<(), NetStreamError> { 146 self.amf0_writer.write_string(&String::from("seek"))?; 147 self.amf0_writer.write_number(transaction_id)?; 148 self.amf0_writer.write_null()?; 149 self.amf0_writer.write_number(ms)?; 150 151 let data = self.amf0_writer.extract_current_bytes(); 152 self.write_chunk(data).await?; 153 154 return Ok(()); 155 } 156 157 async fn pause( 158 &mut self, 159 transaction_id: &f64, 160 pause: &bool, 161 ms: &f64, 162 ) -> Result<(), NetStreamError> { 163 self.amf0_writer.write_string(&String::from("pause"))?; 164 self.amf0_writer.write_number(transaction_id)?; 165 self.amf0_writer.write_null()?; 166 self.amf0_writer.write_bool(pause)?; 167 self.amf0_writer.write_number(ms)?; 168 169 let data = self.amf0_writer.extract_current_bytes(); 170 self.write_chunk(data).await?; 171 172 return Ok(()); 173 } 174 175 async fn on_bw_done( 176 &mut self, 177 transaction_id: &f64, 178 bandwidth: &f64, 179 ) -> Result<(), NetStreamError> { 180 self.amf0_writer.write_string(&String::from("onBWDone"))?; 181 self.amf0_writer.write_number(transaction_id)?; 182 self.amf0_writer.write_null()?; 183 self.amf0_writer.write_number(bandwidth)?; 184 185 let data = self.amf0_writer.extract_current_bytes(); 186 self.write_chunk(data).await?; 187 188 return Ok(()); 189 } 190 191 pub async fn on_status( 192 &mut self, 193 transaction_id: &f64, 194 level: &String, 195 code: &String, 196 description: &String, 197 ) -> Result<(), NetStreamError> { 198 self.amf0_writer.write_string(&String::from("onStatus"))?; 199 self.amf0_writer.write_number(transaction_id)?; 200 self.amf0_writer.write_null()?; 201 202 let mut properties_map = HashMap::new(); 203 204 properties_map.insert( 205 String::from("level"), 206 Amf0ValueType::UTF8String(level.clone()), 207 ); 208 properties_map.insert( 209 String::from("code"), 210 Amf0ValueType::UTF8String(code.clone()), 211 ); 212 properties_map.insert( 213 String::from("description"), 214 Amf0ValueType::UTF8String(description.clone()), 215 ); 216 217 self.amf0_writer.write_object(&properties_map)?; 218 219 let data = self.amf0_writer.extract_current_bytes(); 220 self.write_chunk(data).await?; 221 222 return Ok(()); 223 } 224 } 225