fix RTMP publish single AAC from ffmpeg client.

harlanc · Apr 10, 2024 · 66a1c88 · 66a1c88
1 parent d3a9bbf
commit 66a1c88
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 10 deletions.
diff --git a/library/bytesio/src/bytesio.rs b/library/bytesio/src/bytesio.rs
@@ -33,18 +33,25 @@ pub struct UdpIO {
 
 impl UdpIO {
     pub async fn new(remote_domain: String, remote_port: u16, local_port: u16) -> Option<Self> {
-        let remote_address = format!("{remote_domain}:{remote_port}");
+        let remote_address = if remote_domain == "localhost" {
+            format!("127.0.0.1:{remote_port}")
+        } else {
+            format!("{remote_domain}:{remote_port}")
+        };
         log::info!("remote address: {}", remote_address);
         let local_address = format!("0.0.0.0:{local_port}");
         if let Ok(local_socket) = UdpSocket::bind(local_address).await {
             if let Ok(remote_socket_addr) = remote_address.parse::<SocketAddr>() {
                 if let Err(err) = local_socket.connect(remote_socket_addr).await {
                     log::info!("connect to remote udp socket error: {}", err);
                 }
+
+                return Some(Self {
+                    socket: local_socket,
+                });
+            } else {
+                log::error!("remote_address parse error: {:?}", remote_address);
             }
-            return Some(Self {
-                socket: local_socket,
-            });
         }
 
         None
@@ -75,7 +82,7 @@ impl TNetIO for UdpIO {
             Ok(data) => data,
             Err(err) => Err(BytesIOError {
                 value: BytesIOErrorValue::TimeoutError(err),
-            })
+            }),
         }
     }
 
@@ -120,7 +127,7 @@ impl TNetIO for TcpIO {
             Ok(data) => data,
             Err(err) => Err(BytesIOError {
                 value: BytesIOErrorValue::TimeoutError(err),
-            })
+            }),
         }
     }
 

diff --git a/library/container/flv/src/demuxer.rs b/library/container/flv/src/demuxer.rs
@@ -185,7 +185,10 @@ impl FlvAudioTagDemuxer {
         if tag_header.sound_format == SoundFormat::AAC as u8 {
             match tag_header.aac_packet_type {
                 aac_packet_type::AAC_SEQHDR => {
-                    self.aac_processor.audio_specific_config_load()?;
+                    if self.aac_processor.bytes_reader.len() >= 2 {
+                        self.aac_processor.audio_specific_config_load()?;
+                    }
+
                     return Ok(FlvDemuxerAudioData::new());
                 }
                 aac_packet_type::AAC_RAW => {

diff --git a/protocol/hls/src/flv2hls.rs b/protocol/hls/src/flv2hls.rs
@@ -137,6 +137,10 @@ impl Flv2HlsRemuxer {
                 dts = data.dts;
                 pid = self.audio_pid;
                 payload.extend_from_slice(&data.data[..]);
+
+                if dts - self.last_ts_dts >= self.duration * 1000 {
+                    self.need_new_segment = true;
+                }
             }
             _ => return Ok(()),
         }

diff --git a/protocol/rtmp/src/cache/mod.rs b/protocol/rtmp/src/cache/mod.rs
@@ -33,7 +33,6 @@ pub struct Cache {
 
 impl Cache {
     pub fn new(gop_num: usize, statistic_data_sender: Option<StatisticDataSender>) -> Self {
-
         Cache {
             metadata: metadata::MetaData::new(),
             metadata_timestamp: 0,
@@ -78,7 +77,10 @@ impl Cache {
         let mut reader = BytesReader::new(chunk_body.clone());
         let tag_header = AudioTagHeader::unmarshal(&mut reader)?;
 
-        if tag_header.sound_format == define::SoundFormat::AAC as u8
+        let remain_bytes = reader.extract_remaining_bytes();
+
+        if remain_bytes.len() >= 2
+            && tag_header.sound_format == define::SoundFormat::AAC as u8
             && tag_header.aac_packet_type == define::aac_packet_type::AAC_SEQHDR
         {
             self.audio_seq = chunk_body.clone();
@@ -88,7 +90,7 @@ impl Cache {
                 let mut aac_processor = Mpeg4AacProcessor::default();
 
                 let aac = aac_processor
-                    .extend_data(reader.extract_remaining_bytes())
+                    .extend_data(remain_bytes)
                     .audio_specific_config_load()?;
 
                 let statistic_audio_codec = StatisticData::AudioCodec {