use crate::call::active_call::ActiveCallStateRef;
use crate::callrecord::CallRecordHangupReason;
use crate::event::EventSender;
use crate::media::TrackId;
use crate::media::stream::MediaStream;
use crate::useragent::invitation::PendingDialog;
use anyhow::Result;
use chrono::Utc;
use rsipstack::dialog::DialogId;
use rsipstack::dialog::dialog::{
Dialog, DialogState, DialogStateReceiver, DialogStateSender, TerminatedReason,
};
use rsipstack::dialog::dialog_layer::DialogLayer;
use rsipstack::dialog::invitation::InviteOption;
use std::collections::HashMap;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};
pub struct DialogStateReceiverGuard {
pub(super) dialog_layer: Arc<DialogLayer>,
pub(super) receiver: DialogStateReceiver,
pub(super) dialog_id: Option<DialogId>,
pub(super) hangup_headers: Option<Vec<rsipstack::rsip::Header>>,
}
impl DialogStateReceiverGuard {
pub fn new(
dialog_layer: Arc<DialogLayer>,
receiver: DialogStateReceiver,
hangup_headers: Option<Vec<rsipstack::rsip::Header>>,
) -> Self {
Self {
dialog_layer,
receiver,
dialog_id: None,
hangup_headers,
}
}
pub async fn recv(&mut self) -> Option<DialogState> {
let state = self.receiver.recv().await;
if let Some(ref s) = state {
self.dialog_id = Some(s.id().clone());
}
state
}
fn take_dialog(&mut self) -> Option<Dialog> {
let id = match self.dialog_id.take() {
Some(id) => id,
None => return None,
};
match self.dialog_layer.get_dialog(&id) {
Some(dialog) => {
info!(%id, "dialog removed on drop");
self.dialog_layer.remove_dialog(&id);
return Some(dialog);
}
_ => {}
}
None
}
pub async fn drop_async(&mut self) {
if let Some(dialog) = self.take_dialog() {
if let Err(e) = dialog.hangup_with_headers(self.hangup_headers.take()).await {
warn!(id=%dialog.id(), "error hanging up dialog on drop: {}", e);
}
}
}
}
impl Drop for DialogStateReceiverGuard {
fn drop(&mut self) {
if let Some(dialog) = self.take_dialog() {
crate::spawn(async move {
if let Err(e) = dialog.hangup().await {
warn!(id=%dialog.id(), "error hanging up dialog on drop: {}", e);
}
});
}
}
}
pub(super) struct InviteDialogStates {
pub is_client: bool,
pub session_id: String,
pub track_id: TrackId,
pub cancel_token: CancellationToken,
pub event_sender: EventSender,
pub call_state: ActiveCallStateRef,
pub media_stream: Arc<MediaStream>,
pub terminated_reason: Option<TerminatedReason>,
pub has_early_media: bool,
}
impl InviteDialogStates {
pub(super) fn on_terminated(&mut self) {
let mut call_state_ref = match self.call_state.try_write() {
Ok(cs) => cs,
Err(_) => {
return;
}
};
let reason = &self.terminated_reason;
call_state_ref.last_status_code = match reason {
Some(TerminatedReason::UacCancel) => 487,
Some(TerminatedReason::UacBye) => 200,
Some(TerminatedReason::UacBusy) => 486,
Some(TerminatedReason::UasBye) => 200,
Some(TerminatedReason::UasBusy) => 486,
Some(TerminatedReason::UasDecline) => 603,
Some(TerminatedReason::UacOther(code)) => code.code(),
Some(TerminatedReason::UasOther(code)) => code.code(),
_ => 500, };
if call_state_ref.hangup_reason.is_none() {
call_state_ref.hangup_reason.replace(match reason {
Some(TerminatedReason::UacCancel) => CallRecordHangupReason::Canceled,
Some(TerminatedReason::UacBye) | Some(TerminatedReason::UacBusy) => {
CallRecordHangupReason::ByCaller
}
Some(TerminatedReason::UasBye) | Some(TerminatedReason::UasBusy) => {
CallRecordHangupReason::ByCallee
}
Some(TerminatedReason::UasDecline) => CallRecordHangupReason::ByCallee,
Some(TerminatedReason::UacOther(_)) => CallRecordHangupReason::ByCaller,
Some(TerminatedReason::UasOther(_)) => CallRecordHangupReason::ByCallee,
_ => CallRecordHangupReason::BySystem,
});
};
let initiator = match reason {
Some(TerminatedReason::UacCancel) => "caller".to_string(),
Some(TerminatedReason::UacBye) | Some(TerminatedReason::UacBusy) => {
"caller".to_string()
}
Some(TerminatedReason::UasBye)
| Some(TerminatedReason::UasBusy)
| Some(TerminatedReason::UasDecline) => "callee".to_string(),
_ => "system".to_string(),
};
self.event_sender
.send(crate::event::SessionEvent::TrackEnd {
track_id: self.track_id.clone(),
timestamp: crate::media::get_timestamp(),
duration: call_state_ref
.answer_time
.map(|t| (Utc::now() - t).num_milliseconds())
.unwrap_or_default() as u64,
ssrc: call_state_ref.ssrc,
play_id: None,
})
.ok();
let hangup_event =
call_state_ref.build_hangup_event(self.track_id.clone(), Some(initiator));
self.event_sender.send(hangup_event).ok();
}
}
impl Drop for InviteDialogStates {
fn drop(&mut self) {
self.on_terminated();
self.cancel_token.cancel();
}
}
impl DialogStateReceiverGuard {
pub(self) async fn dialog_event_loop(&mut self, states: &mut InviteDialogStates) -> Result<()> {
while let Some(event) = self.recv().await {
match event {
DialogState::Calling(dialog_id) => {
info!(session_id=states.session_id, %dialog_id, "dialog calling");
states.call_state.write().await.session_id = dialog_id.to_string();
}
DialogState::Trying(_) => {}
DialogState::Early(dialog_id, resp) => {
let code = resp.status_code.code();
let body = resp.body();
let answer = String::from_utf8_lossy(body);
let has_sdp = !answer.is_empty();
info!(session_id=states.session_id, %dialog_id, has_sdp=%has_sdp, "dialog early ({}): \n{}", code, answer);
{
let mut cs = states.call_state.write().await;
if cs.ring_time.is_none() {
cs.ring_time.replace(Utc::now());
}
cs.last_status_code = code;
}
if !states.is_client {
continue;
}
let refer = states.call_state.read().await.is_refer;
states
.event_sender
.send(crate::event::SessionEvent::Ringing {
track_id: states.track_id.clone(),
timestamp: crate::media::get_timestamp(),
early_media: has_sdp,
refer: Some(refer),
})?;
if has_sdp {
states.has_early_media = true;
{
let mut cs = states.call_state.write().await;
if cs.answer.is_none() {
cs.answer = Some(answer.to_string());
}
}
states
.media_stream
.update_remote_description(&states.track_id, &answer.to_string())
.await?;
}
}
DialogState::Confirmed(dialog_id, msg) => {
info!(session_id=states.session_id, %dialog_id, has_early_media=%states.has_early_media, "dialog confirmed");
{
let mut cs = states.call_state.write().await;
cs.session_id = dialog_id.to_string();
cs.answer_time.replace(Utc::now());
cs.last_status_code = 200;
}
if states.is_client {
let answer = String::from_utf8_lossy(msg.body());
let answer = answer.trim();
if !answer.is_empty() {
if states.has_early_media {
info!(
session_id = states.session_id,
"updating remote description with final answer after early media (force=true)"
);
if let Err(e) = states
.media_stream
.update_remote_description_force(
&states.track_id,
&answer.to_string(),
)
.await
{
tracing::warn!(
session_id = states.session_id,
"failed to force update remote description on confirmed: {}",
e
);
}
} else {
if let Err(e) = states
.media_stream
.update_remote_description(
&states.track_id,
&answer.to_string(),
)
.await
{
tracing::warn!(
session_id = states.session_id,
"failed to update remote description on confirmed: {}",
e
);
}
}
}
}
}
DialogState::Info(dialog_id, req, tx_handle) => {
let body_str = String::from_utf8_lossy(req.body());
info!(session_id=states.session_id, %dialog_id, body=%body_str, "dialog info received");
if body_str.starts_with("Signal=") {
let digit = body_str.trim_start_matches("Signal=").chars().next();
if let Some(digit) = digit {
let is_refer = states.call_state.read().await.is_refer;
states.event_sender.send(crate::event::SessionEvent::Dtmf {
track_id: states.track_id.clone(),
timestamp: crate::media::get_timestamp(),
digit: digit.to_string(),
refer: Some(is_refer),
})?;
}
}
tx_handle.reply(rsipstack::rsip::StatusCode::OK).await.ok();
}
DialogState::Message(dialog_id, req, tx_handle) => {
let body_str = String::from_utf8_lossy(req.body()).to_string();
let content_type = req.headers.iter().find_map(|h| {
if let rsipstack::rsip::Header::ContentType(content_type) = h {
Some(content_type.value().to_string())
} else {
None
}
});
info!(
session_id=states.session_id,
%dialog_id,
content_type=content_type.as_deref(),
body=%body_str,
"dialog message received"
);
let is_refer = states.call_state.read().await.is_refer;
states
.event_sender
.send(crate::event::SessionEvent::Message {
track_id: states.track_id.clone(),
timestamp: crate::media::get_timestamp(),
body: body_str,
content_type,
refer: Some(is_refer),
})
.ok();
tx_handle.reply(rsipstack::rsip::StatusCode::OK).await.ok();
}
DialogState::Updated(dialog_id, _req, tx_handle) => {
info!(session_id = states.session_id, %dialog_id, "dialog update received");
let is_refer = states.call_state.read().await.is_refer;
let mut answer_sdp = None;
if let Some(sdp_body) = _req.body().get(..) {
let sdp_str = String::from_utf8_lossy(sdp_body);
if !sdp_str.is_empty()
&& (_req.method == rsipstack::rsip::Method::Invite
|| _req.method == rsipstack::rsip::Method::Update)
{
info!(session_id=states.session_id, %dialog_id, method=%_req.method, "handling re-invite/update offer");
let is_on_hold =
crate::media::negotiate::detect_hold_state_from_sdp(&sdp_str);
info!(session_id=states.session_id, %dialog_id, is_on_hold=%is_on_hold, "detected hold state from re-invite SDP");
if is_on_hold {
states
.media_stream
.hold_track(Some(states.track_id.clone()))
.await;
} else {
states
.media_stream
.resume_track(Some(states.track_id.clone()))
.await;
}
states
.event_sender
.send(crate::event::SessionEvent::Hold {
track_id: states.track_id.clone(),
timestamp: crate::media::get_timestamp(),
on_hold: is_on_hold,
refer: Some(is_refer),
})
.ok();
match states
.media_stream
.handshake(&states.track_id, sdp_str.to_string(), None)
.await
{
Ok(sdp) => answer_sdp = Some(sdp),
Err(e) => {
warn!(
session_id = states.session_id,
"failed to handle re-invite: {}", e
);
}
}
} else {
info!(session_id=states.session_id, %dialog_id, "updating remote description:\n{}", sdp_str);
let is_on_hold =
crate::media::negotiate::detect_hold_state_from_sdp(&sdp_str);
if is_on_hold {
states
.media_stream
.hold_track(Some(states.track_id.clone()))
.await;
states
.event_sender
.send(crate::event::SessionEvent::Hold {
track_id: states.track_id.clone(),
timestamp: crate::media::get_timestamp(),
on_hold: true,
refer: Some(is_refer),
})
.ok();
} else {
states
.media_stream
.resume_track(Some(states.track_id.clone()))
.await;
states
.event_sender
.send(crate::event::SessionEvent::Hold {
track_id: states.track_id.clone(),
timestamp: crate::media::get_timestamp(),
on_hold: false,
refer: Some(is_refer),
})
.ok();
}
states
.media_stream
.update_remote_description(&states.track_id, &sdp_str.to_string())
.await?;
}
}
if let Some(sdp) = answer_sdp {
tx_handle
.respond(
rsipstack::rsip::StatusCode::OK,
Some(vec![rsipstack::rsip::Header::ContentType(
"application/sdp".to_string().into(),
)]),
Some(sdp.into()),
)
.await
.ok();
} else {
tx_handle.reply(rsipstack::rsip::StatusCode::OK).await.ok();
}
}
DialogState::Options(dialog_id, _req, tx_handle) => {
info!(session_id = states.session_id, %dialog_id, "dialog options received");
tx_handle.reply(rsipstack::rsip::StatusCode::OK).await.ok();
}
DialogState::Refer(dialog_id, req, tx_handle) => {
let refer_to = req.headers.iter().find_map(|h| {
if let rsipstack::rsip::Header::ReferTo(h) = h {
return Some(h.value().to_string());
}
None
}).unwrap_or_default();
let referred_by = req.headers.iter().find_map(|h| {
if let rsipstack::rsip::Header::ReferredBy(h) = h {
return Some(h.value().to_string());
}
None
});
info!(session_id = states.session_id, %dialog_id, %refer_to, "received REFER");
tx_handle.reply(rsipstack::rsip::StatusCode::Other(202, "Accepted".into())).await.ok();
let is_refer = states.call_state.read().await.is_refer;
states.event_sender.send(crate::event::SessionEvent::TransferRequest {
track_id: states.track_id.clone(),
timestamp: crate::media::get_timestamp(),
refer_to,
referred_by,
refer: Some(is_refer),
}).ok();
}
DialogState::Terminated(dialog_id, reason) => {
info!(
session_id = states.session_id,
?dialog_id,
?reason,
"dialog terminated"
);
states.terminated_reason = Some(reason.clone());
return Ok(());
}
other_state => {
info!(
session_id = states.session_id,
%other_state,
"dialog received other state"
);
}
}
}
Ok(())
}
pub(super) async fn process_dialog(&mut self, mut states: InviteDialogStates) {
let token = states.cancel_token.clone();
tokio::select! {
_ = token.cancelled() => {
states.terminated_reason = Some(TerminatedReason::UacCancel);
}
_ = self.dialog_event_loop(&mut states) => {}
};
{
let state = states.call_state.read().await;
if let Some(extras) = &state.extras {
if let Some(h_val) = extras.get("_hangup_headers") {
if let Ok(headers_map) =
serde_json::from_value::<HashMap<String, String>>(h_val.clone())
{
let mut headers = Vec::new();
for (k, v) in headers_map {
headers.push(rsipstack::rsip::Header::Other(k.into(), v.into()));
}
if !headers.is_empty() {
if let Some(existing) = &mut self.hangup_headers {
existing.extend(headers);
} else {
self.hangup_headers = Some(headers);
}
}
}
}
}
}
self.drop_async().await;
}
}
#[derive(Clone)]
pub struct Invitation {
pub dialog_layer: Arc<DialogLayer>,
pub pending_dialogs: Arc<std::sync::Mutex<HashMap<DialogId, PendingDialog>>>,
}
impl Invitation {
pub fn new(dialog_layer: Arc<DialogLayer>) -> Self {
Self {
dialog_layer,
pending_dialogs: Arc::new(std::sync::Mutex::new(HashMap::new())),
}
}
pub fn add_pending(&self, dialog_id: DialogId, pending: PendingDialog) {
self.pending_dialogs
.lock()
.map(|mut ps| ps.insert(dialog_id, pending))
.ok();
}
pub fn get_pending_call(&self, dialog_id: &DialogId) -> Option<PendingDialog> {
self.pending_dialogs
.lock()
.ok()
.and_then(|mut ps| ps.remove(dialog_id))
}
pub fn has_pending_call(&self, dialog_id: &DialogId) -> bool {
self.pending_dialogs
.lock()
.ok()
.map(|ps| ps.contains_key(dialog_id))
.unwrap_or(false)
}
pub fn find_dialog_id_by_session_id(&self, session_id: &str) -> Option<DialogId> {
self.pending_dialogs.lock().ok().and_then(|ps| {
ps.iter()
.find(|(id, _)| id.to_string() == session_id)
.map(|(id, _)| id.clone())
})
}
pub async fn hangup(
&self,
dialog_id: DialogId,
code: Option<rsipstack::rsip::StatusCode>,
reason: Option<String>,
) -> Result<()> {
if let Some(call) = self.get_pending_call(&dialog_id) {
call.dialog.reject(code, reason).ok();
}
match self.dialog_layer.get_dialog(&dialog_id) {
Some(dialog) => {
self.dialog_layer.remove_dialog(&dialog_id);
dialog.hangup().await.ok();
}
None => {}
}
Ok(())
}
pub async fn reject(&self, dialog_id: DialogId) -> Result<()> {
if let Some(call) = self.get_pending_call(&dialog_id) {
call.dialog.reject(None, None).ok();
}
match self.dialog_layer.get_dialog(&dialog_id) {
Some(dialog) => {
self.dialog_layer.remove_dialog(&dialog_id);
dialog.hangup().await.ok();
}
None => {}
}
Ok(())
}
pub async fn invite(
&self,
invite_option: InviteOption,
state_sender: DialogStateSender,
) -> Result<(DialogId, Option<Vec<u8>>), rsipstack::Error> {
let (dialog, resp) = self
.dialog_layer
.do_invite(invite_option, state_sender)
.await?;
let offer = match resp {
Some(resp) => match resp.status_code.kind() {
rsipstack::rsip::StatusCodeKind::Successful => {
let offer = resp.body.clone();
Some(offer)
}
_ => {
let reason = resp
.reason_phrase()
.unwrap_or(&resp.status_code.to_string())
.to_string();
return Err(rsipstack::Error::DialogError(
reason,
dialog.id(),
resp.status_code,
));
}
},
None => {
return Err(rsipstack::Error::DialogError(
"no response received".to_string(),
dialog.id(),
rsipstack::rsip::StatusCode::NotAcceptableHere,
));
}
};
Ok((dialog.id(), offer))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::call::active_call::ActiveCallState;
use crate::media::stream::MediaStreamBuilder;
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio_util::sync::CancellationToken;
const EARLY_MEDIA_SDP: &str = "v=0\r\n\
o=- 1000 1 IN IP4 192.168.1.100\r\n\
s=SIP Call\r\n\
t=0 0\r\n\
m=audio 10000 RTP/AVP 0\r\n\
c=IN IP4 192.168.1.100\r\n\
a=rtpmap:0 PCMU/8000\r\n\
a=sendrecv\r\n";
fn make_response_with_body(body: Vec<u8>) -> rsipstack::rsip::Response {
let mut resp = rsipstack::rsip::Response::default();
resp.body = body;
resp
}
#[tokio::test]
async fn test_early_sdp_stored_in_call_state() {
let (event_tx, _event_rx) = tokio::sync::broadcast::channel(16);
let media_stream = Arc::new(
MediaStreamBuilder::new(event_tx.clone())
.with_id("test-stream".to_string())
.build(),
);
let call_state: ActiveCallStateRef = Arc::new(RwLock::new(ActiveCallState::default()));
let cancel_token = CancellationToken::new();
let mut states = InviteDialogStates {
is_client: true,
session_id: "test-session".to_string(),
track_id: "test-track".to_string(),
cancel_token: cancel_token.clone(),
event_sender: event_tx.clone(),
call_state: call_state.clone(),
media_stream: media_stream.clone(),
terminated_reason: None,
has_early_media: false,
};
let early_resp = make_response_with_body(EARLY_MEDIA_SDP.as_bytes().to_vec());
let body = early_resp.body();
let answer = String::from_utf8_lossy(body);
let has_sdp = !answer.is_empty();
if states.is_client && has_sdp {
states.has_early_media = true;
{
let mut cs = states.call_state.write().await;
if cs.answer.is_none() {
cs.answer = Some(answer.to_string());
}
}
}
{
let cs = call_state.read().await;
assert!(
cs.answer.is_some(),
"call_state.answer should be set after 183 with SDP"
);
assert_eq!(
cs.answer.as_deref().unwrap(),
EARLY_MEDIA_SDP,
"call_state.answer should contain the early SDP"
);
}
assert!(states.has_early_media, "has_early_media should be true");
}
#[tokio::test]
async fn test_confirmed_empty_body_keeps_early_sdp() {
let (event_tx, _event_rx) = tokio::sync::broadcast::channel(16);
let media_stream = Arc::new(
MediaStreamBuilder::new(event_tx.clone())
.with_id("test-stream-2".to_string())
.build(),
);
let call_state: ActiveCallStateRef = Arc::new(RwLock::new(ActiveCallState::default()));
let cancel_token = CancellationToken::new();
let mut states = InviteDialogStates {
is_client: true,
session_id: "test-session-2".to_string(),
track_id: "test-track-2".to_string(),
cancel_token: cancel_token.clone(),
event_sender: event_tx.clone(),
call_state: call_state.clone(),
media_stream: media_stream.clone(),
terminated_reason: None,
has_early_media: false,
};
{
let answer_str = EARLY_MEDIA_SDP.to_string();
states.has_early_media = true;
let mut cs = states.call_state.write().await;
if cs.answer.is_none() {
cs.answer = Some(answer_str);
}
}
let confirmed_resp = make_response_with_body(vec![]); {
let mut cs = states.call_state.write().await;
cs.answer_time.replace(chrono::Utc::now());
cs.last_status_code = 200;
}
let body = confirmed_resp.body();
let answer = String::from_utf8_lossy(body);
let answer_trimmed = answer.trim();
if states.is_client && !answer_trimmed.is_empty() {
panic!("Confirmed handler should not update SDP for empty body");
}
{
let cs = call_state.read().await;
assert!(
cs.answer.is_some(),
"call_state.answer must not be None after 200 OK with empty body"
);
let stored_answer = cs.answer.as_deref().unwrap();
assert!(
!stored_answer.is_empty(),
"call_state.answer must not be empty after 200 OK with empty body"
);
assert_eq!(
stored_answer, EARLY_MEDIA_SDP,
"call_state.answer should still be the early SDP after 200 OK with empty body"
);
}
}
#[tokio::test]
async fn test_answer_fallback_to_early_sdp_when_200ok_empty() {
let call_state: ActiveCallStateRef = Arc::new(RwLock::new(ActiveCallState::default()));
{
let mut cs = call_state.write().await;
cs.answer = Some(EARLY_MEDIA_SDP.to_string());
}
let raw_answer: Option<Vec<u8>> = Some(vec![]);
let resolved_answer = match raw_answer {
Some(bytes) => {
let s = String::from_utf8_lossy(&bytes).to_string();
if s.trim().is_empty() {
let cs = call_state.read().await;
match cs.answer.clone() {
Some(early_sdp) if !early_sdp.is_empty() => {
(early_sdp, true )
}
_ => (s, false),
}
} else {
(s, false)
}
}
None => {
let cs = call_state.read().await;
match cs.answer.clone() {
Some(early_sdp) if !early_sdp.is_empty() => (early_sdp, true),
_ => panic!("Expected early SDP fallback"),
}
}
};
let (answer, already_applied) = resolved_answer;
assert!(
!answer.is_empty(),
"Resolved answer must not be empty — should contain the early SDP"
);
assert_eq!(
answer, EARLY_MEDIA_SDP,
"Resolved answer should be the early SDP from the 183 handler"
);
assert!(
already_applied,
"remote_description_already_applied should be true when using early SDP fallback"
);
}
#[tokio::test]
async fn test_answer_uses_200ok_sdp_when_present() {
const FINAL_SDP: &str = "v=0\r\n\
o=- 2000 2 IN IP4 10.0.0.1\r\n\
s=SIP Call\r\n\
t=0 0\r\n\
m=audio 20000 RTP/AVP 0\r\n\
c=IN IP4 10.0.0.1\r\n\
a=rtpmap:0 PCMU/8000\r\n\
a=sendrecv\r\n";
let call_state: ActiveCallStateRef = Arc::new(RwLock::new(ActiveCallState::default()));
{
let mut cs = call_state.write().await;
cs.answer = Some(EARLY_MEDIA_SDP.to_string());
}
let raw_answer: Option<Vec<u8>> = Some(FINAL_SDP.as_bytes().to_vec());
let resolved_answer = match raw_answer {
Some(bytes) => {
let s = String::from_utf8_lossy(&bytes).to_string();
if s.trim().is_empty() {
let cs = call_state.read().await;
match cs.answer.clone() {
Some(early_sdp) if !early_sdp.is_empty() => (early_sdp, true),
_ => (s, false),
}
} else {
(s, false) }
}
None => panic!("Unexpected"),
};
let (answer, already_applied) = resolved_answer;
assert_eq!(
answer, FINAL_SDP,
"When 200 OK has SDP, it should be used (not the early SDP)"
);
assert!(
!already_applied,
"remote_description_already_applied should be false when 200 OK has SDP body"
);
}
}