1 use crate::rtp::errors::PackerError; 2 use crate::rtp::errors::UnPackerError; 3 use crate::rtp::rtcp::rtcp_header::RtcpHeader; 4 use crate::rtp::rtcp::RTCP_RR; 5 use crate::rtp::rtcp::RTCP_SR; 6 use crate::rtp::utils::OnFrameFn; 7 use crate::rtp::utils::OnRtpPacketFn; 8 use crate::rtp::utils::OnRtpPacketFn2; 9 use crate::rtp::RtpPacket; 10 11 use super::rtp::rtp_aac::RtpAacPacker; 12 use super::rtp::rtp_h264::RtpH264Packer; 13 use super::rtp::rtp_h265::RtpH265Packer; 14 15 use super::rtp::rtp_aac::RtpAacUnPacker; 16 use super::rtp::rtp_h264::RtpH264UnPacker; 17 use super::rtp::rtp_h265::RtpH265UnPacker; 18 19 use super::rtp::rtcp::rtcp_context::RtcpContext; 20 use super::rtp::rtcp::rtcp_sr::RtcpSenderReport; 21 use super::rtp::utils::TPacker; 22 use super::rtp::utils::TUnPacker; 23 use super::rtsp_codec::RtspCodecId; 24 use super::rtsp_codec::RtspCodecInfo; 25 use crate::rtp::utils::Marshal; 26 use crate::rtp::utils::Unmarshal; 27 use byteorder::BigEndian; 28 use bytes::BytesMut; 29 use bytesio::bytes_errors::BytesWriteError; 30 use bytesio::bytes_reader::BytesReader; 31 use bytesio::bytes_writer::AsyncBytesWriter; 32 use bytesio::bytesio::TNetIO; 33 use rand::Rng; 34 use std::sync::Arc; 35 use tokio::sync::Mutex; 36 37 pub trait TRtpFunc { create_packer(&mut self, writer: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>)38 fn create_packer(&mut self, writer: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>); create_unpacker(&mut self)39 fn create_unpacker(&mut self); 40 } 41 42 pub struct RtpChannel { 43 codec_info: RtspCodecInfo, 44 pub rtp_packer: Option<Box<dyn TPacker>>, 45 pub rtp_unpacker: Option<Box<dyn TUnPacker>>, 46 ssrc: u32, 47 init_sequence: u16, 48 } 49 50 #[derive(Default)] 51 pub struct RtcpChannel { 52 recv_ctx: RtcpContext, 53 pub send_ctx: RtcpContext, 54 channel_identifier: u8, 55 } 56 57 impl RtpChannel { new(codec_info: RtspCodecInfo) -> Self58 pub fn new(codec_info: RtspCodecInfo) -> Self { 59 let ssrc: u32 = rand::thread_rng().gen(); 60 let mut rtp_channel = RtpChannel { 61 codec_info, 62 ssrc, 63 rtp_packer: None, 64 rtp_unpacker: None, 65 init_sequence: 0, 66 }; 67 rtp_channel.create_unpacker(); 68 rtp_channel 69 } 70 71 //Receive av frame from network -> pack AV frame to RTP packet -> send to stream hub on_packet(&mut self, reader: &mut BytesReader) -> Result<(), UnPackerError>72 pub fn on_packet(&mut self, reader: &mut BytesReader) -> Result<(), UnPackerError> { 73 if let Some(unpacker) = &mut self.rtp_unpacker { 74 unpacker.unpack(reader)?; 75 } 76 Ok(()) 77 } 78 79 //Receive av frame from stream hub -> pack -> send out on_frame( &mut self, nalus: &mut BytesMut, timestamp: u32, ) -> Result<(), PackerError>80 pub async fn on_frame( 81 &mut self, 82 nalus: &mut BytesMut, 83 timestamp: u32, 84 ) -> Result<(), PackerError> { 85 if let Some(packer) = &mut self.rtp_packer { 86 return packer.pack(nalus, timestamp).await; 87 } 88 Ok(()) 89 } 90 91 //Set handler for processing AV frame when unpack a whole AV frame 92 //from rtp packets received from network. on_frame_handler(&mut self, f: OnFrameFn)93 pub fn on_frame_handler(&mut self, f: OnFrameFn) { 94 if let Some(unpacker) = &mut self.rtp_unpacker { 95 unpacker.on_frame_handler(f); 96 } 97 } 98 99 //Set handler for processing rtp packet when packed a rtp packet on_packet_handler(&mut self, f: OnRtpPacketFn)100 pub fn on_packet_handler(&mut self, f: OnRtpPacketFn) { 101 if let Some(packer) = &mut self.rtp_packer { 102 packer.on_packet_handler(f); 103 } 104 } 105 106 //Set handler for processing received AV rtp packet from network on_packet_for_rtcp_handler(&mut self, f: OnRtpPacketFn2)107 pub fn on_packet_for_rtcp_handler(&mut self, f: OnRtpPacketFn2) { 108 if let Some(packer) = &mut self.rtp_packer { 109 packer.on_packet_for_rtcp_handler(f); 110 } 111 } 112 } 113 114 impl TRtpFunc for RtpChannel { create_unpacker(&mut self)115 fn create_unpacker(&mut self) { 116 match self.codec_info.codec_id { 117 RtspCodecId::H264 => { 118 self.rtp_unpacker = Some(Box::new(RtpH264UnPacker::new())); 119 } 120 RtspCodecId::H265 => { 121 self.rtp_unpacker = Some(Box::new(RtpH265UnPacker::new())); 122 } 123 RtspCodecId::AAC => { 124 self.rtp_unpacker = Some(Box::new(RtpAacUnPacker::new())); 125 } 126 RtspCodecId::G711A => {} 127 } 128 } create_packer(&mut self, io: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>)129 fn create_packer(&mut self, io: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>) { 130 match self.codec_info.codec_id { 131 RtspCodecId::H264 => { 132 self.rtp_packer = Some(Box::new(RtpH264Packer::new( 133 self.codec_info.payload_type, 134 self.ssrc, 135 self.init_sequence, 136 1400, 137 io, 138 ))); 139 } 140 RtspCodecId::H265 => { 141 self.rtp_packer = Some(Box::new(RtpH265Packer::new( 142 self.codec_info.payload_type, 143 self.ssrc, 144 self.init_sequence, 145 1400, 146 io, 147 ))); 148 } 149 RtspCodecId::AAC => { 150 self.rtp_packer = Some(Box::new(RtpAacPacker::new( 151 self.codec_info.payload_type, 152 self.ssrc, 153 self.init_sequence, 154 io, 155 ))); 156 } 157 RtspCodecId::G711A => {} 158 } 159 } 160 } 161 162 impl RtcpChannel { set_channel_identifier(&mut self, channel_idendifier: u8)163 pub fn set_channel_identifier(&mut self, channel_idendifier: u8) { 164 self.channel_identifier = channel_idendifier; 165 } 166 on_rtcp( &mut self, reader: &mut BytesReader, rtcp_io: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>, )167 pub async fn on_rtcp( 168 &mut self, 169 reader: &mut BytesReader, 170 rtcp_io: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>, 171 ) { 172 let mut reader_clone = BytesReader::new(reader.get_remaining_bytes()); 173 if let Ok(rtcp_header) = RtcpHeader::unmarshal(&mut reader_clone) { 174 match rtcp_header.payload_type { 175 RTCP_SR => { 176 if let Ok(sr) = RtcpSenderReport::unmarshal(reader) { 177 self.recv_ctx.received_sr(&sr); 178 if let Err(err) = self.send_rr(rtcp_io).await { 179 log::error!("send rr error: {}", err); 180 } 181 } 182 } 183 RTCP_RR => {} 184 _ => {} 185 } 186 } 187 } 188 on_packet(&mut self, packet: RtpPacket)189 pub fn on_packet(&mut self, packet: RtpPacket) { 190 self.recv_ctx.received_rtp(packet); 191 } 192 send_rr( &mut self, rtcp_io: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>, ) -> Result<(), BytesWriteError>193 pub async fn send_rr( 194 &mut self, 195 rtcp_io: Arc<Mutex<Box<dyn TNetIO + Send + Sync>>>, 196 ) -> Result<(), BytesWriteError> { 197 let rr = self.recv_ctx.generate_rr(); 198 199 let net_type = rtcp_io.lock().await.get_net_type(); 200 if let Ok(msg) = rr.marshal() { 201 let mut bytes_writer = AsyncBytesWriter::new(rtcp_io); 202 match net_type { 203 bytesio::bytesio::NetType::TCP => { 204 bytes_writer.write_u8(0x24)?; 205 bytes_writer.write_u8(self.channel_identifier)?; 206 bytes_writer.write_u16::<BigEndian>(msg.len() as u16)?; 207 } 208 bytesio::bytesio::NetType::UDP => {} 209 } 210 bytes_writer.write(&msg)?; 211 bytes_writer.flush().await?; 212 } 213 Ok(()) 214 } 215 } 216