#![deny(missing_docs)]
use std::fmt;
use std::fs;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use serde::Deserialize;
use tokio::sync::Mutex;
use rvoip_sip_core::types::uri::{Scheme, Uri};
use crate::api::audio::{AudioReceiver, AudioSender, AudioStream};
use crate::api::events::Event;
use crate::api::handle::{CallId, SessionHandle};
use crate::api::incoming::{IncomingCall, IncomingCallGuard};
use crate::api::performance::PerformanceConfig;
use crate::api::stream_peer::{EventReceiver, PeerControl, StreamPeer};
use crate::api::unified::{
Config, MediaMode, Registration, RegistrationHandle, RegistrationInfo, RegistrationStatus,
SipTlsMode,
};
use crate::auth::SipClientAuth;
use crate::errors::{Result, SessionError};
use crate::types::Credentials;
pub struct Endpoint {
peer: StreamPeer,
registration: Option<Registration>,
registration_handle: SharedRegistrationHandle,
registrar: Option<String>,
transport: EndpointTransport,
}
impl Endpoint {
pub fn builder() -> EndpointBuilder {
EndpointBuilder::new()
}
pub async fn from_config(config: EndpointConfig) -> Result<Self> {
EndpointBuilder::from_config(config)?.build().await
}
pub async fn from_json_file(path: impl AsRef<Path>) -> Result<Self> {
let text = fs::read_to_string(path.as_ref()).map_err(|err| {
SessionError::ConfigError(format!(
"failed to read endpoint JSON config '{}': {err}",
path.as_ref().display()
))
})?;
let config = serde_json::from_str::<EndpointConfig>(&text).map_err(|err| {
SessionError::ConfigError(format!(
"failed to parse endpoint JSON config '{}': {err}",
path.as_ref().display()
))
})?;
Self::from_config(config).await
}
pub async fn register(&mut self) -> Result<RegistrationHandle> {
let mut stored = self.registration_handle.lock().await;
if let Some(handle) = stored.as_ref() {
return Ok(handle.clone());
}
let registration = self.registration.clone().ok_or_else(|| {
SessionError::ConfigError(
"Endpoint has no complete registration account; set account, password, and registrar"
.to_string(),
)
})?;
let mut b = self
.peer
.register(
registration.registrar.clone(),
registration.username.clone(),
registration.password.clone(),
)
.with_expires(registration.expires);
if let Some(from) = registration.from_uri.clone() {
b = b.with_from_uri(from);
}
if let Some(contact) = registration.contact_uri.clone() {
b = b.with_contact_uri(contact);
}
let handle = b.send().await?;
*stored = Some(handle.clone());
Ok(handle)
}
pub async fn register_and_wait(
&mut self,
timeout: Option<Duration>,
) -> Result<EndpointRegistrationInfo> {
let mut events = self.events().await?;
let handle = self.register().await?;
wait_for_registration_result(&mut events, &handle, timeout).await
}
pub async fn unregister(&mut self) -> Result<()> {
let mut stored = self.registration_handle.lock().await;
if let Some(handle) = stored.take() {
self.peer.unregister(&handle).await?;
}
Ok(())
}
pub async fn call_and_wait(
&self,
target: &str,
timeout: Option<Duration>,
) -> Result<EndpointCall> {
let call_id = self.invite(target)?.send().await?;
let call = self.wrap_call(call_id);
call.wait_for_answered(timeout).await
}
pub async fn wait_for_incoming(&mut self) -> Result<EndpointIncomingCall> {
let incoming = self.peer.wait_for_incoming().await?;
Ok(EndpointIncomingCall::new(
incoming,
self.registrar.clone(),
self.transport,
))
}
pub async fn events(&self) -> Result<EndpointEvents> {
let events = self.peer.control().subscribe_events().await?;
Ok(EndpointEvents::new(
events,
self.peer.control().clone(),
self.registrar.clone(),
self.transport,
))
}
pub fn split(self) -> (EndpointControl, EndpointEvents) {
let registration = self.registration;
let registration_handle = self.registration_handle;
let registrar = self.registrar;
let transport = self.transport;
let (control, events) = self.peer.split();
let endpoint_control = EndpointControl::new(
control.clone(),
registration,
registration_handle,
registrar.clone(),
transport,
);
let endpoint_events = EndpointEvents::new(events, control, registrar, transport);
(endpoint_control, endpoint_events)
}
pub fn control(&self) -> &PeerControl {
self.peer.control()
}
pub fn resolve_target(&self, target: &str) -> Result<String> {
normalize_target(self.registrar.as_deref(), target, self.transport)
}
pub fn invite(&self, target: &str) -> Result<crate::api::send::OutboundCallBuilder> {
let resolved = self.resolve_target(target)?;
Ok(self.peer.control().invite(resolved))
}
pub fn wrap_call(&self, call_id: crate::api::handle::CallId) -> EndpointCall {
let coord = self.peer.control().coordinator().clone();
EndpointCall::new(
crate::api::handle::SessionHandle::new(call_id, coord),
self.registrar.clone(),
self.transport,
)
}
pub fn into_stream_peer(self) -> StreamPeer {
self.peer
}
pub async fn shutdown(self) -> Result<()> {
self.peer.shutdown().await
}
}
type SharedRegistrationHandle = Arc<Mutex<Option<RegistrationHandle>>>;
#[derive(Clone)]
pub struct EndpointControl {
control: PeerControl,
registration: Option<Registration>,
registration_handle: SharedRegistrationHandle,
registrar: Option<String>,
transport: EndpointTransport,
}
impl EndpointControl {
fn new(
control: PeerControl,
registration: Option<Registration>,
registration_handle: SharedRegistrationHandle,
registrar: Option<String>,
transport: EndpointTransport,
) -> Self {
Self {
control,
registration,
registration_handle,
registrar,
transport,
}
}
pub async fn register(&self) -> Result<()> {
let mut stored = self.registration_handle.lock().await;
if stored.is_some() {
return Ok(());
}
let registration = self.registration.clone().ok_or_else(|| {
SessionError::ConfigError(
"Endpoint has no complete registration account; set account, password, and registrar"
.to_string(),
)
})?;
let mut b = self
.control
.coordinator()
.register(
registration.registrar,
registration.username,
registration.password,
)
.with_expires(registration.expires);
if let Some(from) = registration.from_uri {
b = b.with_from_uri(from);
}
if let Some(contact) = registration.contact_uri {
b = b.with_contact_uri(contact);
}
let handle = b.send().await?;
*stored = Some(handle);
Ok(())
}
pub async fn register_and_wait(
&self,
timeout: Option<Duration>,
) -> Result<EndpointRegistrationInfo> {
let mut events = self.events().await?;
self.register().await?;
let handle = self
.registration_handle
.lock()
.await
.clone()
.ok_or_else(|| SessionError::Other("registration handle missing".to_string()))?;
wait_for_registration_result(&mut events, &handle, timeout).await
}
pub async fn registration_info(&self) -> Result<Option<EndpointRegistrationInfo>> {
let handle = self.registration_handle.lock().await.clone();
match handle {
Some(handle) => self
.control
.coordinator()
.registration_info(&handle)
.await
.map(EndpointRegistrationInfo::from)
.map(Some),
None => Ok(None),
}
}
pub async fn unregister(&self) -> Result<()> {
if let Some(handle) = self.registration_handle.lock().await.take() {
self.control.coordinator().unregister(&handle).await?;
}
Ok(())
}
pub async fn unregister_and_wait(&self, timeout: Option<Duration>) -> Result<()> {
if let Some(handle) = self.registration_handle.lock().await.take() {
self.control
.coordinator()
.unregister_and_wait(&handle, timeout)
.await?;
}
Ok(())
}
pub async fn events(&self) -> Result<EndpointEvents> {
let events = self.control.subscribe_events().await?;
Ok(EndpointEvents::new(
events,
self.control.clone(),
self.registrar.clone(),
self.transport,
))
}
pub fn resolve_target(&self, target: &str) -> Result<String> {
normalize_target(self.registrar.as_deref(), target, self.transport)
}
pub fn invite(&self, target: &str) -> Result<crate::api::send::OutboundCallBuilder> {
let resolved = self.resolve_target(target)?;
Ok(self.control.invite(resolved))
}
pub fn wrap_call(&self, call_id: crate::api::handle::CallId) -> EndpointCall {
let coord = self.control.coordinator().clone();
EndpointCall::new(
crate::api::handle::SessionHandle::new(call_id, coord),
self.registrar.clone(),
self.transport,
)
}
pub async fn shutdown(&self) -> Result<()> {
self.control.coordinator().shutdown_gracefully(None).await
}
}
pub struct EndpointEvents {
events: EventReceiver,
control: PeerControl,
registrar: Option<String>,
transport: EndpointTransport,
}
impl EndpointEvents {
fn new(
events: EventReceiver,
control: PeerControl,
registrar: Option<String>,
transport: EndpointTransport,
) -> Self {
Self {
events,
control,
registrar,
transport,
}
}
pub async fn next(&mut self) -> Result<Option<EndpointEvent>> {
Ok(self.events.next().await.map(|event| self.map_event(event)))
}
pub fn try_next(&mut self) -> Option<EndpointEvent> {
self.events.try_next().map(|event| self.map_event(event))
}
fn map_event(&self, event: Event) -> EndpointEvent {
match event {
Event::IncomingCall {
call_id,
from,
to,
sdp,
} => {
let incoming =
IncomingCall::new(call_id, from, to, sdp, self.control.coordinator().clone());
EndpointEvent::IncomingCall(EndpointIncomingCall::new(
incoming,
self.registrar.clone(),
self.transport,
))
}
Event::CallProgress {
call_id,
status_code,
reason,
sdp,
} => EndpointEvent::CallProgress {
call_id: EndpointCallId(call_id),
status_code,
reason,
has_sdp: sdp.is_some(),
},
Event::CallAnswered { call_id, sdp } => EndpointEvent::CallAnswered {
call: EndpointCall::new(
SessionHandle::new(call_id, self.control.coordinator().clone()),
self.registrar.clone(),
self.transport,
),
has_sdp: sdp.is_some(),
},
Event::CallEnded { call_id, reason } => EndpointEvent::CallEnded {
call_id: EndpointCallId(call_id),
reason,
},
Event::CallFailed {
call_id,
status_code,
reason,
} => EndpointEvent::CallFailed {
call_id: EndpointCallId(call_id),
status_code,
reason,
},
Event::CallCancelled { call_id } => EndpointEvent::CallCancelled {
call_id: EndpointCallId(call_id),
},
Event::CallOnHold { call_id } => EndpointEvent::LocalHold {
call_id: EndpointCallId(call_id),
},
Event::CallResumed { call_id } => EndpointEvent::LocalResume {
call_id: EndpointCallId(call_id),
},
Event::RemoteCallOnHold { call_id } => EndpointEvent::RemoteHold {
call_id: EndpointCallId(call_id),
},
Event::RemoteCallResumed { call_id } => EndpointEvent::RemoteResume {
call_id: EndpointCallId(call_id),
},
Event::DtmfReceived { call_id, digit } => EndpointEvent::DtmfReceived {
call_id: EndpointCallId(call_id),
digit,
},
Event::RegistrationSuccess {
registrar,
expires,
contact,
} => EndpointEvent::RegistrationChanged(EndpointRegistrationInfo {
status: EndpointRegistrationStatus::Registered,
registrar: Some(registrar),
contact: Some(contact),
expires_secs: Some(expires),
accepted_expires_secs: Some(expires),
next_refresh_in: None,
retry_count: 0,
last_failure: None,
}),
Event::RegistrationFailed {
registrar,
status_code,
reason,
} => EndpointEvent::RegistrationChanged(EndpointRegistrationInfo {
status: EndpointRegistrationStatus::Failed,
registrar: Some(registrar),
contact: None,
expires_secs: None,
accepted_expires_secs: None,
next_refresh_in: None,
retry_count: 0,
last_failure: Some(format!("{status_code} {reason}")),
}),
Event::UnregistrationSuccess { registrar } => {
EndpointEvent::RegistrationChanged(EndpointRegistrationInfo {
status: EndpointRegistrationStatus::Unregistered,
registrar: Some(registrar),
contact: None,
expires_secs: None,
accepted_expires_secs: None,
next_refresh_in: None,
retry_count: 0,
last_failure: None,
})
}
Event::UnregistrationFailed { registrar, reason } => {
EndpointEvent::RegistrationChanged(EndpointRegistrationInfo {
status: EndpointRegistrationStatus::Failed,
registrar: Some(registrar),
contact: None,
expires_secs: None,
accepted_expires_secs: None,
next_refresh_in: None,
retry_count: 0,
last_failure: Some(reason),
})
}
Event::NetworkError { call_id, error } => EndpointEvent::NetworkError {
call_id: call_id.map(EndpointCallId),
error,
},
Event::SipTrace(trace) => EndpointEvent::SipTrace(EndpointSipTrace {
direction: trace.direction,
transport: trace.transport,
local_addr: trace.local_addr,
remote_addr: trace.remote_addr,
timestamp_unix_millis: trace.timestamp_unix_millis,
start_line: trace.start_line,
sip_call_id: trace.sip_call_id,
session_id: trace.session_id.map(EndpointCallId),
raw_message: trace.raw_message,
original_len: trace.original_len,
truncated: trace.truncated,
redacted: trace.redacted,
}),
other => EndpointEvent::Info {
call_id: other.call_id().cloned().map(EndpointCallId),
message: format!("{other:?}"),
},
}
}
}
pub enum EndpointEvent {
IncomingCall(EndpointIncomingCall),
CallProgress {
call_id: EndpointCallId,
status_code: u16,
reason: String,
has_sdp: bool,
},
CallAnswered {
call: EndpointCall,
has_sdp: bool,
},
CallEnded {
call_id: EndpointCallId,
reason: String,
},
CallFailed {
call_id: EndpointCallId,
status_code: u16,
reason: String,
},
CallCancelled {
call_id: EndpointCallId,
},
LocalHold {
call_id: EndpointCallId,
},
LocalResume {
call_id: EndpointCallId,
},
RemoteHold {
call_id: EndpointCallId,
},
RemoteResume {
call_id: EndpointCallId,
},
DtmfReceived {
call_id: EndpointCallId,
digit: char,
},
RegistrationChanged(EndpointRegistrationInfo),
SipTrace(EndpointSipTrace),
NetworkError {
call_id: Option<EndpointCallId>,
error: String,
},
Info {
call_id: Option<EndpointCallId>,
message: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EndpointSipTrace {
pub direction: crate::api::events::SipTraceDirection,
pub transport: String,
pub local_addr: String,
pub remote_addr: String,
pub timestamp_unix_millis: u64,
pub start_line: String,
pub sip_call_id: Option<String>,
pub session_id: Option<EndpointCallId>,
pub raw_message: String,
pub original_len: usize,
pub truncated: bool,
pub redacted: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct EndpointCallId(CallId);
impl fmt::Display for EndpointCallId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(f)
}
}
#[derive(Clone)]
pub struct EndpointCall {
handle: SessionHandle,
registrar: Option<String>,
transport: EndpointTransport,
}
impl EndpointCall {
fn new(handle: SessionHandle, registrar: Option<String>, transport: EndpointTransport) -> Self {
Self {
handle,
registrar,
transport,
}
}
pub fn id(&self) -> EndpointCallId {
EndpointCallId(self.handle.id().clone())
}
pub fn as_session_handle(&self) -> &SessionHandle {
&self.handle
}
pub async fn wait_for_answered(&self, timeout: Option<Duration>) -> Result<Self> {
let handle = self.handle.wait_for_answered(timeout).await?;
Ok(Self::new(handle, self.registrar.clone(), self.transport))
}
pub async fn wait_for_end(&self, timeout: Option<Duration>) -> Result<String> {
self.handle.wait_for_end(timeout).await
}
pub async fn hangup(&self) -> Result<()> {
self.handle.hangup().await
}
pub async fn hangup_and_wait(&self, timeout: Option<Duration>) -> Result<String> {
self.handle.hangup_and_wait(timeout).await
}
pub async fn hold(&self) -> Result<()> {
self.handle.hold().await
}
pub async fn resume(&self) -> Result<()> {
self.handle.resume().await
}
pub async fn mute(&self) -> Result<()> {
self.handle.mute().await
}
pub async fn unmute(&self) -> Result<()> {
self.handle.unmute().await
}
pub async fn send_dtmf(&self, digit: char) -> Result<()> {
self.handle.send_dtmf(digit).await
}
pub async fn transfer(&self, target: &str) -> Result<()> {
let target = normalize_target(self.registrar.as_deref(), target, self.transport)?;
self.handle.transfer_blind(&target).await
}
pub async fn audio(&self) -> Result<EndpointAudio> {
self.handle.audio().await.map(EndpointAudio::new)
}
}
impl fmt::Debug for EndpointCall {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("EndpointCall")
.field("id", &self.id().to_string())
.finish()
}
}
pub struct EndpointIncomingCall {
incoming: IncomingCall,
registrar: Option<String>,
transport: EndpointTransport,
}
impl EndpointIncomingCall {
fn new(
incoming: IncomingCall,
registrar: Option<String>,
transport: EndpointTransport,
) -> Self {
Self {
incoming,
registrar,
transport,
}
}
pub fn id(&self) -> EndpointCallId {
EndpointCallId(self.incoming.call_id.clone())
}
pub fn from(&self) -> &str {
&self.incoming.from
}
pub fn to(&self) -> &str {
&self.incoming.to
}
pub async fn answer(self) -> Result<EndpointCall> {
let handle = self.incoming.accept().await?;
Ok(EndpointCall::new(handle, self.registrar, self.transport))
}
pub async fn accept(self) -> Result<EndpointCall> {
self.answer().await
}
pub fn defer(self, watchdog: Duration) -> IncomingCallGuard {
self.incoming.defer(watchdog)
}
pub async fn decline(self) -> Result<()> {
self.reject(603, "Decline").await
}
pub async fn busy(self) -> Result<()> {
self.reject(486, "Busy Here").await
}
pub async fn reject(self, status: u16, reason: &str) -> Result<()> {
self.incoming.reject(status, reason);
Ok(())
}
pub async fn redirect_to(self, target: impl Into<String>) -> Result<()> {
self.incoming.redirect_to(target).await
}
pub async fn redirect_with_contacts<I, S>(self, status: u16, contacts: I) -> Result<()>
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.incoming.redirect_with_contacts(status, contacts).await
}
}
impl fmt::Debug for EndpointIncomingCall {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("EndpointIncomingCall")
.field("id", &self.id().to_string())
.field("from", &self.from())
.field("to", &self.to())
.finish()
}
}
pub struct EndpointAudio {
stream: AudioStream,
}
impl EndpointAudio {
fn new(stream: AudioStream) -> Self {
Self { stream }
}
pub fn split(self) -> (EndpointAudioSender, EndpointAudioReceiver) {
let (sender, receiver) = self.stream.split();
(
EndpointAudioSender { sender },
EndpointAudioReceiver { receiver },
)
}
}
#[derive(Clone)]
pub struct EndpointAudioSender {
sender: AudioSender,
}
impl EndpointAudioSender {
pub async fn send(&self, frame: EndpointAudioFrame) -> Result<()> {
self.sender.send(frame.into()).await
}
pub fn is_open(&self) -> bool {
self.sender.is_open()
}
}
pub struct EndpointAudioReceiver {
receiver: AudioReceiver,
}
impl EndpointAudioReceiver {
pub async fn recv(&mut self) -> Option<EndpointAudioFrame> {
self.receiver.recv().await.map(EndpointAudioFrame::from)
}
pub fn try_recv(&mut self) -> Option<EndpointAudioFrame> {
self.receiver.try_recv().map(EndpointAudioFrame::from)
}
}
#[derive(Debug, Clone, Deserialize)]
pub struct EndpointAudioFrame {
pub samples: Vec<i16>,
pub sample_rate: u32,
pub channels: u8,
pub timestamp: u32,
}
impl EndpointAudioFrame {
pub fn new(samples: Vec<i16>, sample_rate: u32, channels: u8, timestamp: u32) -> Self {
Self {
samples,
sample_rate,
channels,
timestamp,
}
}
pub fn pcmu_sized_mono_8khz(samples: Vec<i16>, timestamp: u32) -> Self {
Self::new(samples, 8_000, 1, timestamp)
}
pub fn samples_per_channel(&self) -> usize {
self.samples.len() / self.channels.max(1) as usize
}
}
impl From<EndpointAudioFrame> for rvoip_media_core::types::AudioFrame {
fn from(frame: EndpointAudioFrame) -> Self {
rvoip_media_core::types::AudioFrame::new(
frame.samples,
frame.sample_rate,
frame.channels,
frame.timestamp,
)
}
}
impl From<rvoip_media_core::types::AudioFrame> for EndpointAudioFrame {
fn from(frame: rvoip_media_core::types::AudioFrame) -> Self {
Self {
samples: frame.samples,
sample_rate: frame.sample_rate,
channels: frame.channels,
timestamp: frame.timestamp,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EndpointRegistrationStatus {
Registering,
Registered,
Unregistering,
Unregistered,
Failed,
}
#[derive(Debug, Clone)]
pub struct EndpointRegistrationInfo {
pub status: EndpointRegistrationStatus,
pub registrar: Option<String>,
pub contact: Option<String>,
pub expires_secs: Option<u32>,
pub accepted_expires_secs: Option<u32>,
pub next_refresh_in: Option<Duration>,
pub retry_count: u32,
pub last_failure: Option<String>,
}
impl From<RegistrationInfo> for EndpointRegistrationInfo {
fn from(info: RegistrationInfo) -> Self {
Self {
status: match info.status {
RegistrationStatus::Registering => EndpointRegistrationStatus::Registering,
RegistrationStatus::Registered => EndpointRegistrationStatus::Registered,
RegistrationStatus::Unregistering => EndpointRegistrationStatus::Unregistering,
RegistrationStatus::Unregistered => EndpointRegistrationStatus::Unregistered,
RegistrationStatus::Failed => EndpointRegistrationStatus::Failed,
},
registrar: info.registrar,
contact: info.contact,
expires_secs: info.expires_secs,
accepted_expires_secs: info.accepted_expires_secs,
next_refresh_in: info.next_refresh_in,
retry_count: info.retry_count,
last_failure: info.last_failure,
}
}
}
#[derive(Debug, Clone)]
pub struct SipAccount {
pub registrar: String,
pub username: String,
pub auth_username: Option<String>,
pub password: String,
pub expires: u32,
pub from_uri: Option<String>,
pub contact_uri: Option<String>,
}
impl SipAccount {
pub fn new(
registrar: impl Into<String>,
username: impl Into<String>,
password: impl Into<String>,
) -> Self {
Self {
registrar: registrar.into(),
username: username.into(),
auth_username: None,
password: password.into(),
expires: 3600,
from_uri: None,
contact_uri: None,
}
}
pub fn effective_auth_username(&self) -> &str {
self.auth_username
.as_deref()
.unwrap_or(self.username.as_str())
}
pub fn credentials(&self) -> Credentials {
Credentials::new(self.effective_auth_username(), self.password.clone())
}
pub fn registration(&self) -> Registration {
let mut registration = Registration::new(
self.registrar.clone(),
self.effective_auth_username().to_string(),
self.password.clone(),
)
.expires(self.expires);
if let Some(from_uri) = &self.from_uri {
registration = registration.from_uri(from_uri.clone());
}
if let Some(contact_uri) = &self.contact_uri {
registration = registration.contact_uri(contact_uri.clone());
}
registration
}
pub fn endpoint_account(&self) -> EndpointAccount {
EndpointAccount {
registrar: self.registrar.clone(),
username: self.username.clone(),
auth_username: self.auth_username.clone(),
password: self.password.clone(),
expires: self.expires,
from_uri: self.from_uri.clone(),
contact_uri: self.contact_uri.clone(),
}
}
pub fn auth_username(mut self, username: impl Into<String>) -> Self {
self.auth_username = Some(username.into());
self
}
pub fn expires(mut self, seconds: u32) -> Self {
self.expires = seconds;
self
}
pub fn from_uri(mut self, uri: impl Into<String>) -> Self {
self.from_uri = Some(uri.into());
self
}
pub fn contact_uri(mut self, uri: impl Into<String>) -> Self {
self.contact_uri = Some(uri.into());
self
}
}
#[derive(Debug, Clone)]
pub struct EndpointAccount {
pub registrar: String,
pub username: String,
pub auth_username: Option<String>,
pub password: String,
pub expires: u32,
pub from_uri: Option<String>,
pub contact_uri: Option<String>,
}
impl EndpointAccount {
pub fn new(
registrar: impl Into<String>,
username: impl Into<String>,
password: impl Into<String>,
) -> Self {
Self {
registrar: registrar.into(),
username: username.into(),
auth_username: None,
password: password.into(),
expires: 3600,
from_uri: None,
contact_uri: None,
}
}
pub fn auth_username(mut self, username: impl Into<String>) -> Self {
self.auth_username = Some(username.into());
self
}
pub fn expires(mut self, seconds: u32) -> Self {
self.expires = seconds;
self
}
pub fn from_uri(mut self, uri: impl Into<String>) -> Self {
self.from_uri = Some(uri.into());
self
}
pub fn contact_uri(mut self, uri: impl Into<String>) -> Self {
self.contact_uri = Some(uri.into());
self
}
}
impl From<SipAccount> for EndpointAccount {
fn from(account: SipAccount) -> Self {
account.endpoint_account()
}
}
impl From<EndpointAccount> for SipAccount {
fn from(account: EndpointAccount) -> Self {
Self {
registrar: account.registrar,
username: account.username,
auth_username: account.auth_username,
password: account.password,
expires: account.expires,
from_uri: account.from_uri,
contact_uri: account.contact_uri,
}
}
}
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EndpointConfig {
pub name: Option<String>,
pub profile: Option<EndpointProfileName>,
pub bind: Option<SocketAddr>,
pub advertise: Option<SocketAddr>,
pub account: Option<EndpointAccountConfig>,
pub network: Option<EndpointNetworkConfig>,
pub media: Option<EndpointMediaConfig>,
pub performance: Option<PerformanceConfig>,
pub auto_180_ringing: Option<bool>,
pub auto_100_trying: Option<bool>,
pub fast_auto_accept_incoming_calls: Option<bool>,
pub cleanup_diagnostics: Option<bool>,
pub cleanup_diagnostic_events: Option<bool>,
pub app_event_channel_capacity: Option<usize>,
pub sip_transaction_command_channel_capacity: Option<usize>,
pub server_call_admission_limit: Option<usize>,
pub server_call_admission_soft_limit: Option<usize>,
pub server_call_admission_pacing_delay_ms: Option<u64>,
pub server_overload_retry_after_secs: Option<u32>,
#[cfg(feature = "perf-tests")]
pub perf_max_rss_growth_mb_per_hr: Option<f64>,
pub srtp_diagnostics: Option<bool>,
pub rtp_diagnostics: Option<bool>,
pub media_sdp_diagnostics: Option<bool>,
pub sip_trace: Option<crate::api::events::SipTraceConfig>,
pub register_on_start: Option<bool>,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EndpointAccountConfig {
pub registrar: String,
pub username: String,
pub auth_username: Option<String>,
pub password: String,
pub expires: Option<u32>,
pub from_uri: Option<String>,
pub contact_uri: Option<String>,
}
impl TryFrom<EndpointAccountConfig> for EndpointAccount {
type Error = SessionError;
fn try_from(config: EndpointAccountConfig) -> Result<Self> {
let mut account = EndpointAccount::new(config.registrar, config.username, config.password);
if let Some(auth_username) = config.auth_username {
account = account.auth_username(auth_username);
}
if let Some(expires) = config.expires {
account = account.expires(expires);
}
if let Some(from_uri) = config.from_uri {
account = account.from_uri(from_uri);
}
if let Some(contact_uri) = config.contact_uri {
account = account.contact_uri(contact_uri);
}
Ok(account)
}
}
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EndpointNetworkConfig {
pub bind: Option<SocketAddr>,
pub advertise: Option<SocketAddr>,
pub transport: Option<EndpointTransport>,
pub stun: Option<String>,
pub outbound_proxy: Option<String>,
pub sip_instance: Option<String>,
pub tls_bind: Option<SocketAddr>,
pub tls_cert_path: Option<PathBuf>,
pub tls_key_path: Option<PathBuf>,
pub udp_parse_workers: Option<usize>,
pub udp_parse_queue_capacity: Option<usize>,
}
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EndpointMediaConfig {
pub public_address: Option<String>,
pub port_start: Option<u16>,
pub port_end: Option<u16>,
pub enabled: Option<bool>,
pub signaling_only_rtp_port: Option<u16>,
pub srtp: Option<EndpointSrtpMode>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum EndpointProfileName {
Local,
LanPbx,
AsteriskUdp,
AsteriskTlsSrtp,
FreeswitchInternal,
FreeswitchTlsSrtp,
CarrierSbc,
}
impl From<EndpointProfileName> for EndpointProfile {
fn from(profile: EndpointProfileName) -> Self {
match profile {
EndpointProfileName::Local => EndpointProfile::Local,
EndpointProfileName::LanPbx => EndpointProfile::LanPbx,
EndpointProfileName::AsteriskUdp => EndpointProfile::AsteriskUdp,
EndpointProfileName::AsteriskTlsSrtp => EndpointProfile::AsteriskTlsSrtpRegisteredFlow,
EndpointProfileName::FreeswitchInternal => EndpointProfile::FreeSwitchInternal,
EndpointProfileName::FreeswitchTlsSrtp => {
EndpointProfile::FreeSwitchTlsSrtpReachableContact
}
EndpointProfileName::CarrierSbc => EndpointProfile::CarrierSbc,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum EndpointTransport {
Udp,
Tcp,
Tls,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum EndpointSrtpMode {
Off,
Offer,
Required,
}
#[derive(Debug, Clone)]
pub enum EndpointProfile {
Local,
LanPbx,
AsteriskUdp,
AsteriskTlsSrtpRegisteredFlow,
FreeSwitchInternal,
FreeSwitchTlsSrtpReachableContact,
CarrierSbc,
Custom(Config),
}
impl Default for EndpointProfile {
fn default() -> Self {
Self::Local
}
}
pub struct EndpointBuilder {
name: Option<String>,
profile: EndpointProfile,
bind_addr: Option<SocketAddr>,
advertised_addr: Option<SocketAddr>,
tls_bind_addr: Option<SocketAddr>,
tls_cert_path: Option<std::path::PathBuf>,
tls_key_path: Option<std::path::PathBuf>,
media_port_start: Option<u16>,
media_port_end: Option<u16>,
media_public_addr: Option<SocketAddr>,
media_mode: Option<MediaMode>,
stun_server: Option<String>,
outbound_proxy_uri: Option<String>,
sip_instance: Option<String>,
transport: EndpointTransport,
sip_udp_parse_workers: Option<usize>,
sip_udp_parse_queue_capacity: Option<usize>,
performance: Option<PerformanceConfig>,
srtp_mode: Option<EndpointSrtpMode>,
auto_180_ringing: Option<bool>,
auto_100_trying: Option<bool>,
fast_auto_accept_incoming_calls: Option<bool>,
cleanup_diagnostics: Option<bool>,
cleanup_diagnostic_events: Option<bool>,
app_event_channel_capacity: Option<usize>,
sip_transaction_command_channel_capacity: Option<usize>,
server_call_admission_limit: Option<usize>,
server_call_admission_soft_limit: Option<usize>,
server_call_admission_pacing_delay_ms: Option<u64>,
server_overload_retry_after_secs: Option<u32>,
#[cfg(feature = "perf-tests")]
perf_max_rss_growth_mb_per_hr: Option<f64>,
srtp_diagnostics: Option<bool>,
rtp_diagnostics: Option<bool>,
media_sdp_diagnostics: Option<bool>,
account_username: Option<String>,
auth_username: Option<String>,
password: Option<String>,
auth: Option<SipClientAuth>,
registrar: Option<String>,
expires: u32,
sip_trace: Option<crate::api::events::SipTraceConfig>,
from_uri: Option<String>,
contact_uri: Option<String>,
configurators: Vec<Box<dyn FnOnce(&mut Config) + Send>>,
}
impl EndpointBuilder {
pub fn new() -> Self {
Self {
name: None,
profile: EndpointProfile::Local,
bind_addr: None,
advertised_addr: None,
tls_bind_addr: None,
tls_cert_path: None,
tls_key_path: None,
media_port_start: None,
media_port_end: None,
media_public_addr: None,
media_mode: None,
stun_server: None,
outbound_proxy_uri: None,
sip_instance: None,
transport: EndpointTransport::Udp,
sip_udp_parse_workers: None,
sip_udp_parse_queue_capacity: None,
performance: None,
srtp_mode: None,
auto_180_ringing: None,
auto_100_trying: None,
fast_auto_accept_incoming_calls: None,
cleanup_diagnostics: None,
cleanup_diagnostic_events: None,
app_event_channel_capacity: None,
sip_transaction_command_channel_capacity: None,
server_call_admission_limit: None,
server_call_admission_soft_limit: None,
server_call_admission_pacing_delay_ms: None,
server_overload_retry_after_secs: None,
#[cfg(feature = "perf-tests")]
perf_max_rss_growth_mb_per_hr: None,
srtp_diagnostics: None,
rtp_diagnostics: None,
media_sdp_diagnostics: None,
account_username: None,
auth_username: None,
password: None,
auth: None,
registrar: None,
expires: 3600,
sip_trace: None,
from_uri: None,
contact_uri: None,
configurators: Vec::new(),
}
}
pub fn from_config(config: EndpointConfig) -> Result<Self> {
let mut builder = EndpointBuilder::new();
if let Some(name) = config.name {
builder = builder.name(name);
}
if let Some(profile) = config.profile {
builder = builder.profile(profile.into());
}
if let Some(performance) = config.performance {
builder = builder.performance_config(performance);
}
if let Some(bind) = config.bind.or(config.network.as_ref().and_then(|n| n.bind)) {
builder = builder.bind_addr(bind);
}
if let Some(advertise) = config
.advertise
.or(config.network.as_ref().and_then(|n| n.advertise))
{
builder = builder.advertised_addr(advertise);
}
if let Some(account) = config.account {
builder = builder.endpoint_account(account.try_into()?);
}
if let Some(auto_180_ringing) = config.auto_180_ringing {
builder = builder.auto_180_ringing(auto_180_ringing);
}
if let Some(auto_100_trying) = config.auto_100_trying {
builder = builder.auto_100_trying(auto_100_trying);
}
if let Some(fast_auto_accept) = config.fast_auto_accept_incoming_calls {
builder = builder.fast_auto_accept_incoming_calls(fast_auto_accept);
}
if let Some(cleanup_diagnostics) = config.cleanup_diagnostics {
builder = builder.cleanup_diagnostics(cleanup_diagnostics);
}
if let Some(cleanup_diagnostic_events) = config.cleanup_diagnostic_events {
builder = builder.cleanup_diagnostic_events(cleanup_diagnostic_events);
}
if let Some(capacity) = config.app_event_channel_capacity {
builder = builder.app_event_channel_capacity(capacity);
}
if let Some(capacity) = config.sip_transaction_command_channel_capacity {
builder = builder.sip_transaction_command_channel_capacity(capacity);
}
if let Some(limit) = config.server_call_admission_limit {
builder = builder.server_call_admission_limit(limit);
}
if let Some(limit) = config.server_call_admission_soft_limit {
builder = builder.server_call_admission_soft_limit(limit);
}
if let Some(delay_ms) = config.server_call_admission_pacing_delay_ms {
builder = builder.server_call_admission_pacing_delay_ms(delay_ms);
}
if let Some(seconds) = config.server_overload_retry_after_secs {
builder = builder.server_overload_retry_after_secs(seconds);
}
#[cfg(feature = "perf-tests")]
if let Some(limit) = config.perf_max_rss_growth_mb_per_hr {
builder = builder.perf_max_rss_growth_mb_per_hr(limit);
}
if let Some(srtp_diagnostics) = config.srtp_diagnostics {
builder = builder.srtp_diagnostics(srtp_diagnostics);
}
if let Some(rtp_diagnostics) = config.rtp_diagnostics {
builder = builder.rtp_diagnostics(rtp_diagnostics);
}
if let Some(media_sdp_diagnostics) = config.media_sdp_diagnostics {
builder = builder.media_sdp_diagnostics(media_sdp_diagnostics);
}
if let Some(network) = config.network {
if let Some(transport) = network.transport {
builder = builder.transport(transport);
}
if let Some(stun) = network.stun {
builder = builder.stun_server(stun);
}
if let Some(proxy) = network.outbound_proxy {
builder = builder.outbound_proxy(proxy);
}
if let Some(instance) = network.sip_instance {
builder = builder.sip_instance(instance);
}
if let Some(tls_bind) = network.tls_bind {
builder = builder.tls_bind_addr(tls_bind);
}
if let Some(path) = network.tls_cert_path {
builder = builder.tls_cert_path(path);
}
if let Some(path) = network.tls_key_path {
builder = builder.tls_key_path(path);
}
if let Some(workers) = network.udp_parse_workers {
builder = builder.sip_udp_parse_workers(workers);
}
if let Some(capacity) = network.udp_parse_queue_capacity {
builder = builder.sip_udp_parse_queue_capacity(capacity);
}
}
if let Some(media) = config.media {
if let Some(public) = media.public_address {
builder = builder.media_public_addr(parse_media_public_address(&public)?);
}
if let Some(start) = media.port_start {
let end = media.port_end.unwrap_or(start);
builder = builder.media_ports(start, end);
} else if let Some(end) = media.port_end {
builder = builder.media_ports(Config::DEFAULT_MEDIA_PORT_START, end);
}
if let Some(srtp) = media.srtp {
builder = builder.srtp(srtp);
}
if media.enabled == Some(false) || media.signaling_only_rtp_port.is_some() {
builder = builder.signaling_only_media(media.signaling_only_rtp_port.unwrap_or(9));
} else if media.enabled == Some(true) {
builder = builder.media_enabled(true);
}
}
if let Some(sip_trace) = config.sip_trace {
builder = builder.sip_trace(sip_trace);
}
Ok(builder)
}
pub fn name(mut self, name: impl Into<String>) -> Self {
self.name = Some(name.into());
self
}
pub fn account(mut self, username: impl Into<String>) -> Self {
self.account_username = Some(username.into());
self
}
pub fn endpoint_account(mut self, account: EndpointAccount) -> Self {
self.registrar = Some(account.registrar);
self.account_username = Some(account.username);
self.auth_username = account.auth_username;
self.password = Some(account.password);
self.expires = account.expires;
self.from_uri = account.from_uri;
self.contact_uri = account.contact_uri;
self
}
pub fn sip_account(self, account: SipAccount) -> Self {
self.endpoint_account(account.into())
}
pub fn auth_username(mut self, username: impl Into<String>) -> Self {
self.auth_username = Some(username.into());
self
}
pub fn password(mut self, password: impl Into<String>) -> Self {
self.password = Some(password.into());
self
}
pub fn auth(mut self, auth: SipClientAuth) -> Self {
self.auth = Some(auth);
self
}
pub fn bearer_token(mut self, token: impl Into<String>) -> Self {
self.auth = Some(SipClientAuth::bearer_token(token));
self
}
pub fn basic_credentials(
mut self,
username: impl Into<String>,
password: impl Into<String>,
) -> Self {
self.auth = Some(SipClientAuth::basic(username, password));
self
}
pub fn registrar(mut self, registrar: impl Into<String>) -> Self {
self.registrar = Some(registrar.into());
self
}
pub fn expires(mut self, seconds: u32) -> Self {
self.expires = seconds;
self
}
pub fn profile(mut self, profile: EndpointProfile) -> Self {
self.profile = profile;
self
}
pub fn config(self, config: EndpointConfig) -> Result<Self> {
let mut configured = EndpointBuilder::from_config(config)?;
if self.name.is_some() {
configured.name = self.name;
}
if self.sip_trace.is_some() {
configured.sip_trace = self.sip_trace;
}
Ok(configured)
}
pub fn bind_addr(mut self, addr: SocketAddr) -> Self {
self.bind_addr = Some(addr);
self
}
pub fn advertised_addr(mut self, addr: SocketAddr) -> Self {
self.advertised_addr = Some(addr);
self
}
pub fn tls_bind_addr(mut self, addr: SocketAddr) -> Self {
self.tls_bind_addr = Some(addr);
self
}
pub fn tls_cert_path(mut self, path: impl Into<std::path::PathBuf>) -> Self {
self.tls_cert_path = Some(path.into());
self
}
pub fn tls_key_path(mut self, path: impl Into<std::path::PathBuf>) -> Self {
self.tls_key_path = Some(path.into());
self
}
pub fn media_ports(mut self, start: u16, end: u16) -> Self {
self.media_port_start = Some(start);
self.media_port_end = Some(end);
self
}
pub fn media_enabled(mut self, enabled: bool) -> Self {
self.media_mode = Some(if enabled {
MediaMode::Enabled
} else {
MediaMode::SignalingOnly { sdp_rtp_port: 9 }
});
self
}
pub fn signaling_only_media(mut self, sdp_rtp_port: u16) -> Self {
self.media_mode = Some(MediaMode::SignalingOnly { sdp_rtp_port });
self
}
pub fn media_public_addr(mut self, addr: SocketAddr) -> Self {
self.media_public_addr = Some(addr);
self
}
pub fn media_public_ip(mut self, addr: IpAddr) -> Self {
self.media_public_addr = Some(SocketAddr::new(addr, 0));
self
}
pub fn stun_server(mut self, server: impl Into<String>) -> Self {
self.stun_server = Some(server.into());
self
}
pub fn outbound_proxy(mut self, uri: impl Into<String>) -> Self {
self.outbound_proxy_uri = Some(uri.into());
self
}
pub fn sip_instance(mut self, urn: impl Into<String>) -> Self {
self.sip_instance = Some(urn.into());
self
}
pub fn transport(mut self, transport: EndpointTransport) -> Self {
self.transport = transport;
self
}
pub fn sip_udp_parse_workers(mut self, workers: usize) -> Self {
self.sip_udp_parse_workers = Some(workers);
self
}
pub fn sip_udp_parse_queue_capacity(mut self, capacity: usize) -> Self {
self.sip_udp_parse_queue_capacity = Some(capacity);
self
}
pub fn performance_config(mut self, performance: PerformanceConfig) -> Self {
self.performance = Some(performance);
self
}
pub fn pbx_media_server_performance(mut self, capacity: usize) -> Self {
self.performance = Some(PerformanceConfig::pbx_media_server(capacity));
self
}
pub fn signaling_only_server_high_performance(mut self, capacity: usize) -> Self {
self.performance = Some(PerformanceConfig::signaling_only_server_high_performance(
capacity,
));
self
}
pub fn signaling_only_server_high_performance_with_port(
mut self,
capacity: usize,
sdp_rtp_port: u16,
) -> Self {
self.performance = Some(
PerformanceConfig::signaling_only_server_high_performance(capacity)
.with_signaling_only_rtp_port(sdp_rtp_port),
);
self
}
pub fn auto_180_ringing(mut self, enabled: bool) -> Self {
self.auto_180_ringing = Some(enabled);
self
}
pub fn auto_100_trying(mut self, enabled: bool) -> Self {
self.auto_100_trying = Some(enabled);
self
}
pub fn fast_auto_accept_incoming_calls(mut self, enabled: bool) -> Self {
self.fast_auto_accept_incoming_calls = Some(enabled);
self
}
pub fn cleanup_diagnostics(mut self, enabled: bool) -> Self {
self.cleanup_diagnostics = Some(enabled);
self
}
pub fn cleanup_diagnostic_events(mut self, enabled: bool) -> Self {
self.cleanup_diagnostic_events = Some(enabled);
self
}
pub fn app_event_channel_capacity(mut self, capacity: usize) -> Self {
self.app_event_channel_capacity = Some(capacity);
self
}
pub fn sip_transaction_command_channel_capacity(mut self, capacity: usize) -> Self {
self.sip_transaction_command_channel_capacity = Some(capacity);
self
}
pub fn server_call_admission_limit(mut self, limit: usize) -> Self {
self.server_call_admission_limit = Some(limit);
self
}
pub fn server_call_admission_soft_limit(mut self, limit: usize) -> Self {
self.server_call_admission_soft_limit = Some(limit);
self
}
pub fn server_call_admission_pacing_delay_ms(mut self, delay_ms: u64) -> Self {
self.server_call_admission_pacing_delay_ms = Some(delay_ms);
self
}
pub fn server_overload_retry_after_secs(mut self, seconds: u32) -> Self {
self.server_overload_retry_after_secs = Some(seconds);
self
}
#[cfg(feature = "perf-tests")]
pub fn perf_max_rss_growth_mb_per_hr(mut self, limit: f64) -> Self {
self.perf_max_rss_growth_mb_per_hr = Some(limit);
self
}
pub fn srtp_diagnostics(mut self, enabled: bool) -> Self {
self.srtp_diagnostics = Some(enabled);
self
}
pub fn rtp_diagnostics(mut self, enabled: bool) -> Self {
self.rtp_diagnostics = Some(enabled);
self
}
pub fn media_sdp_diagnostics(mut self, enabled: bool) -> Self {
self.media_sdp_diagnostics = Some(enabled);
self
}
pub fn srtp(mut self, mode: EndpointSrtpMode) -> Self {
self.srtp_mode = Some(mode);
self
}
pub fn enable_sip_trace(mut self) -> Self {
self.sip_trace = Some(crate::api::events::SipTraceConfig::enabled());
self
}
pub fn sip_trace(mut self, config: crate::api::events::SipTraceConfig) -> Self {
self.sip_trace = Some(config);
self
}
pub fn from_uri(mut self, uri: impl Into<String>) -> Self {
self.from_uri = Some(uri.into());
self
}
pub fn contact_uri(mut self, uri: impl Into<String>) -> Self {
self.contact_uri = Some(uri.into());
self
}
pub fn configure(mut self, f: impl FnOnce(&mut Config) + Send + 'static) -> Self {
self.configurators.push(Box::new(f));
self
}
pub async fn build(self) -> Result<Endpoint> {
let parts = self.build_parts()?;
let peer = StreamPeer::with_config(parts.config).await?;
Ok(Endpoint {
peer,
registration: parts.registration,
registration_handle: Arc::new(Mutex::new(None)),
registrar: parts.registrar,
transport: parts.transport,
})
}
fn build_parts(self) -> Result<EndpointParts> {
let mut config = self.profile_config()?;
let registrar = self
.registrar
.clone()
.map(|uri| apply_transport_to_uri(&uri, self.transport, true));
let account_username = self.account_username.clone();
if let (Some(username), Some(password)) = (&account_username, &self.password) {
let auth_username = self.auth_username.as_deref().unwrap_or(username);
config.credentials = Some(Credentials::new(auth_username, password));
}
if let Some(auth) = self.auth {
config.auth = Some(auth);
}
if let Some(performance) = self.performance {
config = config.try_with_performance_config(performance)?;
}
if self.media_port_start.is_some() || self.media_port_end.is_some() {
let media_port_start = self.media_port_start.unwrap_or(config.media_port_start);
let media_port_end = self.media_port_end.unwrap_or(config.media_port_end);
config = config.with_media_ports(media_port_start, media_port_end);
}
if let Some(addr) = self.media_public_addr {
config.media_public_addr = Some(addr);
}
if let Some(mode) = self.media_mode {
config.media_mode = mode;
}
if let Some(stun) = self.stun_server {
config.stun_server = Some(stun);
}
if let Some(outbound_proxy) = self.outbound_proxy_uri.as_ref() {
config.outbound_proxy_uri =
Some(apply_transport_to_uri(outbound_proxy, self.transport, true));
}
if let Some(srtp_mode) = self.srtp_mode {
match srtp_mode {
EndpointSrtpMode::Off => {
config.offer_srtp = false;
config.srtp_required = false;
}
EndpointSrtpMode::Offer => {
config.offer_srtp = true;
config.srtp_required = false;
}
EndpointSrtpMode::Required => {
config.offer_srtp = true;
config.srtp_required = true;
}
}
}
if let Some(sip_trace) = self.sip_trace {
config.sip_trace = sip_trace;
}
if let Some(workers) = self.sip_udp_parse_workers {
config.sip_udp_parse_workers = Some(workers);
}
if let Some(capacity) = self.sip_udp_parse_queue_capacity {
config.sip_udp_parse_queue_capacity = Some(capacity);
}
if let Some(auto_180_ringing) = self.auto_180_ringing {
config.auto_180_ringing = auto_180_ringing;
}
if let Some(auto_100_trying) = self.auto_100_trying {
config.auto_100_trying = auto_100_trying;
}
if let Some(fast_auto_accept) = self.fast_auto_accept_incoming_calls {
config.fast_auto_accept_incoming_calls = fast_auto_accept;
}
if let Some(cleanup_diagnostics) = self.cleanup_diagnostics {
config.cleanup_diagnostics = cleanup_diagnostics;
}
if let Some(cleanup_diagnostic_events) = self.cleanup_diagnostic_events {
config.cleanup_diagnostic_events = cleanup_diagnostic_events;
}
if let Some(capacity) = self.app_event_channel_capacity {
config = config.with_app_event_channel_capacity(capacity);
}
if let Some(capacity) = self.sip_transaction_command_channel_capacity {
config = config.with_sip_transaction_command_channel_capacity(capacity);
}
if let Some(limit) = self.server_call_admission_limit {
config = config.with_server_call_admission_limit(limit);
}
if let Some(limit) = self.server_call_admission_soft_limit {
config = config.with_server_call_admission_soft_limit(limit);
}
if let Some(delay_ms) = self.server_call_admission_pacing_delay_ms {
config = config.with_server_call_admission_pacing_delay_ms(delay_ms);
}
if let Some(seconds) = self.server_overload_retry_after_secs {
config = config.with_server_overload_retry_after_secs(seconds);
}
#[cfg(feature = "perf-tests")]
if let Some(limit) = self.perf_max_rss_growth_mb_per_hr {
config.perf_max_rss_growth_mb_per_hr = Some(limit);
}
if let Some(srtp_diagnostics) = self.srtp_diagnostics {
config.srtp_diagnostics = srtp_diagnostics;
}
if let Some(rtp_diagnostics) = self.rtp_diagnostics {
config.rtp_diagnostics = rtp_diagnostics;
}
if let Some(media_sdp_diagnostics) = self.media_sdp_diagnostics {
config.media_sdp_diagnostics = media_sdp_diagnostics;
}
if self.transport == EndpointTransport::Tls && config.sip_tls_mode == SipTlsMode::Disabled {
config.sip_tls_mode = SipTlsMode::ClientOnly;
}
let derived_from_uri = match (&self.from_uri, &account_username, ®istrar) {
(Some(uri), _, _) => Some(uri.clone()),
(None, Some(username), Some(registrar)) => Some(account_aor_uri(registrar, username)?),
_ => None,
};
if let Some(from_uri) = &derived_from_uri {
config.local_uri = from_uri.clone();
}
if let Some(contact_uri) = &self.contact_uri {
config.contact_uri = Some(contact_uri.clone());
}
for configure in self.configurators {
configure(&mut config);
}
let registration = match (
registrar.as_ref(),
account_username.as_ref(),
self.password.as_ref(),
) {
(Some(registrar), Some(username), Some(password)) => {
let auth_username = self.auth_username.as_deref().unwrap_or(username);
let mut registration = Registration::new(
registrar.clone(),
auth_username.to_string(),
password.clone(),
)
.expires(self.expires);
if let Some(from_uri) = derived_from_uri {
registration = registration.from_uri(from_uri);
}
if let Some(contact_uri) = self.contact_uri {
registration = registration.contact_uri(contact_uri);
}
Some(registration)
}
_ => None,
};
Ok(EndpointParts {
config,
registration,
registrar,
transport: self.transport,
})
}
fn profile_config(&self) -> Result<Config> {
let name = self
.name
.as_deref()
.or(self.account_username.as_deref())
.unwrap_or("endpoint");
match &self.profile {
EndpointProfile::Local => {
let bind = self
.bind_addr
.unwrap_or_else(|| SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5060));
if bind.ip().is_loopback() {
Ok(Config::local(name, bind.port()))
} else {
let mut config = Config::on(name, bind.ip(), bind.port());
config.bind_addr = bind;
Ok(config)
}
}
EndpointProfile::LanPbx => {
let bind = self.bind_addr.unwrap_or_else(default_udp_bind);
let advertised = self.advertised_addr.ok_or_else(|| {
SessionError::ConfigError(
"EndpointProfile::LanPbx requires advertised_addr".to_string(),
)
})?;
Ok(Config::lan_pbx(name, bind, advertised))
}
EndpointProfile::AsteriskUdp => {
let bind = self.bind_addr.unwrap_or_else(default_udp_bind);
if let Some(advertised) = self.advertised_addr {
Ok(Config::lan_pbx(name, bind, advertised))
} else if bind.ip().is_loopback() {
let mut config = Config::local(name, bind.port());
config.bind_addr = bind;
Ok(config)
} else if bind.ip().is_unspecified() {
Err(SessionError::ConfigError(
"EndpointProfile::AsteriskUdp with an unspecified bind address requires advertised_addr"
.to_string(),
))
} else {
let mut config = Config::on(name, bind.ip(), bind.port());
config.bind_addr = bind;
Ok(config)
}
}
EndpointProfile::AsteriskTlsSrtpRegisteredFlow => {
let bind = self.bind_addr.unwrap_or_else(default_tls_bind);
Ok(Config::asterisk_tls_registered_flow(
name,
bind,
self.sip_instance
.clone()
.unwrap_or_else(generate_sip_instance),
))
}
EndpointProfile::FreeSwitchInternal => {
let bind = self.bind_addr.unwrap_or_else(default_udp_bind);
Ok(Config::freeswitch_internal(name, bind))
}
EndpointProfile::FreeSwitchTlsSrtpReachableContact => {
let bind = self.bind_addr.unwrap_or_else(default_udp_bind);
let tls_bind = self.tls_bind_addr.unwrap_or_else(default_tls_bind);
let cert = self.tls_cert_path.clone().ok_or_else(|| {
SessionError::ConfigError(
"EndpointProfile::FreeSwitchTlsSrtpReachableContact requires tls_cert_path"
.to_string(),
)
})?;
let key = self.tls_key_path.clone().ok_or_else(|| {
SessionError::ConfigError(
"EndpointProfile::FreeSwitchTlsSrtpReachableContact requires tls_key_path"
.to_string(),
)
})?;
Ok(Config::freeswitch_tls_srtp_reachable_contact(
name, bind, tls_bind, cert, key,
))
}
EndpointProfile::CarrierSbc => {
let bind = self.bind_addr.unwrap_or_else(default_tls_bind);
let public = self.advertised_addr.ok_or_else(|| {
SessionError::ConfigError(
"EndpointProfile::CarrierSbc requires advertised_addr".to_string(),
)
})?;
let outbound_proxy = self.outbound_proxy_uri.clone().ok_or_else(|| {
SessionError::ConfigError(
"EndpointProfile::CarrierSbc requires outbound_proxy".to_string(),
)
})?;
Ok(Config::carrier_sbc(
name,
bind,
public,
outbound_proxy,
self.sip_instance
.clone()
.unwrap_or_else(generate_sip_instance),
))
}
EndpointProfile::Custom(config) => Ok(config.clone()),
}
}
}
impl Default for EndpointBuilder {
fn default() -> Self {
Self::new()
}
}
struct EndpointParts {
config: Config,
registration: Option<Registration>,
registrar: Option<String>,
transport: EndpointTransport,
}
fn default_udp_bind() -> SocketAddr {
SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 5060)
}
fn default_tls_bind() -> SocketAddr {
SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 5061)
}
fn generate_sip_instance() -> String {
format!("urn:uuid:{}", uuid::Uuid::new_v4())
}
async fn wait_for_registration_result(
events: &mut EndpointEvents,
handle: &RegistrationHandle,
timeout: Option<Duration>,
) -> Result<EndpointRegistrationInfo> {
let coordinator = events.control.coordinator().clone();
let registrar = coordinator
.registration_info(handle)
.await?
.registrar
.unwrap_or_default();
let fut = async {
loop {
match events.next().await? {
Some(EndpointEvent::RegistrationChanged(info))
if registrar.is_empty()
|| info.registrar.as_deref() == Some(registrar.as_str()) =>
{
if info.status == EndpointRegistrationStatus::Registered {
return coordinator
.registration_info(handle)
.await
.map(EndpointRegistrationInfo::from);
}
if info.status == EndpointRegistrationStatus::Failed {
return Err(SessionError::Other(format!(
"registration failed for {}: {}",
info.registrar.unwrap_or_default(),
info.last_failure
.unwrap_or_else(|| "unknown error".to_string())
)));
}
}
Some(_) => {}
None => {
return Err(SessionError::Other(
"event stream closed while waiting for registration".to_string(),
));
}
}
}
};
match timeout {
Some(duration) => tokio::time::timeout(duration, fut)
.await
.map_err(|_| SessionError::Timeout("register_and_wait timed out".to_string()))?,
None => fut.await,
}
}
fn normalize_target(
registrar: Option<&str>,
target: &str,
transport: EndpointTransport,
) -> Result<String> {
let target = target.trim();
if target.is_empty() {
return Err(SessionError::InvalidInput(
"call target must not be empty".to_string(),
));
}
let lower = target.to_ascii_lowercase();
if lower.starts_with("sip:") || lower.starts_with("sips:") || lower.starts_with("tel:") {
return Ok(apply_transport_to_uri(target, transport, false));
}
let registrar = registrar.ok_or_else(|| {
SessionError::ConfigError(
"bare call targets require EndpointBuilder::registrar".to_string(),
)
})?;
let registrar = apply_transport_to_uri(registrar, transport, true);
let mut registrar_uri = parse_uri(®istrar, "registrar")?;
if target.contains('@') {
return Ok(format!("{}:{}", registrar_uri.scheme, target));
}
registrar_uri.user = Some(target.to_string());
registrar_uri.password = None;
registrar_uri.headers.clear();
Ok(registrar_uri.to_string())
}
fn apply_transport_to_uri(
uri: &str,
transport: EndpointTransport,
registrar_or_proxy: bool,
) -> String {
match transport {
EndpointTransport::Udp => uri.to_string(),
EndpointTransport::Tcp => {
if uri.contains(";transport=") {
uri.to_string()
} else {
format!("{uri};transport=tcp")
}
}
EndpointTransport::Tls => {
let tls_uri = if uri.to_ascii_lowercase().starts_with("sip:") {
format!("sips:{}", &uri[4..])
} else {
uri.to_string()
};
if registrar_or_proxy || tls_uri.contains(";transport=") {
tls_uri
} else {
format!("{tls_uri};transport=tls")
}
}
}
}
fn parse_media_public_address(value: &str) -> Result<SocketAddr> {
if let Ok(addr) = value.parse::<SocketAddr>() {
return Ok(addr);
}
let ip = value.parse::<IpAddr>().map_err(|err| {
SessionError::InvalidInput(format!("invalid media public address '{value}': {err}"))
})?;
Ok(SocketAddr::new(ip, 0))
}
fn account_aor_uri(registrar: &str, username: &str) -> Result<String> {
let mut uri = parse_uri(registrar, "registrar")?;
uri.user = Some(username.to_string());
uri.password = None;
uri.port = None;
uri.parameters.clear();
uri.headers.clear();
Ok(uri.to_string())
}
fn parse_uri(value: &str, label: &str) -> Result<Uri> {
let uri = Uri::from_str(value).map_err(|err| {
SessionError::InvalidInput(format!("invalid {label} URI '{value}': {err}"))
})?;
match uri.scheme {
Scheme::Sip | Scheme::Sips => Ok(uri),
_ => Err(SessionError::InvalidInput(format!(
"{label} URI must use sip: or sips:"
))),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::api::unified::{SipContactMode, SipTlsMode};
#[test]
fn endpoint_builder_maps_asterisk_tls_profile() {
let parts = Endpoint::builder()
.name("alice")
.account("1001")
.password("secret")
.registrar("sips:pbx.example.test:5061;transport=tls")
.profile(EndpointProfile::AsteriskTlsSrtpRegisteredFlow)
.sip_instance("urn:uuid:00000000-0000-0000-0000-000000000001")
.build_parts()
.unwrap();
assert_eq!(parts.config.sip_tls_mode, SipTlsMode::ClientOnly);
assert_eq!(
parts.config.sip_contact_mode,
SipContactMode::RegisteredFlowSymmetric
);
assert!(parts.config.offer_srtp);
assert!(parts.config.srtp_required);
assert_eq!(parts.config.local_uri, "sips:1001@pbx.example.test");
assert!(parts.registration.is_some());
}
#[test]
fn endpoint_builder_creates_registration_defaults() {
let parts = Endpoint::builder()
.account("1001")
.auth_username("auth1001")
.password("secret")
.registrar("sip:pbx.example.test")
.contact_uri("sip:1001@192.0.2.10:5060")
.expires(600)
.build_parts()
.unwrap();
let registration = parts.registration.unwrap();
assert_eq!(registration.registrar, "sip:pbx.example.test");
assert_eq!(registration.username, "auth1001");
assert_eq!(registration.password, "secret");
assert_eq!(registration.expires, 600);
assert_eq!(
registration.from_uri.as_deref(),
Some("sip:1001@pbx.example.test")
);
assert_eq!(
registration.contact_uri.as_deref(),
Some("sip:1001@192.0.2.10:5060")
);
}
#[test]
fn sip_account_derives_compatible_registration_endpoint_account_and_credentials() {
let account = SipAccount::new("sip:pbx.example.test", "1001", "secret")
.auth_username("auth1001")
.expires(600)
.from_uri("sip:1001@pbx.example.test")
.contact_uri("sip:1001@192.0.2.10:5060");
let credentials = account.credentials();
assert_eq!(credentials.username, "auth1001");
assert_eq!(credentials.password, "secret");
let registration = account.registration();
assert_eq!(registration.registrar, "sip:pbx.example.test");
assert_eq!(registration.username, "auth1001");
assert_eq!(registration.password, "secret");
assert_eq!(registration.expires, 600);
assert_eq!(
registration.from_uri.as_deref(),
Some("sip:1001@pbx.example.test")
);
assert_eq!(
registration.contact_uri.as_deref(),
Some("sip:1001@192.0.2.10:5060")
);
let endpoint_account = account.endpoint_account();
assert_eq!(endpoint_account.username, "1001");
assert_eq!(endpoint_account.auth_username.as_deref(), Some("auth1001"));
let legacy: EndpointAccount = account.clone().into();
let round_trip: SipAccount = legacy.into();
assert_eq!(round_trip.effective_auth_username(), "auth1001");
assert_eq!(round_trip.username, "1001");
}
#[test]
fn endpoint_normalizes_bare_extension_through_registrar() {
let target = normalize_target(
Some("sips:pbx.example.test:5061;transport=tls"),
"1002",
EndpointTransport::Udp,
)
.unwrap();
assert_eq!(target, "sips:1002@pbx.example.test:5061;transport=tls");
}
#[test]
fn endpoint_leaves_full_sip_uri_unchanged() {
let target = normalize_target(
Some("sips:pbx.example.test:5061"),
"sip:bob@example.test",
EndpointTransport::Udp,
)
.unwrap();
assert_eq!(target, "sip:bob@example.test");
}
#[test]
fn endpoint_requires_registrar_for_bare_target() {
let err = normalize_target(None, "1002", EndpointTransport::Udp).unwrap_err();
assert!(err.to_string().contains("registrar"));
}
#[test]
fn endpoint_transport_rewrites_tls_target() {
let target = normalize_target(
Some("sip:pbx.example.test:5060"),
"1002",
EndpointTransport::Tls,
)
.unwrap();
assert_eq!(target, "sips:1002@pbx.example.test:5060");
}
#[test]
fn endpoint_json_config_maps_builder_fields() {
let config = serde_json::from_str::<EndpointConfig>(
r#"{
"name": "alice",
"profile": "asterisk-udp",
"auto180Ringing": false,
"auto100Trying": false,
"fastAutoAcceptIncomingCalls": true,
"cleanupDiagnostics": true,
"cleanupDiagnosticEvents": true,
"appEventChannelCapacity": 512,
"srtpDiagnostics": true,
"rtpDiagnostics": true,
"mediaSdpDiagnostics": true,
"account": {
"username": "1001",
"password": "secret",
"registrar": "sip:pbx.example.test"
},
"network": {
"bind": "127.0.0.1:5060",
"transport": "tcp",
"stun": "stun.example.test:3478",
"udpParseWorkers": 4,
"udpParseQueueCapacity": 8192
},
"media": {
"publicAddress": "192.0.2.10",
"enabled": false,
"signalingOnlyRtpPort": 9,
"srtp": "offer"
}
}"#,
)
.unwrap();
let parts = EndpointBuilder::from_config(config)
.unwrap()
.build_parts()
.unwrap();
assert_eq!(parts.transport, EndpointTransport::Tcp);
assert_eq!(
parts.config.stun_server.as_deref(),
Some("stun.example.test:3478")
);
assert!(parts.config.offer_srtp);
assert!(!parts.config.srtp_required);
assert!(!parts.config.auto_180_ringing);
assert!(!parts.config.auto_100_trying);
assert!(parts.config.fast_auto_accept_incoming_calls);
assert!(parts.config.cleanup_diagnostics);
assert!(parts.config.cleanup_diagnostic_events);
assert_eq!(parts.config.global_event_channel_capacity, 512);
assert_eq!(parts.config.session_event_dispatcher_channel_capacity, 512);
assert!(parts.config.srtp_diagnostics);
assert!(parts.config.rtp_diagnostics);
assert!(parts.config.media_sdp_diagnostics);
assert_eq!(parts.config.sip_udp_parse_workers, Some(4));
assert_eq!(parts.config.sip_udp_parse_queue_capacity, Some(8192));
assert_eq!(
parts.config.media_mode,
MediaMode::SignalingOnly { sdp_rtp_port: 9 }
);
assert_eq!(
parts.config.media_public_addr,
Some("192.0.2.10:0".parse().unwrap())
);
assert_eq!(
parts.registrar.as_deref(),
Some("sip:pbx.example.test;transport=tcp")
);
}
#[test]
fn endpoint_json_performance_profile_maps_into_config() {
let config = serde_json::from_str::<EndpointConfig>(
r#"{
"name": "perf",
"performance": {
"profile": "pbx-media-server",
"capacity": 2000
},
"network": {
"udpParseWorkers": 2
},
"sipTransactionCommandChannelCapacity": 256,
"serverCallAdmissionLimit": 3000,
"serverCallAdmissionSoftLimit": 2500,
"serverCallAdmissionPacingDelayMs": 3,
"serverOverloadRetryAfterSecs": 2
}"#,
)
.unwrap();
let parts = EndpointBuilder::from_config(config)
.unwrap()
.build_parts()
.unwrap();
assert!(parts.config.fast_auto_accept_incoming_calls);
assert_eq!(parts.config.media_mode, MediaMode::Enabled);
assert_eq!(parts.config.media_port_start, 16_384);
assert_eq!(parts.config.media_port_capacity, Some(49_152));
assert_eq!(parts.config.media_session_capacity, Some(2_000));
assert_eq!(parts.config.sip_udp_parse_workers, Some(2));
assert_eq!(
parts.config.sip_udp_parse_dispatch,
Some(rvoip_sip_transport::UdpParseDispatch::RoundRobin)
);
assert_eq!(
parts.config.sip_transaction_command_channel_capacity,
Some(256)
);
assert_eq!(parts.config.server_call_capacity, Some(2_000));
assert_eq!(parts.config.server_call_admission_limit, Some(3_000));
assert_eq!(parts.config.server_call_admission_soft_limit, Some(2_500));
assert_eq!(parts.config.server_call_admission_pacing_delay_ms, Some(3));
assert_eq!(parts.config.server_overload_retry_after_secs, Some(2));
}
#[test]
fn endpoint_json_endpoint_performance_recipe_is_default_shape() {
let config = serde_json::from_str::<EndpointConfig>(
r#"{
"name": "softphone",
"performance": {
"profile": "endpoint"
}
}"#,
)
.unwrap();
let parts = EndpointBuilder::from_config(config)
.unwrap()
.build_parts()
.unwrap();
assert!(parts.config.auto_180_ringing);
assert!(parts.config.auto_100_trying);
assert!(!parts.config.fast_auto_accept_incoming_calls);
assert_eq!(parts.config.media_mode, MediaMode::Enabled);
assert_eq!(parts.config.sip_udp_parse_workers, None);
assert_eq!(parts.config.sip_transaction_command_channel_capacity, None);
}
#[test]
fn endpoint_json_signaling_only_performance_profile_maps_into_config() {
let config = serde_json::from_str::<EndpointConfig>(
r#"{
"name": "perf",
"performance": {
"profile": "signaling-only-server-high-performance",
"capacity": 2000,
"signalingOnlyRtpPort": 4000
}
}"#,
)
.unwrap();
let parts = EndpointBuilder::from_config(config)
.unwrap()
.build_parts()
.unwrap();
assert_eq!(
parts.config.media_mode,
MediaMode::SignalingOnly { sdp_rtp_port: 4000 }
);
assert_eq!(parts.config.sip_udp_parse_workers, Some(4));
assert_eq!(
parts.config.sip_transaction_command_channel_capacity,
Some(128)
);
assert_eq!(parts.config.server_call_capacity, Some(2_000));
assert_eq!(parts.config.server_call_admission_limit, Some(2_000));
assert_eq!(parts.config.server_call_admission_soft_limit, Some(1_800));
assert_eq!(parts.config.server_call_admission_pacing_delay_ms, Some(1));
}
#[test]
fn endpoint_json_config_accepts_partial_sip_trace_config() {
let config = serde_json::from_str::<EndpointConfig>(
r#"{
"sipTrace": {
"enabled": true,
"redactSensitiveHeaders": false
}
}"#,
)
.unwrap();
let trace = config.sip_trace.unwrap();
assert!(trace.enabled);
assert_eq!(
trace.capacity,
crate::api::events::SipTraceConfig::DEFAULT_CAPACITY
);
assert!(!trace.redact_sensitive_headers);
assert!(trace.include_body);
}
#[test]
fn sip_client_example_stays_on_endpoint_surface() {
let source = [
include_str!("../../examples/sip_client/main.rs"),
include_str!("../../examples/sip_client/audio.rs"),
include_str!("../../examples/sip_client/config.rs"),
include_str!("../../examples/sip_client/runtime.rs"),
include_str!("../../examples/sip_client/smoke.rs"),
include_str!("../../examples/sip_client/ui.rs"),
]
.join("\n");
for banned in [
"StreamPeer",
"PeerControl",
"UnifiedCoordinator",
"RegistrationHandle",
"SessionHandle",
"SipTlsMode",
"rvoip_media_core",
] {
assert!(
!source.contains(banned),
"sip_client example must not reference lower-level API {banned}"
);
}
}
#[test]
fn endpoint_audio_roundtrip_stays_on_endpoint_surface() {
let source = include_str!("../../examples/endpoint/04_audio_roundtrip/main.rs");
for banned in [
"StreamPeer",
"PeerControl",
"UnifiedCoordinator",
"RegistrationHandle",
"SessionHandle",
"as_session_handle",
"rvoip_media_core",
] {
assert!(
!source.contains(banned),
"endpoint audio roundtrip example must not reference lower-level API {banned}"
);
}
}
}