Merge pull request restsend#5 from yeoleobun/tts

fix(tts): timestamp gap between tts track, and clock rate for g722

Author: shenjinti

src/media/tests/tts_track.rs

@@ -537,7 +537,7 @@ async fn test_tts_track_end_of_stream() -> Result<()> {
 
 #[tokio::test]
 async fn test_tts_track_base64() -> Result<()> {
-    // Create a command channel
+        // Create a command channel
     let (command_tx, command_rx) = mpsc::unbounded_channel();
 
     // Create a TtsTrack with non-streaming mode
@@ -567,16 +567,27 @@ async fn test_tts_track_base64() -> Result<()> {
     command_tx.send(SynthesisCommand {
         text,
         base64: true,
+        end_of_stream: true,
         ..Default::default()
     })?;
 
-    let mut bytes_received = 0;
+    let mut sample_received = 0;
     let timeout = tokio::time::sleep(Duration::from_millis(3000));
     tokio::pin!(timeout);
     loop {
         tokio::select! {
-            _ = &mut timeout => {
-                break;
+            biased;
+            packet = packet_rx.recv() => {
+                match packet {
+                    Some(packet) => {
+                        if let Samples::PCM { samples } = &packet.samples {
+                            sample_received += samples.len();
+                        }
+                    }
+                    None => {
+                        break
+                    }
+                }
             }
             event = event_rx.recv() => {
                 match event {
@@ -589,20 +600,14 @@ async fn test_tts_track_base64() -> Result<()> {
                     _ => {}
                 }
             }
-            packet = packet_rx.recv() => {
-                match packet {
-                    Some(packet) => {
-                        if let Samples::PCM { samples } = &packet.samples {
-                            bytes_received += samples.len();
-                        }
-                    }
-                    None => {
-                        break
-                    }
-                }
+            _ = &mut timeout => {
+                break;
             }
+
+
         }
     }
-    assert!(bytes_received >= 16000, "Not enough bytes");
+
+    assert!(sample_received >= 8000, "Not enough bytes");
     Ok(())
 }

src/media/track/rtc.rs

@@ -21,7 +21,10 @@ use rustrtc::{
         track::SampleStreamTrack,
     },
 };
-use std::{sync::Arc, time::Duration};
+use std::{
+    sync::Arc,
+    time::{Duration, Instant},
+};
 use tokio::sync::Mutex;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, info};
@@ -61,10 +64,11 @@ pub struct RtcTrack {
     local_source: Option<Arc<SampleStreamSource>>,
     encoder: TrackCodec,
     ssrc: u32,
-    payload_type: Arc<std::sync::atomic::AtomicU8>,
+    payload_type: u8,
     pub peer_connection: Option<Arc<PeerConnection>>,
-    next_rtp_timestamp: Arc<std::sync::atomic::AtomicU32>,
-    next_rtp_sequence_number: Arc<std::sync::atomic::AtomicU16>,
+    next_rtp_timestamp: u32,
+    next_rtp_sequence_number: u16,
+    last_packet_time: Option<Instant>,
 }
 
 impl RtcTrack {
@@ -85,10 +89,11 @@ impl RtcTrack {
             local_source: None,
             encoder: TrackCodec::new(),
             ssrc: 0,
-            payload_type: Arc::new(std::sync::atomic::AtomicU8::new(0)),
+            payload_type: 0,
             peer_connection: None,
-            next_rtp_timestamp: Arc::new(std::sync::atomic::AtomicU32::new(0)),
-            next_rtp_sequence_number: Arc::new(std::sync::atomic::AtomicU16::new(0)),
+            next_rtp_timestamp: 0,
+            next_rtp_sequence_number: 0,
+            last_packet_time: None,
         }
     }
 
@@ -164,8 +169,7 @@ impl RtcTrack {
             .payload_type
             .unwrap_or_else(|| codec.payload_type());
 
-        self.payload_type
-            .store(payload_type, std::sync::atomic::Ordering::SeqCst);
+        self.payload_type = payload_type;
 
         let params = RtpCodecParameters {
             clock_rate: codec.clock_rate(),
@@ -181,7 +185,7 @@ impl RtcTrack {
             peer_connection.clone(),
             self.track_id.clone(),
             self.processor_chain.clone(),
-            self.payload_type.clone(),
+            self.payload_type,
         );
 
         if self.rtc_config.mode == TransportMode::Rtp {
@@ -195,7 +199,7 @@ impl RtcTrack {
                         self.track_id.clone(),
                         self.cancel_token.clone(),
                         self.processor_chain.clone(),
-                        self.payload_type.clone(),
+                        self.payload_type,
                     );
                 }
             }
@@ -209,7 +213,7 @@ impl RtcTrack {
         pc: Arc<PeerConnection>,
         track_id: TrackId,
         processor_chain: ProcessorChain,
-        default_payload_type: Arc<std::sync::atomic::AtomicU8>,
+        default_payload_type: u8,
     ) {
         let cancel_token = self.cancel_token.clone();
         let packet_sender = self.packet_sender.clone();
@@ -282,7 +286,7 @@ impl RtcTrack {
         track_id: TrackId,
         cancel_token: CancellationToken,
         processor_chain: ProcessorChain,
-        default_payload_type: Arc<std::sync::atomic::AtomicU8>,
+        default_payload_type: u8,
     ) {
         let (tx, mut rx) =
             tokio::sync::mpsc::unbounded_channel::<rustrtc::media::frame::AudioFrame>();
@@ -303,7 +307,7 @@ impl RtcTrack {
                     &track_id_proc,
                     &packet_sender_proc,
                     &mut processor_chain_proc,
-                    default_payload_type.clone(),
+                    default_payload_type,
                 )
                 .await;
             }
@@ -335,13 +339,11 @@ impl RtcTrack {
         track_id: &TrackId,
         packet_sender: &Arc<Mutex<Option<TrackPacketSender>>>,
         processor_chain: &mut ProcessorChain,
-        default_payload_type: Arc<std::sync::atomic::AtomicU8>,
+        default_payload_type: u8,
     ) {
         let packet_sender = packet_sender.lock().await;
         if let Some(sender) = packet_sender.as_ref() {
-            let payload_type = frame
-                .payload_type
-                .unwrap_or_else(|| default_payload_type.load(std::sync::atomic::Ordering::SeqCst));
+            let payload_type = frame.payload_type.unwrap_or(default_payload_type);
             let src_codec = match CodecType::try_from(payload_type) {
                 Ok(c) => c,
                 Err(_) => {
@@ -401,8 +403,7 @@ impl RtcTrack {
                     if let Some(codec) = codec {
                         if codec != CodecType::TelephoneEvent {
                             info!(track_id=%self.track_id, "Negotiated primary audio PT {} ({:?})", pt, codec);
-                            self.payload_type
-                                .store(pt, std::sync::atomic::Ordering::SeqCst);
+                            self.payload_type = pt;
                             break;
                         }
                     }
@@ -512,22 +513,32 @@ impl Track for RtcTrack {
                     let (_, encoded) = self.encoder.encode(payload_type, packet.clone());
                     let target_codec = CodecType::try_from(payload_type)?;
                     if !encoded.is_empty() {
-                        let target_samples =
-                            (samples.len() as u64 * target_codec.samplerate() as u64
-                                / packet.sample_rate as u64) as u32;
-                        let sample_count_per_channel =
-                            target_samples / self.track_config.channels as u32;
-                        let rtp_timestamp = self.next_rtp_timestamp.fetch_add(
-                            sample_count_per_channel,
-                            std::sync::atomic::Ordering::SeqCst,
-                        );
-                        let sequence_number = self
-                            .next_rtp_sequence_number
-                            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                        let clock_rate = target_codec.clock_rate();
+
+                        let now = Instant::now();
+                        if let Some(last_time) = self.last_packet_time {
+                            let elapsed = now.duration_since(last_time);
+                            if elapsed.as_millis() > 50 {
+                                let gap_increment =
+                                    (elapsed.as_millis() as u32 * clock_rate) / 1000;
+                                self.next_rtp_timestamp += gap_increment;
+                            }
+                        }
+
+                        self.last_packet_time = Some(now);
+
+                        let timestamp_increment = (samples.len() as u64 * clock_rate as u64
+                            / packet.sample_rate as u64
+                            / self.track_config.channels as u64)
+                            as u32;
+                        let rtp_timestamp = self.next_rtp_timestamp;
+                        self.next_rtp_timestamp += timestamp_increment;
+                        let sequence_number = self.next_rtp_sequence_number;
+                        self.next_rtp_sequence_number += 1;
 
                         let frame = RtcAudioFrame {
                             data: Bytes::from(encoded),
-                            clock_rate: target_codec.clock_rate(),
+                            clock_rate,
                             payload_type: Some(payload_type),
                             sequence_number: Some(sequence_number),
                             rtp_timestamp,
@@ -540,28 +551,36 @@ impl Track for RtcTrack {
                     payload_type,
                     sequence_number,
                 } => {
-                    let target_sample_rate = match *payload_type {
-                        0 | 8 | 18 => 8000,
-                        9 => 16000,
+                    let clock_rate = match *payload_type {
+                        0 | 8 | 9 | 18 => 8000,
                         111 => 48000,
                         _ => packet.sample_rate,
                     };
 
-                    // Estimate samples if we don't have them
+                    let now = Instant::now();
+                    if let Some(last_time) = self.last_packet_time {
+                        let elapsed = now.duration_since(last_time);
+                        if elapsed.as_millis() > 50 {
+                            let gap_increment = (elapsed.as_millis() as u32 * clock_rate) / 1000;
+                            self.next_rtp_timestamp += gap_increment;
+                        }
+                    }
+                    self.last_packet_time = Some(now);
+
                     let increment = match *payload_type {
                         0 | 8 | 18 => payload.len() as u32,
-                        9 => (payload.len() * 2) as u32,
-                        111 => (target_sample_rate / 50) as u32, // Assume 20ms for Opus if unknown
-                        _ => (target_sample_rate / 50) as u32,
+                        9 => payload.len() as u32,
+                        111 => (clock_rate / 50) as u32,
+                        _ => (clock_rate / 50) as u32,
                     };
-                    let rtp_timestamp = self
-                        .next_rtp_timestamp
-                        .fetch_add(increment, std::sync::atomic::Ordering::SeqCst);
+
+                    let rtp_timestamp = self.next_rtp_timestamp;
+                    self.next_rtp_timestamp += increment;
                     let sequence_number = *sequence_number;
 
                     let frame = RtcAudioFrame {
                         data: Bytes::from(payload.clone()),
-                        clock_rate: target_sample_rate,
+                        clock_rate,
                         payload_type: Some(*payload_type),
                         sequence_number: Some(sequence_number),
                         rtp_timestamp,
@@ -577,7 +596,7 @@ impl Track for RtcTrack {
 
 impl RtcTrack {
     fn get_payload_type(&self) -> u8 {
-        let pt = self.payload_type.load(std::sync::atomic::Ordering::SeqCst);
+        let pt = self.payload_type;
         if pt != 0 {
             return pt;
         }

src/media/track/tts.rs

@@ -1,12 +1,6 @@
 use crate::{
     event::{EventSender, SessionEvent},
-    media::AudioFrame,
-    media::Samples,
-    media::{
-        cache,
-        processor::ProcessorChain,
-        track::{Track, TrackConfig, TrackId, TrackPacketSender},
-    },
+    media::{AudioFrame, Samples, cache, processor::ProcessorChain, track::{Track, TrackConfig, TrackId, TrackPacketSender}},
     synthesis::{
         Subtitle, SynthesisClient, SynthesisCommand, SynthesisCommandReceiver,
         SynthesisCommandSender, SynthesisEvent, bytes_size_to_duration,
@@ -18,6 +12,7 @@ use audio_codec::bytes_to_samples;
 use base64::{Engine, prelude::BASE64_STANDARD};
 use bytes::{Bytes, BytesMut};
 use futures::StreamExt;
+use unic_emoji::char::is_emoji;
 use std::{
     collections::{HashMap, VecDeque},
     sync::{
@@ -97,6 +92,12 @@ struct TtsTask {
     graceful: Arc<AtomicBool>,
 }
 
+pub fn strip_emoji_chars(text: &str) -> String {
+    text.chars()
+        .filter(|&c| c.is_ascii() || !is_emoji(c))
+        .collect()
+}
+
 impl TtsTask {
     async fn run(mut self) -> Result<()> {
         let mut stream;
@@ -370,6 +371,8 @@ impl TtsTask {
 
         let emit_entry = self.get_emit_entry_mut(assume_seq);
 
+        let text = strip_emoji_chars(text);
+
         // if text is empty:
         // in streaming mode, skip it
         // in non streaming mode, set entry[seq] finished to true

src/synthesis/mod.rs

@@ -28,7 +28,6 @@ pub struct SynthesisCommand {
 }
 pub type SynthesisCommandSender = mpsc::UnboundedSender<SynthesisCommand>;
 pub type SynthesisCommandReceiver = mpsc::UnboundedReceiver<SynthesisCommand>;
-pub use self::tencent_cloud::strip_emoji_chars;
 
 #[derive(Debug, Clone, Serialize, Hash, Eq, PartialEq)]
 pub enum SynthesisType {

src/synthesis/tencent_cloud.rs

@@ -20,7 +20,6 @@ use tokio_tungstenite::{
     MaybeTlsStream, WebSocketStream, connect_async, tungstenite::protocol::Message,
 };
 use tracing::{debug, warn};
-use unic_emoji::char::is_emoji;
 use urlencoding;
 use uuid::Uuid;
 
@@ -98,16 +97,6 @@ impl From<&TencentSubtitle> for Subtitle {
     }
 }
 
-// tencent cloud will crash if text contains emoji
-// Only remove non-ASCII emoji characters. Keep all ASCII (digits, letters, punctuation),
-// since some ASCII (e.g., '0'..'9', '#', '*') are marked with the Unicode Emoji property
-// due to keycap sequences but are safe and expected in text.
-pub fn strip_emoji_chars(text: &str) -> String {
-    text.chars()
-        .filter(|&c| c.is_ascii() || !is_emoji(c))
-        .collect()
-}
-
 // construct request url
 // for non-streaming client, text is Some
 // session_id is used for tencent cloud tts service, not the session_id of media session
@@ -316,8 +305,7 @@ impl SynthesisClient for RealTimeClient {
         option: Option<SynthesisOption>,
     ) -> Result<()> {
         if let Some(tx) = &self.tx {
-            let text = strip_emoji_chars(text);
-            tx.send((text, cmd_seq, option))?;
+            tx.send((text.to_string(), cmd_seq, option))?;
         } else {
             return Err(anyhow::anyhow!("TencentCloud TTS: missing client sender"));
         };
@@ -419,7 +407,6 @@ impl SynthesisClient for StreamingClient {
         _option: Option<SynthesisOption>,
     ) -> Result<()> {
         if let Some(sink) = &mut self.sink {
-            let text = strip_emoji_chars(text);
             let request = WebSocketRequest::synthesis_action(&self.session_id, &text);
             let data = serde_json::to_string(&request)?;
             sink.send(Message::Text(data.into())).await?;