Skip to content

Commit f659f1a

Browse files
shenjintishenjinti
committed
feat: enhance SIP session handling with media path updates and fallback options in conference setup
1 parent 825be50 commit f659f1a

4 files changed

Lines changed: 240 additions & 80 deletions

File tree

‎src/proxy/proxy_call/sip_session.rs‎

Lines changed: 85 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -7553,6 +7553,7 @@ impl SipSession {
75537553
match self.accept_call(None, answer_sdp, None).await {
75547554
Ok(()) => {
75557555
self.update_leg_state(&leg_id, LegState::Connected);
7556+
self.update_media_path().await;
75567557
CommandResult::success()
75577558
}
75587559
Err(e) => CommandResult::failure(e.to_string()),
@@ -8176,68 +8177,97 @@ impl SipSession {
81768177
let invite_handle = crate::utils::spawn(async move {
81778178
let leg_id = leg_id_for_spawn;
81788179
let (state_tx, mut state_rx) = tokio::sync::mpsc::unbounded_channel();
8179-
let invitation = dialog_layer.do_invite(invite_option, state_tx).boxed();
8180+
let mut invitation = dialog_layer.do_invite(invite_option, state_tx).boxed();
81808181

8181-
let result = match invitation.await {
8182-
Ok((dialog, response)) => {
8183-
if let Some(ref resp) = response {
8184-
let status_code = resp.status_code.code();
8185-
if StatusCode::from(status_code).kind()
8186-
== rsipstack::sip::StatusCodeKind::Successful
8187-
{
8188-
info!(%leg_id, status = %status_code, "SIP leg answered successfully");
8182+
let mut result: Result<rsipstack::dialog::client_dialog::ClientInviteDialog, String> = Err("not started".to_string());
8183+
let mut state_rx_open = true;
8184+
8185+
loop {
8186+
tokio::select! {
8187+
biased;
8188+
r = &mut invitation, if !result.is_ok() => {
8189+
match r {
8190+
Ok((dialog, response)) => {
8191+
if let Some(ref resp) = response {
8192+
let status_code = resp.status_code.code();
8193+
if StatusCode::from(status_code).kind()
8194+
== rsipstack::sip::StatusCodeKind::Successful
8195+
{
8196+
info!(%leg_id, status = %status_code, "SIP leg answered successfully");
8197+
8198+
let answer_sdp = if !resp.body().is_empty() {
8199+
let sdp = String::from_utf8_lossy(resp.body()).to_string();
8200+
if let Err(e) =
8201+
peer.update_remote_description(&track_id, &sdp).await
8202+
{
8203+
warn!(%leg_id, error = %e, "Failed to set remote description on leg peer");
8204+
} else {
8205+
info!(%leg_id, "Remote description set successfully");
8206+
}
8207+
Some(sdp)
8208+
} else {
8209+
None
8210+
};
81898211

8190-
// Extract SDP answer from response body
8191-
let answer_sdp = if !resp.body().is_empty() {
8192-
let sdp = String::from_utf8_lossy(resp.body()).to_string();
8212+
let _ = cmd_tx.send(CallCommand::LegConnected {
8213+
leg_id: leg_id.clone(),
8214+
answer_sdp,
8215+
}).await;
81938216

8194-
// Set remote description on the peer
8195-
if let Err(e) =
8196-
peer.update_remote_description(&track_id, &sdp).await
8197-
{
8198-
warn!(%leg_id, error = %e, "Failed to set remote description on leg peer");
8217+
result = Ok(dialog);
8218+
break;
8219+
} else {
8220+
warn!(%leg_id, status = %status_code, "SIP leg rejected");
8221+
let _ = cmd_tx.send(CallCommand::LegFailed {
8222+
leg_id: leg_id.clone(),
8223+
reason: format!("Rejected with {}", status_code),
8224+
}).await;
8225+
result = Err(format!("Rejected with {}", status_code));
8226+
break;
8227+
}
81998228
} else {
8200-
info!(%leg_id, "Remote description set successfully");
8229+
warn!(%leg_id, "SIP leg timeout (no response)");
8230+
let _ = cmd_tx.send(CallCommand::LegFailed {
8231+
leg_id: leg_id.clone(),
8232+
reason: "Timeout".to_string(),
8233+
}).await;
8234+
result = Err("Timeout".to_string());
8235+
break;
82018236
}
8202-
Some(sdp)
8203-
} else {
8204-
None
8205-
};
8206-
8207-
// Send LegConnected with SDP answer
8208-
let _ = cmd_tx.send(CallCommand::LegConnected {
8209-
leg_id: leg_id.clone(),
8210-
answer_sdp,
8211-
});
8212-
8213-
// Return dialog for further processing
8214-
Ok(dialog)
8215-
} else {
8216-
warn!(%leg_id, status = %status_code, "SIP leg rejected");
8217-
let _ = cmd_tx.send(CallCommand::LegFailed {
8218-
leg_id: leg_id.clone(),
8219-
reason: format!("Rejected with {}", status_code),
8220-
});
8221-
Err(format!("Rejected with {}", status_code))
8237+
}
8238+
Err(e) => {
8239+
warn!(%leg_id, error = %e, "SIP leg failed");
8240+
let _ = cmd_tx.send(CallCommand::LegFailed {
8241+
leg_id: leg_id.clone(),
8242+
reason: e.to_string(),
8243+
}).await;
8244+
result = Err(e.to_string());
8245+
break;
8246+
}
8247+
}
8248+
}
8249+
state = state_rx.recv(), if state_rx_open => {
8250+
match state {
8251+
Some(rsipstack::dialog::dialog::DialogState::Early(_, ref resp)) => {
8252+
info!(%leg_id, "SIP leg early media (183)");
8253+
let body = resp.body();
8254+
if !body.is_empty() {
8255+
let sdp = String::from_utf8_lossy(body).to_string();
8256+
if let Err(e) =
8257+
peer.update_remote_description(&track_id, &sdp).await
8258+
{
8259+
warn!(%leg_id, error = %e, "Failed to set early media remote description");
8260+
} else {
8261+
info!(%leg_id, "Early media remote description set");
8262+
}
8263+
}
8264+
}
8265+
Some(_) => {}
8266+
None => { state_rx_open = false; }
82228267
}
8223-
} else {
8224-
warn!(%leg_id, "SIP leg timeout (no response)");
8225-
let _ = cmd_tx.send(CallCommand::LegFailed {
8226-
leg_id: leg_id.clone(),
8227-
reason: "Timeout".to_string(),
8228-
});
8229-
Err("Timeout".to_string())
82308268
}
82318269
}
8232-
Err(e) => {
8233-
warn!(%leg_id, error = %e, "SIP leg failed");
8234-
let _ = cmd_tx.send(CallCommand::LegFailed {
8235-
leg_id: leg_id.clone(),
8236-
reason: e.to_string(),
8237-
});
8238-
Err(e.to_string())
8239-
}
8240-
};
8270+
}
82418271

82428272
// Process dialog state changes (e.g., BYE from remote)
82438273
if let Ok(dialog) = result {
@@ -8257,7 +8287,7 @@ impl SipSession {
82578287
let _ = cmd_tx.send(CallCommand::LegFailed {
82588288
leg_id: leg_id.clone(),
82598289
reason: "Remote hung up".to_string(),
8260-
});
8290+
}).await;
82618291
break;
82628292
}
82638293
Some(_) => {}
@@ -8266,7 +8296,6 @@ impl SipSession {
82668296
}
82678297
}
82688298
}
8269-
// Keep dialog alive
82708299
let _ = dialog;
82718300
});
82728301
}
@@ -8338,13 +8367,6 @@ impl SipSession {
83388367
async fn stop_all_bridges(&mut self) {
83398368
self.stop_conference_bridges().await;
83408369
self.stop_direct_bridge().await;
8341-
// Abort all leg-specific spawned tasks
8342-
for (leg_id, handles) in self.legs.drain_tasks() {
8343-
for handle in handles {
8344-
handle.abort();
8345-
}
8346-
info!(%leg_id, "Aborted tasks for leg");
8347-
}
83488370
info!("All bridges stopped");
83498371
}
83508372

‎src/proxy/proxy_call/sip_session/conference.rs‎

Lines changed: 32 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -57,8 +57,28 @@ impl SipSession {
5757
info!(%leg_id, "Added participant to conference");
5858

5959
if let Some(peer) = peer {
60+
let fallback_pc = if leg_id.0 == "caller" {
61+
self.media.media_bridge.as_ref().map(|b| b.caller_pc().clone())
62+
} else if leg_id.0 == "callee" {
63+
self.media.media_bridge.as_ref().map(|b| b.callee_pc().clone())
64+
} else {
65+
None
66+
};
67+
68+
let fallback_sender = if leg_id.0 == "caller" {
69+
if let Some(b) = self.media.media_bridge.as_ref() {
70+
b.get_caller_sender().await
71+
} else { None }
72+
} else if leg_id.0 == "callee" {
73+
if let Some(b) = self.media.media_bridge.as_ref() {
74+
b.get_callee_sender().await
75+
} else { None }
76+
} else {
77+
None
78+
};
79+
6080
if let Err(e) = self
61-
.start_conference_media_bridge_for_peer(&conf_id_str, &leg_id, &peer)
81+
.start_conference_media_bridge_for_peer(&conf_id_str, &leg_id, &peer, fallback_pc, fallback_sender)
6282
.await
6383
{
6484
warn!(%leg_id, "Failed to start conference media bridge for dynamic leg: {}", e);
@@ -88,6 +108,8 @@ impl SipSession {
88108
conf_id: &str,
89109
leg_id: &LegId,
90110
peer: &Arc<dyn MediaPeer>,
111+
fallback_pc: Option<rustrtc::PeerConnection>,
112+
fallback_sender: Option<rustrtc::media::SampleStreamSource>,
91113
) -> Result<crate::call::runtime::ConferenceBridgeHandle> {
92114
use rustrtc::media::MediaSample;
93115

@@ -101,6 +123,10 @@ impl SipSession {
101123
}
102124
}
103125

126+
if audio_sender.is_none() {
127+
audio_sender = fallback_sender;
128+
}
129+
104130
let (tx, mut rx) = tokio::sync::mpsc::channel::<MediaSample>(100);
105131

106132
if let Some(sender) = audio_sender {
@@ -147,7 +173,7 @@ impl SipSession {
147173
}
148174

149175
let audio_receiver = self
150-
.create_audio_receiver_from_peer(peer)
176+
.create_audio_receiver_from_peer(peer, fallback_pc)
151177
.await
152178
.map_err(|e| anyhow!("Failed to create audio receiver for dynamic leg: {}", e))?;
153179

@@ -283,7 +309,7 @@ impl SipSession {
283309
.push(forwarder_handle);
284310

285311
let audio_receiver = if is_callee {
286-
self.create_audio_receiver_from_peer(&peer).await
312+
self.create_audio_receiver_from_peer(&peer, None).await
287313
} else {
288314
self.create_audio_receiver().await
289315
}
@@ -325,6 +351,7 @@ impl SipSession {
325351
pub(super) async fn create_audio_receiver_from_peer(
326352
&self,
327353
peer: &Arc<dyn MediaPeer>,
354+
fallback_pc: Option<rustrtc::PeerConnection>,
328355
) -> Result<Box<dyn crate::call::runtime::conference_media_bridge::AudioReceiver>> {
329356
let mut pc = None;
330357
for _ in 0..150 {
@@ -342,7 +369,7 @@ impl SipSession {
342369
tokio::time::sleep(Duration::from_millis(20)).await;
343370
}
344371

345-
let pc = pc.ok_or_else(|| anyhow!("No peer connection found for conference input"))?;
372+
let pc = pc.or(fallback_pc).ok_or_else(|| anyhow!("No peer connection found for conference input"))?;
346373

347374
let decoder = self
348375
.create_audio_decoder()
@@ -451,7 +478,7 @@ impl SipSession {
451478

452479
let peer = self.legs.peers.get(&leg_id).cloned();
453480
let bridge_result = if let Some(peer) = peer {
454-
self.start_conference_media_bridge_for_peer(&conf_id, &leg_id, &peer)
481+
self.start_conference_media_bridge_for_peer(&conf_id, &leg_id, &peer, None, None)
455482
.await
456483
} else {
457484
self.start_conference_media_bridge(&conf_id, &leg_id)

0 commit comments

Comments
 (0)