serve.zone/siprouter
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mixer): enhance mixer functionality with interaction and tool legs

Browse Source 
- Updated mixer to handle participant and isolated leg roles, allowing for IVR and consent interactions.
- Introduced commands for starting and canceling interactions, managing tool legs for recording and transcription.
- Implemented per-source audio handling for tool legs, enabling separate audio processing.
- Enhanced DTMF handling to forward events between participant legs only.
- Added support for PCM recording directly from tool legs, with WAV file generation.
- Updated TypeScript definitions and functions to support new interaction and tool leg features.
This commit is contained in:
Jürgen Kunz
2026-04-10 14:54:21 +00:00
parent 6a130db7c7
commit 7d59361352
13 changed files with 1448 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
rust/crates/proxy-engine/src/call_manager.rs
View File

@@ -12,8 +12,8 @@ use crate::mixer::spawn_mixer;
use crate::registrar::Registrar;
use crate::rtp::RtpPortPool;
use crate::sip_leg::{SipLeg, SipLegAction, SipLegConfig};
use sip_proto::helpers::{generate_call_id, parse_sdp_endpoint};
use sip_proto::message::SipMessage;
use sip_proto::helpers::{build_sdp, generate_call_id, generate_tag, parse_sdp_endpoint, SdpOptions};
use sip_proto::message::{ResponseOptions, SipMessage};
use sip_proto::rewrite::{rewrite_sdp, rewrite_sip_uri};
use std::collections::HashMap;
use std::net::SocketAddr;
@@ -167,6 +167,16 @@ impl CallManager {
if let Some(leg) = call.legs.get_mut(leg_id) {
leg.state = LegState::Ringing;
}
// Forward 180 Ringing to device if this is a device-originated call.
if let Some(device_invite) = &call.device_invite {
let device_leg = call.legs.values().find(|l| l.kind == LegKind::SipDevice);
if let Some(dev) = device_leg {
if let Some(dev_addr) = dev.signaling_addr {
let ringing = SipMessage::create_response(180, "Ringing", device_invite, None);
let _ = socket.send_to(&ringing.serialize(), dev_addr).await;
}
}
}
}
emit_event(&self.out_tx, "call_ringing", serde_json::json!({ "call_id": call_id }));
emit_event(&self.out_tx, "leg_state_changed",
@@ -187,7 +197,7 @@ impl CallManager {
remote
};
// Wire the leg to the mixer if remote media is known.
// Wire the provider leg to the mixer if remote media is known.
if let (Some(remote_addr), Some(rtp_socket)) = (remote, rtp_socket_clone) {
let channels = create_leg_channels();
spawn_sip_inbound(rtp_socket.clone(), channels.inbound_tx);
@@ -198,6 +208,66 @@ impl CallManager {
}
}
// For device-originated calls: send 200 OK to device and wire device leg.
if let Some(call) = self.calls.get(call_id) {
if let Some(device_invite) = call.device_invite.clone() {
let device_leg_info: Option<(SocketAddr, u16, Arc<UdpSocket>, Option<SocketAddr>, String)> =
call.legs.values().find(|l| l.kind == LegKind::SipDevice).and_then(|dev| {
Some((
dev.signaling_addr?,
dev.rtp_port,
dev.rtp_socket.clone()?,
dev.remote_media,
dev.id.clone(),
))
});
if let Some((dev_addr, dev_rtp_port, dev_rtp_socket, dev_remote, dev_leg_id)) = device_leg_info {
// Build SDP pointing device to our device_rtp port.
// Use LAN IP for the device (it's on the local network).
let call_ref = self.calls.get(call_id).unwrap();
let prov_leg = call_ref.legs.values().find(|l| l.kind == LegKind::SipProvider);
let lan_ip_str = prov_leg
.and_then(|l| l.sip_leg.as_ref())
.map(|sl| sl.config.lan_ip.clone())
.unwrap_or_else(|| "0.0.0.0".to_string());
let sdp = build_sdp(&SdpOptions {
ip: &lan_ip_str,
port: dev_rtp_port,
..Default::default()
});
let ok = SipMessage::create_response(200, "OK", &device_invite, Some(ResponseOptions {
to_tag: Some(generate_tag()),
contact: Some(format!("<sip:{}:{}>", lan_ip_str, 5060)),
body: Some(sdp),
content_type: Some("application/sdp".to_string()),
extra_headers: None,
}));
let _ = socket.send_to(&ok.serialize(), dev_addr).await;
// Update device leg state.
if let Some(call) = self.calls.get_mut(call_id) {
if let Some(dev_leg) = call.legs.get_mut(&dev_leg_id) {
dev_leg.state = LegState::Connected;
}
}
// Wire device leg to mixer.
if let Some(dev_remote_addr) = dev_remote {
let dev_channels = create_leg_channels();
spawn_sip_inbound(dev_rtp_socket.clone(), dev_channels.inbound_tx);
spawn_sip_outbound(dev_rtp_socket, dev_remote_addr, dev_channels.outbound_rx);
if let Some(call) = self.calls.get(call_id) {
call.add_leg_to_mixer(&dev_leg_id, sip_pt, dev_channels.inbound_rx, dev_channels.outbound_tx)
.await;
}
}
}
}
}
emit_event(&self.out_tx, "call_answered", serde_json::json!({
"call_id": call_id,
"provider_media_addr": remote.map(|a| a.ip().to_string()),
@@ -209,6 +279,34 @@ impl CallManager {
}
SipLegAction::Terminated(reason) => {
let duration = self.calls.get(call_id).map(|c| c.duration_secs()).unwrap_or(0);
// Notify device if this is a device-originated outbound call.
if let Some(call) = self.calls.get(call_id) {
if let Some(device_invite) = &call.device_invite {
let device_leg = call.legs.values().find(|l| l.kind == LegKind::SipDevice);
if let Some(dev) = device_leg {
if let Some(dev_addr) = dev.signaling_addr {
// Map reason to SIP response code.
let code: u16 = if reason.starts_with("rejected_") {
reason.strip_prefix("rejected_")
.and_then(|s| s.parse().ok())
.unwrap_or(503)
} else if reason == "bye" {
// Provider sent BYE — send BYE to device too.
// (200 OK already connected; just let terminate_call handle it)
0
} else {
503
};
if code > 0 && dev.state != LegState::Connected {
let resp = SipMessage::create_response(code, "Service Unavailable", device_invite, None);
let _ = socket.send_to(&resp.serialize(), dev_addr).await;
}
}
}
}
}
if let Some(call) = self.calls.get_mut(call_id) {
if let Some(leg) = call.legs.get_mut(leg_id) {
leg.state = LegState::Terminated;
@@ -277,13 +375,48 @@ impl CallManager {
// Get this leg's RTP port (for SDP rewriting — tell the other side to send RTP here).
let this_rtp_port = call.legs.get(this_leg_id).map(|l| l.rtp_port).unwrap_or(0);
// Check if the other leg is a B2BUA leg (has SipLeg for proper dialog mgmt).
let other_has_sip_leg = call.legs.get(&other_leg_id)
.map(|l| l.sip_leg.is_some())
.unwrap_or(false);
if msg.is_request() {
let method = msg.method().unwrap_or("");
// ACK: In hybrid B2BUA mode, the device's ACK for our 200 OK
// is absorbed silently (provider's 200 was already ACKed by SipLeg).
if method == "ACK" {
if other_has_sip_leg {
return true; // Absorb — provider ACK handled by SipLeg.
}
// Pure passthrough: forward ACK normally.
let _ = socket.send_to(&msg.serialize(), forward_to).await;
return true;
}
// INVITE retransmit: the call already exists, re-send 100 Trying.
if method == "INVITE" {
let trying = SipMessage::create_response(100, "Trying", msg, None);
let _ = socket.send_to(&trying.serialize(), from_addr).await;
return true;
}
if method == "BYE" {
let ok = SipMessage::create_response(200, "OK", msg, None);
let _ = socket.send_to(&ok.serialize(), from_addr).await;
let _ = socket.send_to(&msg.serialize(), forward_to).await;
// If other leg has SipLeg, use build_hangup for proper dialog teardown.
if other_has_sip_leg {
if let Some(other) = call.legs.get_mut(&other_leg_id) {
if let Some(sip_leg) = &mut other.sip_leg {
if let Some(hangup_buf) = sip_leg.build_hangup() {
let _ = socket.send_to(&hangup_buf, sip_leg.config.sip_target).await;
}
}
}
} else {
let _ = socket.send_to(&msg.serialize(), forward_to).await;
}
let duration = call.duration_secs();
emit_event(
@@ -302,7 +435,19 @@ impl CallManager {
if method == "CANCEL" {
let ok = SipMessage::create_response(200, "OK", msg, None);
let _ = socket.send_to(&ok.serialize(), from_addr).await;
let _ = socket.send_to(&msg.serialize(), forward_to).await;
// If other leg has SipLeg, use build_hangup (produces CANCEL for early dialog).
if other_has_sip_leg {
if let Some(other) = call.legs.get_mut(&other_leg_id) {
if let Some(sip_leg) = &mut other.sip_leg {
if let Some(hangup_buf) = sip_leg.build_hangup() {
let _ = socket.send_to(&hangup_buf, sip_leg.config.sip_target).await;
}
}
}
} else {
let _ = socket.send_to(&msg.serialize(), forward_to).await;
}
let duration = call.duration_secs();
emit_event(
@@ -559,6 +704,7 @@ impl CallManager {
rtp_port: provider_rtp.port,
remote_media: provider_media,
signaling_addr: Some(from_addr),
metadata: HashMap::new(),
},
);
@@ -578,6 +724,7 @@ impl CallManager {
rtp_port: device_rtp.port,
remote_media: None, // Learned from device's 200 OK.
signaling_addr: Some(device_addr),
metadata: HashMap::new(),
},
);
@@ -686,6 +833,7 @@ impl CallManager {
rtp_port: rtp_alloc.port,
remote_media: None,
signaling_addr: Some(provider_dest),
metadata: HashMap::new(),
},
);
@@ -697,8 +845,12 @@ impl CallManager {
Some(call_id)
}
/// Create an outbound passthrough call (device → provider).
pub async fn create_outbound_passthrough(
/// Create a device-originated outbound call (device → provider) using hybrid B2BUA.
///
/// The device side is a simple passthrough leg (no SipLeg needed).
/// The provider side uses a full SipLeg for proper dialog management,
/// 407 auth, correct From URI, and public IP in SDP.
pub async fn create_device_outbound_call(
&mut self,
invite: &SipMessage,
from_addr: SocketAddr,
@@ -707,19 +859,28 @@ impl CallManager {
rtp_pool: &mut RtpPortPool,
socket: &UdpSocket,
public_ip: Option<&str>,
registered_aor: &str,
) -> Option<String> {
let call_id = self.next_call_id();
let lan_ip = &config.proxy.lan_ip;
let lan_port = config.proxy.lan_port;
let pub_ip = public_ip.unwrap_or(lan_ip.as_str());
let sip_call_id = invite.call_id().to_string();
let callee = invite.request_uri().unwrap_or("").to_string();
let device_sip_call_id = invite.call_id().to_string();
let dialed_number = invite
.request_uri()
.and_then(|uri| SipMessage::extract_uri(uri))
.unwrap_or(invite.request_uri().unwrap_or(""))
.to_string();
let provider_dest: SocketAddr = match provider_config.outbound_proxy.to_socket_addr() {
Some(a) => a,
None => return None,
};
// Send 100 Trying to device immediately to stop retransmissions.
let trying = SipMessage::create_response(100, "Trying", invite, None);
let _ = socket.send_to(&trying.serialize(), from_addr).await;
// Allocate RTP ports for both legs.
let device_rtp = match rtp_pool.allocate().await {
Some(a) => a,
@@ -741,9 +902,10 @@ impl CallManager {
mixer_cmd_tx,
mixer_task,
);
call.callee_number = Some(callee);
call.callee_number = Some(dialed_number.clone());
call.device_invite = Some(invite.clone());
// Device leg.
// --- Device leg (passthrough, no SipLeg) ---
let device_leg_id = format!("{call_id}-dev");
let mut device_media: Option<SocketAddr> = None;
if invite.has_sdp_body() {
@@ -759,20 +921,45 @@ impl CallManager {
LegInfo {
id: device_leg_id.clone(),
kind: LegKind::SipDevice,
state: LegState::Connected,
state: LegState::Inviting, // Not connected yet — waiting for provider answer
codec_pt,
sip_leg: None,
sip_call_id: Some(sip_call_id.clone()),
sip_call_id: Some(device_sip_call_id.clone()),
webrtc_session_id: None,
rtp_socket: Some(device_rtp.socket.clone()),
rtp_port: device_rtp.port,
remote_media: device_media,
signaling_addr: Some(from_addr),
metadata: HashMap::new(),
},
);
// Provider leg.
// Register device's SIP Call-ID → device leg.
self.sip_index
.insert(device_sip_call_id, (call_id.clone(), device_leg_id));
// --- Provider leg (B2BUA with SipLeg) ---
let provider_leg_id = format!("{call_id}-prov");
let provider_sip_call_id = generate_call_id(None);
let leg_config = SipLegConfig {
lan_ip: lan_ip.clone(),
lan_port,
public_ip: public_ip.map(|s| s.to_string()),
sip_target: provider_dest,
username: Some(provider_config.username.clone()),
password: Some(provider_config.password.clone()),
registered_aor: Some(registered_aor.to_string()),
codecs: provider_config.codecs.clone(),
rtp_port: provider_rtp.port,
};
let mut sip_leg = SipLeg::new(provider_leg_id.clone(), leg_config);
// Build proper To URI and send INVITE.
let to_uri = format!("sip:{}@{}", dialed_number, provider_config.domain);
sip_leg.send_invite(registered_aor, &to_uri, &provider_sip_call_id, socket).await;
call.legs.insert(
provider_leg_id.clone(),
LegInfo {
@@ -780,38 +967,20 @@ impl CallManager {
kind: LegKind::SipProvider,
state: LegState::Inviting,
codec_pt,
sip_leg: None,
sip_call_id: Some(sip_call_id.clone()),
sip_leg: Some(sip_leg),
sip_call_id: Some(provider_sip_call_id.clone()),
webrtc_session_id: None,
rtp_socket: Some(provider_rtp.socket.clone()),
rtp_port: provider_rtp.port,
remote_media: None,
signaling_addr: Some(provider_dest),
metadata: HashMap::new(),
},
);
// Register provider's SIP Call-ID → provider leg.
self.sip_index
.insert(sip_call_id.clone(), (call_id.clone(), device_leg_id));
// Forward INVITE to provider with SDP rewriting.
let mut fwd_invite = invite.clone();
fwd_invite.prepend_header("Record-Route", &format!("<sip:{lan_ip}:{lan_port};lr>"));
if let Some(contact) = fwd_invite.get_header("Contact").map(|s| s.to_string()) {
let new_contact = rewrite_sip_uri(&contact, pub_ip, lan_port);
if new_contact != contact {
fwd_invite.set_header("Contact", &new_contact);
}
}
// Tell provider to send RTP to our provider_rtp port.
if fwd_invite.has_sdp_body() {
let (new_body, _) = rewrite_sdp(&fwd_invite.body, pub_ip, provider_rtp.port);
fwd_invite.body = new_body;
fwd_invite.update_content_length();
}
let _ = socket.send_to(&fwd_invite.serialize(), provider_dest).await;
.insert(provider_sip_call_id, (call_id.clone(), provider_leg_id));
self.calls.insert(call_id.clone(), call);
Some(call_id)
@@ -872,6 +1041,7 @@ impl CallManager {
rtp_port: rtp_alloc.port,
remote_media: None,
signaling_addr: Some(provider_dest),
metadata: HashMap::new(),
};
self.sip_index
@@ -1099,6 +1269,7 @@ impl CallManager {
rtp_port: rtp_alloc.port,
remote_media: Some(provider_media),
signaling_addr: Some(from_addr),
metadata: HashMap::new(),
},
);