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 networkio::{bytes_writer::BytesWriter, networkio::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 pub async fn release_stream( 96 &mut self, 97 transaction_id: &f64, 98 stream_name: &String, 99 ) -> Result<(), NetStreamError> { 100 self.amf0_writer 101 .write_string(&String::from("releaseStream"))?; 102 self.amf0_writer.write_number(transaction_id)?; 103 self.amf0_writer.write_null()?; 104 self.amf0_writer.write_string(stream_name)?; 105 106 let data = self.amf0_writer.extract_current_bytes(); 107 self.write_chunk(data).await?; 108 109 return Ok(()); 110 } 111 112 pub async fn fcpublish( 113 &mut self, 114 transaction_id: &f64, 115 stream_name: &String, 116 ) -> Result<(), NetStreamError> { 117 self.amf0_writer.write_string(&String::from("FCPublish"))?; 118 self.amf0_writer.write_number(transaction_id)?; 119 self.amf0_writer.write_null()?; 120 self.amf0_writer.write_string(stream_name)?; 121 122 let data = self.amf0_writer.extract_current_bytes(); 123 self.write_chunk(data).await?; 124 125 return Ok(()); 126 } 127 128 #[allow(dead_code)] 129 async fn receive_audio( 130 &mut self, 131 transaction_id: &f64, 132 enable: &bool, 133 ) -> Result<(), NetStreamError> { 134 self.amf0_writer 135 .write_string(&String::from("receiveAudio"))?; 136 self.amf0_writer.write_number(transaction_id)?; 137 self.amf0_writer.write_null()?; 138 self.amf0_writer.write_bool(enable)?; 139 140 let data = self.amf0_writer.extract_current_bytes(); 141 self.write_chunk(data).await?; 142 143 return Ok(()); 144 } 145 #[allow(dead_code)] 146 async fn receive_video( 147 &mut self, 148 transaction_id: &f64, 149 enable: &bool, 150 ) -> Result<(), NetStreamError> { 151 self.amf0_writer 152 .write_string(&String::from("receiveVideo"))?; 153 self.amf0_writer.write_number(transaction_id)?; 154 self.amf0_writer.write_null()?; 155 self.amf0_writer.write_bool(enable)?; 156 157 let data = self.amf0_writer.extract_current_bytes(); 158 self.write_chunk(data).await?; 159 160 return Ok(()); 161 } 162 pub async fn publish( 163 &mut self, 164 transaction_id: &f64, 165 stream_name: &String, 166 stream_type: &String, 167 ) -> Result<(), NetStreamError> { 168 self.amf0_writer.write_string(&String::from("publish"))?; 169 self.amf0_writer.write_number(transaction_id)?; 170 self.amf0_writer.write_null()?; 171 self.amf0_writer.write_string(stream_name)?; 172 self.amf0_writer.write_string(stream_type)?; 173 174 let data = self.amf0_writer.extract_current_bytes(); 175 self.write_chunk(data).await?; 176 177 return Ok(()); 178 } 179 #[allow(dead_code)] 180 async fn seek(&mut self, transaction_id: &f64, ms: &f64) -> Result<(), NetStreamError> { 181 self.amf0_writer.write_string(&String::from("seek"))?; 182 self.amf0_writer.write_number(transaction_id)?; 183 self.amf0_writer.write_null()?; 184 self.amf0_writer.write_number(ms)?; 185 186 let data = self.amf0_writer.extract_current_bytes(); 187 self.write_chunk(data).await?; 188 189 return Ok(()); 190 } 191 #[allow(dead_code)] 192 async fn pause( 193 &mut self, 194 transaction_id: &f64, 195 pause: &bool, 196 ms: &f64, 197 ) -> Result<(), NetStreamError> { 198 self.amf0_writer.write_string(&String::from("pause"))?; 199 self.amf0_writer.write_number(transaction_id)?; 200 self.amf0_writer.write_null()?; 201 self.amf0_writer.write_bool(pause)?; 202 self.amf0_writer.write_number(ms)?; 203 204 let data = self.amf0_writer.extract_current_bytes(); 205 self.write_chunk(data).await?; 206 207 return Ok(()); 208 } 209 210 #[allow(dead_code)] 211 async fn on_bw_done( 212 &mut self, 213 transaction_id: &f64, 214 bandwidth: &f64, 215 ) -> Result<(), NetStreamError> { 216 self.amf0_writer.write_string(&String::from("onBWDone"))?; 217 self.amf0_writer.write_number(transaction_id)?; 218 self.amf0_writer.write_null()?; 219 self.amf0_writer.write_number(bandwidth)?; 220 221 let data = self.amf0_writer.extract_current_bytes(); 222 self.write_chunk(data).await?; 223 224 return Ok(()); 225 } 226 227 pub async fn on_status( 228 &mut self, 229 transaction_id: &f64, 230 level: &String, 231 code: &String, 232 description: &String, 233 ) -> Result<(), NetStreamError> { 234 self.amf0_writer.write_string(&String::from("onStatus"))?; 235 self.amf0_writer.write_number(transaction_id)?; 236 self.amf0_writer.write_null()?; 237 238 let mut properties_map = HashMap::new(); 239 240 properties_map.insert( 241 String::from("level"), 242 Amf0ValueType::UTF8String(level.clone()), 243 ); 244 properties_map.insert( 245 String::from("code"), 246 Amf0ValueType::UTF8String(code.clone()), 247 ); 248 properties_map.insert( 249 String::from("description"), 250 Amf0ValueType::UTF8String(description.clone()), 251 ); 252 253 self.amf0_writer.write_object(&properties_map)?; 254 255 let data = self.amf0_writer.extract_current_bytes(); 256 self.write_chunk(data).await?; 257 258 return Ok(()); 259 } 260 } 261