active_call/media/track/
rtc.rs

1use super::track_codec::TrackCodec;
2use crate::{
3    event::{EventSender, SessionEvent},
4    media::AudioFrame,
5    media::{
6        processor::ProcessorChain,
7        track::{Track, TrackConfig, TrackId, TrackPacketSender},
8    },
9};
10use anyhow::Result;
11use async_trait::async_trait;
12use audio_codec::CodecType;
13use bytes::Bytes;
14use futures::StreamExt;
15use rustrtc::{
16    AudioCapability, IceServer, MediaKind, PeerConnection, PeerConnectionEvent,
17    PeerConnectionState, RtcConfiguration, RtpCodecParameters, SdpType, TransportMode,
18    config::MediaCapabilities,
19    media::{
20        MediaStreamTrack, SampleStreamSource, frame::AudioFrame as RtcAudioFrame, sample_track,
21        track::SampleStreamTrack,
22    },
23};
24use std::{
25    sync::Arc,
26    time::{Duration, Instant},
27};
28use tokio::sync::Mutex;
29use tokio_util::sync::CancellationToken;
30use tracing::{debug, info};
31
32#[derive(Clone)]
33pub struct RtcTrackConfig {
34    pub mode: TransportMode,
35    pub ice_servers: Option<Vec<IceServer>>,
36    pub external_ip: Option<String>,
37    pub rtp_port_range: Option<(u16, u16)>,
38    pub preferred_codec: Option<CodecType>,
39    pub codecs: Vec<CodecType>,
40    pub payload_type: Option<u8>,
41}
42
43impl Default for RtcTrackConfig {
44    fn default() -> Self {
45        Self {
46            mode: TransportMode::WebRtc, // Default WebRTC behavior
47            ice_servers: None,
48            external_ip: None,
49            rtp_port_range: None,
50            preferred_codec: None,
51            codecs: Vec::new(),
52            payload_type: None,
53        }
54    }
55}
56
57pub struct RtcTrack {
58    track_id: TrackId,
59    track_config: TrackConfig,
60    rtc_config: RtcTrackConfig,
61    processor_chain: ProcessorChain,
62    packet_sender: Arc<Mutex<Option<TrackPacketSender>>>,
63    cancel_token: CancellationToken,
64    local_source: Option<Arc<SampleStreamSource>>,
65    encoder: TrackCodec,
66    ssrc: u32,
67    payload_type: Option<u8>,
68    pub peer_connection: Option<Arc<PeerConnection>>,
69    next_rtp_timestamp: u32,
70    next_rtp_sequence_number: u16,
71    last_packet_time: Option<Instant>,
72    last_remote_sdp: Option<String>,
73    need_marker: bool,
74}
75
76impl RtcTrack {
77    pub fn new(
78        cancel_token: CancellationToken,
79        id: TrackId,
80        track_config: TrackConfig,
81        rtc_config: RtcTrackConfig,
82    ) -> Self {
83        let processor_chain = ProcessorChain::new(track_config.samplerate);
84        Self {
85            track_id: id,
86            track_config,
87            rtc_config,
88            processor_chain,
89            packet_sender: Arc::new(Mutex::new(None)),
90            cancel_token,
91            local_source: None,
92            encoder: TrackCodec::new(),
93            ssrc: 0,
94            payload_type: None,
95            peer_connection: None,
96            next_rtp_timestamp: 0,
97            next_rtp_sequence_number: 0,
98            last_packet_time: None,
99            last_remote_sdp: None,
100            need_marker: false,
101        }
102    }
103
104    pub fn with_ssrc(mut self, ssrc: u32) -> Self {
105        self.ssrc = ssrc;
106        self
107    }
108
109    pub fn create_audio_track(
110        _codec: CodecType,
111        _stream_id: Option<String>,
112    ) -> (Arc<SampleStreamSource>, Arc<SampleStreamTrack>) {
113        let (source, track, _) = sample_track(rustrtc::media::MediaKind::Audio, 100);
114        (Arc::new(source), track)
115    }
116
117    pub async fn local_description(&self) -> Result<String> {
118        let pc = self
119            .peer_connection
120            .as_ref()
121            .ok_or_else(|| anyhow::anyhow!("No PeerConnection"))?;
122        let offer = pc.create_offer().await?;
123        pc.set_local_description(offer.clone())?;
124        Ok(offer.to_sdp_string())
125    }
126
127    pub async fn create(&mut self) -> Result<()> {
128        if self.peer_connection.is_some() {
129            return Ok(());
130        }
131
132        let mut config = RtcConfiguration::default();
133        if self.ssrc != 0 {
134            config.ssrc_start = self.ssrc;
135        }
136        config.transport_mode = self.rtc_config.mode.clone();
137
138        if let Some(ice_servers) = &self.rtc_config.ice_servers {
139            config.ice_servers = ice_servers.clone();
140        }
141
142        if let Some(external_ip) = &self.rtc_config.external_ip {
143            config.external_ip = Some(external_ip.clone());
144        }
145
146        if !self.rtc_config.codecs.is_empty() {
147            let mut caps = MediaCapabilities::default();
148            caps.audio.clear();
149
150            for codec in &self.rtc_config.codecs {
151                let cap = match codec {
152                    CodecType::PCMU => AudioCapability::pcmu(),
153                    CodecType::PCMA => AudioCapability::pcma(),
154                    CodecType::G722 => AudioCapability::g722(),
155                    CodecType::G729 => AudioCapability::g729(),
156                    CodecType::TelephoneEvent => AudioCapability::telephone_event(),
157                    #[cfg(feature = "opus")]
158                    CodecType::Opus => AudioCapability::opus(),
159                };
160                caps.audio.push(cap);
161            }
162            config.media_capabilities = Some(caps);
163        }
164
165        let peer_connection = Arc::new(PeerConnection::new(config));
166        self.peer_connection = Some(peer_connection.clone());
167
168        let default_codec = CodecType::G722;
169        let codec = self.rtc_config.preferred_codec.unwrap_or(default_codec);
170
171        let (source, track) = Self::create_audio_track(codec, Some(self.track_id.clone()));
172        self.local_source = Some(source);
173
174        let payload_type = self
175            .rtc_config
176            .payload_type
177            .unwrap_or_else(|| codec.payload_type());
178
179        self.payload_type = Some(payload_type);
180
181        let params = RtpCodecParameters {
182            clock_rate: codec.clock_rate(),
183            channels: codec.channels() as u8,
184            payload_type,
185            ..Default::default()
186        };
187
188        peer_connection.add_track_with_stream_id(track, self.track_id.clone(), params)?;
189
190        // Spawn Handler Logic
191        self.spawn_handlers(
192            peer_connection.clone(),
193            self.track_id.clone(),
194            self.processor_chain.clone(),
195            payload_type,
196        );
197
198        if self.rtc_config.mode == TransportMode::Rtp {
199            for transceiver in peer_connection.get_transceivers() {
200                if let Some(receiver) = transceiver.receiver() {
201                    let track = receiver.track();
202                    info!(track_id=%self.track_id, "RTP mode: starting receiver track handler");
203                    Self::spawn_track_handler(
204                        track,
205                        self.packet_sender.clone(),
206                        self.track_id.clone(),
207                        self.cancel_token.clone(),
208                        self.processor_chain.clone(),
209                        self.get_payload_type(),
210                    );
211                }
212            }
213        }
214
215        Ok(())
216    }
217
218    fn spawn_handlers(
219        &self,
220        pc: Arc<PeerConnection>,
221        track_id: TrackId,
222        processor_chain: ProcessorChain,
223        default_payload_type: u8,
224    ) {
225        let cancel_token = self.cancel_token.clone();
226        let packet_sender = self.packet_sender.clone();
227        let pc_clone = pc.clone();
228        let track_id_log = track_id.clone();
229        let is_webrtc = self.rtc_config.mode != TransportMode::Rtp;
230
231        // 1. Event Loop
232        crate::spawn(async move {
233            info!(track_id=%track_id_log, "RtcTrack event loop started");
234            let mut events = futures::stream::unfold(pc_clone.clone(), |pc| async move {
235                pc.recv().await.map(|ev| (ev, pc))
236            })
237            .take_until(cancel_token.cancelled())
238            .boxed();
239
240            let mut event_count = 0;
241            while let Some(event) = events.next().await {
242                event_count += 1;
243                let event_type = match &event {
244                    PeerConnectionEvent::Track(_) => "Track",
245                    PeerConnectionEvent::DataChannel(_) => "DataChannel",
246                };
247                debug!(track_id=%track_id_log, "Received PeerConnectionEvent #{}: {}", event_count, event_type);
248
249                if let PeerConnectionEvent::Track(transceiver) = event {
250                    if let Some(receiver) = transceiver.receiver() {
251                        let track = receiver.track();
252                        info!(track_id=%track_id_log, "New track received");
253
254                        Self::spawn_track_handler(
255                            track,
256                            packet_sender.clone(),
257                            track_id_log.clone(),
258                            cancel_token.clone(),
259                            processor_chain.clone(),
260                            default_payload_type.clone(),
261                        );
262                    }
263                }
264            }
265            debug!(track_id=%track_id_log, "RtcTrack event loop ended, total events: {}", event_count);
266        });
267
268        // 2. State Monitoring
269        if is_webrtc {
270            let pc_state = pc.clone();
271            let cancel_token_state = self.cancel_token.clone();
272            let mut state_rx = pc_state.subscribe_peer_state();
273            let track_id_state = track_id.clone();
274
275            crate::spawn(async move {
276                while state_rx.changed().await.is_ok() {
277                    let s = *state_rx.borrow();
278                    debug!(track_id=%track_id_state, "peer connection state changed: {:?}", s);
279                    match s {
280                        PeerConnectionState::Disconnected
281                        | PeerConnectionState::Closed
282                        | PeerConnectionState::Failed => {
283                            info!(
284                                track_id = %track_id_state,
285                                "peer connection is {:?}, try to close", s
286                            );
287                            cancel_token_state.cancel();
288                            pc_state.close();
289                            break;
290                        }
291                        _ => {}
292                    }
293                }
294            });
295        }
296    }
297
298    fn spawn_track_handler(
299        track: Arc<SampleStreamTrack>,
300        packet_sender_arc: Arc<Mutex<Option<TrackPacketSender>>>,
301        track_id: TrackId,
302        cancel_token: CancellationToken,
303        processor_chain: ProcessorChain,
304        default_payload_type: u8,
305    ) {
306        let (tx, mut rx) =
307            tokio::sync::mpsc::unbounded_channel::<rustrtc::media::frame::AudioFrame>();
308
309        // Processing Worker
310        let track_id_proc = track_id.clone();
311        let packet_sender_proc = packet_sender_arc.clone();
312        let mut processor_chain_proc = processor_chain.clone();
313        let cancel_token_proc = cancel_token.clone();
314        crate::spawn(async move {
315            info!(track_id=%track_id_proc, "RtcTrack processing worker started");
316            while let Some(frame) = rx.recv().await {
317                if cancel_token_proc.is_cancelled() {
318                    break;
319                }
320                Self::process_audio_frame(
321                    frame,
322                    &track_id_proc,
323                    &packet_sender_proc,
324                    &mut processor_chain_proc,
325                    default_payload_type,
326                )
327                .await;
328            }
329            info!(track_id=%track_id_proc, "RtcTrack processing worker stopped");
330        });
331
332        // Receiving Worker
333        crate::spawn(async move {
334            let mut samples =
335                futures::stream::unfold(
336                    track,
337                    |t| async move { t.recv().await.ok().map(|s| (s, t)) },
338                )
339                .take_until(cancel_token.cancelled())
340                .boxed();
341
342            while let Some(sample) = samples.next().await {
343                if let rustrtc::media::frame::MediaSample::Audio(frame) = sample {
344                    if let Err(_) = tx.send(frame) {
345                        break;
346                    }
347                }
348            }
349        });
350    }
351
352    async fn process_audio_frame(
353        frame: rustrtc::media::frame::AudioFrame,
354        track_id: &TrackId,
355        packet_sender: &Arc<Mutex<Option<TrackPacketSender>>>,
356        processor_chain: &mut ProcessorChain,
357        default_payload_type: u8,
358    ) {
359        let packet_sender = packet_sender.lock().await;
360        if let Some(sender) = packet_sender.as_ref() {
361            let payload_type = frame.payload_type.unwrap_or(default_payload_type);
362            let src_codec = match CodecType::try_from(payload_type) {
363                Ok(c) => c,
364                Err(_) => {
365                    debug!(track_id=%track_id, "Unknown payload type {}, skipping frame", payload_type);
366                    return;
367                }
368            };
369
370            let mut af = AudioFrame {
371                track_id: track_id.clone(),
372                samples: crate::media::Samples::RTP {
373                    payload_type,
374                    payload: frame.data.to_vec(),
375                    sequence_number: frame.sequence_number.unwrap_or(0),
376                },
377                timestamp: crate::media::get_timestamp(),
378                sample_rate: src_codec.samplerate(),
379                channels: src_codec.channels(),
380            };
381            if let Err(e) = processor_chain.process_frame(&mut af) {
382                debug!(track_id=%track_id, "processor_chain process_frame error: {:?}", e);
383            }
384
385            sender.send(af).ok();
386        }
387    }
388
389    pub fn parse_sdp_payload_types(&mut self, sdp_type: SdpType, sdp_str: &str) -> Result<()> {
390        use crate::media::negotiate::parse_rtpmap;
391        let sdp = rustrtc::SessionDescription::parse(sdp_type, sdp_str)?;
392
393        if let Some(media) = sdp
394            .media_sections
395            .iter()
396            .find(|m| m.kind == MediaKind::Audio)
397        {
398            for attr in &media.attributes {
399                if attr.key == "rtpmap" {
400                    if let Some(value) = &attr.value {
401                        if let Ok((pt, codec, _, _)) = parse_rtpmap(value) {
402                            self.encoder.set_payload_type(pt, codec.clone());
403                            self.processor_chain.codec.set_payload_type(pt, codec);
404                        }
405                    }
406                }
407            }
408
409            for fmt in &media.formats {
410                if let Ok(pt) = fmt.parse::<u8>() {
411                    let codec = self
412                        .encoder
413                        .payload_type_map
414                        .get(&pt)
415                        .cloned()
416                        .or_else(|| CodecType::try_from(pt).ok());
417
418                    if let Some(codec) = codec {
419                        if codec != CodecType::TelephoneEvent {
420                            info!(track_id=%self.track_id, "Negotiated primary audio PT {} ({:?})", pt, codec);
421                            self.payload_type = Some(pt);
422                            break;
423                        }
424                    }
425                }
426            }
427        }
428        Ok(())
429    }
430
431    fn normalize_sdp(sdp: &str) -> String {
432        sdp.lines()
433            .map(|line| {
434                if line.starts_with("o=") {
435                    let parts: Vec<&str> = line.split_whitespace().collect();
436                    if parts.len() >= 3 {
437                        return format!("o= {} {}", parts[1], parts[2]);
438                    }
439                }
440                line.to_string()
441            })
442            .filter(|line| {
443                !line.starts_with("t=") &&  // timing line can vary
444                !line.starts_with("a=ssrc:") &&  // SSRC attributes (but SSRC change shows in o= version)
445                !line.starts_with("a=msid:") &&  // media stream ID
446                !line.trim().is_empty()
447            })
448            .collect::<Vec<_>>()
449            .join("\n")
450    }
451
452    async fn update_remote_description_internal(
453        &mut self,
454        answer: &String,
455        force_update: bool,
456    ) -> Result<()> {
457        if let Some(pc) = &self.peer_connection {
458            if !force_update {
459                if let Some(ref last_sdp) = self.last_remote_sdp {
460                    if Self::normalize_sdp(last_sdp) == Self::normalize_sdp(answer) {
461                        debug!(track_id=%self.track_id, "SDP unchanged, skipping update_remote_description");
462                        return Ok(());
463                    }
464                }
465            } else {
466                debug!(track_id=%self.track_id, "Force update requested, skipping SDP comparison");
467            }
468
469            let is_first_remote_sdp = self.last_remote_sdp.is_none();
470
471            let sdp_obj = rustrtc::SessionDescription::parse(rustrtc::SdpType::Answer, answer)?;
472            match pc.set_remote_description(sdp_obj.clone()).await {
473                Ok(_) => {
474                    debug!(track_id=%self.track_id, "set_remote_description succeeded");
475                    self.last_remote_sdp = Some(answer.clone());
476                }
477                Err(e) => {
478                    if self.rtc_config.mode == TransportMode::Rtp {
479                        info!(track_id=%self.track_id, "set_remote_description failed ({}), attempting to re-sync state for SIP update", e);
480
481                        if let Some(current_local) = pc.local_description() {
482                            let sdp = current_local.to_sdp_string();
483                            for line in sdp.lines() {
484                                if line.starts_with("a=ssrc:") {
485                                    info!(track_id=%self.track_id, "SSRC before re-sync: {}", line);
486                                }
487                            }
488                        }
489
490                        let offer = pc.create_offer().await?;
491
492                        let sdp = offer.to_sdp_string();
493                        for line in sdp.lines() {
494                            if line.starts_with("a=ssrc:") {
495                                info!(track_id=%self.track_id, "SSRC in new offer (re-sync): {}", line);
496                            }
497                        }
498
499                        pc.set_local_description(offer)?;
500                        pc.set_remote_description(sdp_obj).await?;
501                        self.last_remote_sdp = Some(answer.clone());
502                        info!(track_id=%self.track_id, "successfully re-synced WebRTC state for SIP update");
503                    } else {
504                        return Err(e.into());
505                    }
506                }
507            }
508
509            if is_first_remote_sdp && self.rtc_config.mode != TransportMode::Rtp {
510                for transceiver in pc.get_transceivers() {
511                    if let Some(receiver) = transceiver.receiver() {
512                        let track = receiver.track();
513                        info!(track_id=%self.track_id, "WebRTC mode: manually starting receiver track handler after first answer");
514                        Self::spawn_track_handler(
515                            track,
516                            self.packet_sender.clone(),
517                            self.track_id.clone(),
518                            self.cancel_token.clone(),
519                            self.processor_chain.clone(),
520                            self.get_payload_type(),
521                        );
522                    }
523                }
524            }
525
526            // Extract negotiated payload types from SDP string
527            self.parse_sdp_payload_types(rustrtc::SdpType::Answer, answer)?;
528        }
529        Ok(())
530    }
531}
532
533#[async_trait]
534impl Track for RtcTrack {
535    fn ssrc(&self) -> u32 {
536        self.ssrc
537    }
538    fn id(&self) -> &TrackId {
539        &self.track_id
540    }
541    fn config(&self) -> &TrackConfig {
542        &self.track_config
543    }
544    fn processor_chain(&mut self) -> &mut ProcessorChain {
545        &mut self.processor_chain
546    }
547
548    async fn handshake(&mut self, offer: String, _: Option<Duration>) -> Result<String> {
549        info!(track_id=%self.track_id, "rtc handshake start");
550        self.create().await?;
551
552        let pc = self.peer_connection.clone().ok_or_else(|| {
553            anyhow::anyhow!("No PeerConnection available for track {}", self.track_id)
554        })?;
555
556        debug!(track_id=%self.track_id, "Before set_remote_description: transceivers count = {}", pc.get_transceivers().len());
557        for (i, t) in pc.get_transceivers().iter().enumerate() {
558            debug!(track_id=%self.track_id, "  Transceiver #{}: kind={:?}, mid={:?}, direction={:?}", 
559                i, t.kind(), t.mid(), t.direction());
560        }
561
562        let sdp = rustrtc::SessionDescription::parse(rustrtc::SdpType::Offer, &offer)?;
563        pc.set_remote_description(sdp.clone()).await?;
564
565        debug!(track_id=%self.track_id, "After set_remote_description: transceivers count = {}", pc.get_transceivers().len());
566        for (i, t) in pc.get_transceivers().iter().enumerate() {
567            debug!(track_id=%self.track_id, "  Transceiver #{}: kind={:?}, mid={:?}, direction={:?}, has_receiver={}", 
568                i, t.kind(), t.mid(), t.direction(), t.receiver().is_some());
569        }
570
571        // CRITICAL FIX: When server-initiated signaling (common WebRTC pattern),
572        // Track events fire when remote peer sends offer, not when we accept answer.
573        // Since we received remote offer here and added local track first,
574        // we must manually start receiver tracks as Track events won't fire.
575        if self.rtc_config.mode != TransportMode::Rtp {
576            for transceiver in pc.get_transceivers() {
577                if let Some(receiver) = transceiver.receiver() {
578                    let track = receiver.track();
579                    info!(track_id=%self.track_id, "WebRTC handshake: manually starting receiver track handler for browser audio");
580                    Self::spawn_track_handler(
581                        track,
582                        self.packet_sender.clone(),
583                        self.track_id.clone(),
584                        self.cancel_token.clone(),
585                        self.processor_chain.clone(),
586                        self.get_payload_type(),
587                    );
588                }
589            }
590        }
591
592        self.parse_sdp_payload_types(rustrtc::SdpType::Offer, &offer)?;
593
594        let mut answer = pc.create_answer().await?;
595        crate::media::negotiate::intersect_answer(&sdp, &mut answer);
596
597        pc.set_local_description(answer.clone())?;
598
599        if self.rtc_config.mode != TransportMode::Rtp {
600            pc.wait_for_gathering_complete().await;
601        }
602
603        let final_answer = pc
604            .local_description()
605            .ok_or(anyhow::anyhow!("No local description"))?;
606
607        Ok(final_answer.to_sdp_string())
608    }
609
610    async fn update_remote_description(&mut self, answer: &String) -> Result<()> {
611        self.update_remote_description_internal(answer, false).await
612    }
613
614    async fn update_remote_description_force(&mut self, answer: &String) -> Result<()> {
615        self.update_remote_description_internal(answer, true).await
616    }
617
618    async fn start(
619        &mut self,
620        event_sender: EventSender,
621        packet_sender: TrackPacketSender,
622    ) -> Result<()> {
623        *self.packet_sender.lock().await = Some(packet_sender.clone());
624        let token_clone = self.cancel_token.clone();
625        let event_sender_clone = event_sender.clone();
626        let track_id = self.track_id.clone();
627        let ssrc = self.ssrc;
628
629        if self.rtc_config.mode != TransportMode::Rtp {
630            let start_time = crate::media::get_timestamp();
631            crate::spawn(async move {
632                token_clone.cancelled().await;
633                let _ = event_sender_clone.send(SessionEvent::TrackEnd {
634                    track_id,
635                    timestamp: crate::media::get_timestamp(),
636                    duration: crate::media::get_timestamp() - start_time,
637                    ssrc,
638                    play_id: None,
639                });
640            });
641        }
642
643        Ok(())
644    }
645
646    async fn stop(&self) -> Result<()> {
647        self.cancel_token.cancel();
648        if let Some(pc) = &self.peer_connection {
649            pc.close();
650        }
651        Ok(())
652    }
653
654    async fn send_packet(&mut self, packet: &AudioFrame) -> Result<()> {
655        let packet = packet.clone();
656
657        if let Some(source) = &self.local_source {
658            match &packet.samples {
659                crate::media::Samples::PCM { samples } => {
660                    let payload_type = self.get_payload_type();
661                    let (_, encoded) = self.encoder.encode(payload_type, packet.clone());
662                    let target_codec = CodecType::try_from(payload_type)?;
663                    if !encoded.is_empty() {
664                        let clock_rate = target_codec.clock_rate();
665
666                        let now = Instant::now();
667                        if let Some(last_time) = self.last_packet_time {
668                            let elapsed = now.duration_since(last_time);
669                            if elapsed.as_millis() > 50 {
670                                let gap_increment =
671                                    (elapsed.as_millis() as u32 * clock_rate) / 1000;
672                                self.next_rtp_timestamp += gap_increment;
673                                self.need_marker = true;
674                            }
675                        }
676
677                        self.last_packet_time = Some(now);
678
679                        let timestamp_increment = (samples.len() as u64 * clock_rate as u64
680                            / packet.sample_rate as u64
681                            / self.track_config.channels as u64)
682                            as u32;
683                        let rtp_timestamp = self.next_rtp_timestamp;
684                        self.next_rtp_timestamp += timestamp_increment;
685                        let sequence_number = self.next_rtp_sequence_number;
686                        self.next_rtp_sequence_number += 1;
687
688                        let mut marker = false;
689                        if self.need_marker {
690                            marker = true;
691                            self.need_marker = false;
692                        }
693
694                        let frame = RtcAudioFrame {
695                            data: Bytes::from(encoded),
696                            clock_rate,
697                            payload_type: Some(payload_type),
698                            sequence_number: Some(sequence_number),
699                            rtp_timestamp,
700                            marker,
701                            ..Default::default()
702                        };
703                        source.send_audio(frame).await.ok();
704                    }
705                }
706                crate::media::Samples::RTP {
707                    payload,
708                    payload_type,
709                    sequence_number,
710                } => {
711                    let clock_rate = match *payload_type {
712                        0 | 8 | 9 | 18 => 8000,
713                        111 => 48000,
714                        _ => packet.sample_rate,
715                    };
716
717                    let now = Instant::now();
718                    if let Some(last_time) = self.last_packet_time {
719                        let elapsed = now.duration_since(last_time);
720                        if elapsed.as_millis() > 50 {
721                            let gap_increment = (elapsed.as_millis() as u32 * clock_rate) / 1000;
722                            self.next_rtp_timestamp += gap_increment;
723                            self.need_marker = true;
724                        }
725                    }
726                    self.last_packet_time = Some(now);
727
728                    let increment = match *payload_type {
729                        0 | 8 | 18 => payload.len() as u32,
730                        9 => payload.len() as u32,
731                        111 => (clock_rate / 50) as u32,
732                        _ => (clock_rate / 50) as u32,
733                    };
734
735                    let rtp_timestamp = self.next_rtp_timestamp;
736                    self.next_rtp_timestamp += increment;
737                    let sequence_number = *sequence_number;
738
739                    let mut marker = false;
740                    if self.need_marker {
741                        marker = true;
742                        self.need_marker = false;
743                    }
744
745                    let frame = RtcAudioFrame {
746                        data: Bytes::from(payload.clone()),
747                        clock_rate,
748                        payload_type: Some(*payload_type),
749                        sequence_number: Some(sequence_number),
750                        rtp_timestamp,
751                        marker,
752                        ..Default::default()
753                    };
754                    source.send_audio(frame).await.ok();
755                }
756                _ => {}
757            }
758        }
759        Ok(())
760    }
761}
762
763impl RtcTrack {
764    fn get_payload_type(&self) -> u8 {
765        if let Some(pt) = self.payload_type {
766            return pt;
767        }
768
769        self.rtc_config.payload_type.unwrap_or_else(|| {
770            match self.rtc_config.preferred_codec.unwrap_or(CodecType::Opus) {
771                CodecType::PCMU => 0,
772                CodecType::PCMA => 8,
773                CodecType::Opus => 111,
774                CodecType::G722 => 9,
775                _ => 111,
776            }
777        })
778    }
779}
780
781#[cfg(test)]
782mod tests {
783    use super::*;
784    use crate::media::track::TrackConfig;
785
786    #[test]
787    fn test_parse_sdp_payload_types() {
788        let track_id = "test-track".to_string();
789        let cancel_token = CancellationToken::new();
790        let mut track = RtcTrack::new(
791            cancel_token,
792            track_id,
793            TrackConfig::default(),
794            RtcTrackConfig::default(),
795        );
796
797        // Case 1: Multiple audio codecs, telephone-event at the end. Primary should be PCMA (8)
798        let sdp1 = "v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\ns=-\r\nc=IN IP4 127.0.0.1\r\nt=0 0\r\nm=audio 1234 RTP/AVP 8 0 101\r\na=rtpmap:8 PCMA/8000\r\na=rtpmap:0 PCMU/8000\r\na=rtpmap:101 telephone-event/8000\r\n";
799        track
800            .parse_sdp_payload_types(rustrtc::SdpType::Offer, sdp1)
801            .expect("parse offer");
802        assert_eq!(track.get_payload_type(), 8);
803
804        // Case 2: telephone-event at the beginning, should skip it and pick PCMU (0)
805        let mut rtc_config = RtcTrackConfig::default();
806        rtc_config.preferred_codec = Some(CodecType::PCMU);
807        let mut track2 = RtcTrack::new(
808            CancellationToken::new(),
809            "test-track-2".to_string(),
810            TrackConfig::default(),
811            rtc_config,
812        );
813
814        let sdp2 = "v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\ns=-\r\nc=IN IP4 127.0.0.1\r\nt=0 0\r\nm=audio 1234 RTP/AVP 101 0 8\r\na=rtpmap:101 telephone-event/8000\r\na=rtpmap:0 PCMU/8000\r\na=rtpmap:8 PCMA/8000\r\n";
815        track2
816            .parse_sdp_payload_types(rustrtc::SdpType::Offer, sdp2)
817            .expect("parse offer");
818        assert_eq!(track2.get_payload_type(), 0);
819
820        // Case 3: Opus with dynamic payload type 111
821        let sdp3 = "v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\ns=-\r\nc=IN IP4 127.0.0.1\r\nt=0 0\r\nm=audio 1234 RTP/AVP 111 101\r\na=rtpmap:111 opus/48000/2\r\na=rtpmap:101 telephone-event/8000\r\n";
822        track
823            .parse_sdp_payload_types(rustrtc::SdpType::Offer, sdp3)
824            .expect("parse offer");
825        assert_eq!(track.get_payload_type(), 111);
826    }
827}