1use crate::{
2 call::{ActiveCallRef, sip::Invitation},
3 callrecord::{
4 CallRecordFormatter, CallRecordManagerBuilder, CallRecordSender, DefaultCallRecordFormatter,
5 },
6 config::Config,
7 locator::RewriteTargetLocator,
8 useragent::{
9 RegisterOption,
10 invitation::{
11 FnCreateInvitationHandler, PendingDialog, PendingDialogGuard,
12 default_create_invite_handler,
13 },
14 public_address::{
15 LearningMessageInspector, SharedPublicAddress, build_contact, build_public_contact_uri,
16 find_local_addr_for_uri,
17 },
18 registration::{RegistrationHandle, UserCredential},
19 },
20};
21
22use crate::media::{cache::set_cache_dir, engine::StreamEngine};
23use anyhow::Result;
24use arc_swap::ArcSwap;
25use chrono::{DateTime, Local};
26use futures::FutureExt;
27use humantime::parse_duration;
28use rsipstack::rsip::prelude::HeadersExt;
29use rsipstack::transaction::{
30 Endpoint, TransactionReceiver,
31 endpoint::{TargetLocator, TransportEventInspector},
32};
33use rsipstack::{dialog::dialog_layer::DialogLayer, transaction::endpoint::MessageInspector};
34use std::future::pending;
35use std::str::FromStr;
36use std::sync::{Arc, RwLock};
37use std::time::Duration;
38use std::{collections::HashMap, net::SocketAddr};
39use std::{collections::HashSet, time::Instant};
40use std::{
41 path::Path,
42 sync::atomic::{AtomicBool, AtomicU64, Ordering},
43};
44use tokio::select;
45use tokio::sync::Mutex;
46use tokio_util::sync::CancellationToken;
47use tracing::{info, warn};
48
49pub struct AppStateInner {
50 pub config: Arc<Config>,
51 pub token: CancellationToken,
52 pub stream_engine: Arc<StreamEngine>,
53 pub callrecord_sender: Option<CallRecordSender>,
54 pub endpoint: Endpoint,
55 pub registration_handles: Mutex<HashMap<String, CancellationToken>>,
56 pub alive_users: Arc<RwLock<HashSet<String>>>,
57 pub dialog_layer: Arc<DialogLayer>,
58 pub create_invitation_handler: Option<FnCreateInvitationHandler>,
59 pub invitation: Invitation,
60 pub routing_state: Arc<crate::call::RoutingState>,
61 pub pending_playbooks: Arc<Mutex<HashMap<String, (String, Instant)>>>,
62 pub learned_public_address: SharedPublicAddress,
63
64 pub active_calls: Arc<std::sync::Mutex<HashMap<String, ActiveCallRef>>>,
65 pub total_calls: AtomicU64,
66 pub total_failed_calls: AtomicU64,
67 pub uptime: DateTime<Local>,
68 pub shutting_down: Arc<AtomicBool>,
69}
70
71pub type AppState = Arc<AppStateInner>;
72
73pub struct AppStateBuilder {
74 pub config: Option<Config>,
75 pub stream_engine: Option<Arc<StreamEngine>>,
76 pub callrecord_sender: Option<CallRecordSender>,
77 pub callrecord_formatter: Option<Arc<dyn CallRecordFormatter>>,
78 pub cancel_token: Option<CancellationToken>,
79 pub create_invitation_handler: Option<FnCreateInvitationHandler>,
80 pub config_path: Option<String>,
81
82 pub message_inspector: Option<Box<dyn MessageInspector>>,
83 pub target_locator: Option<Box<dyn TargetLocator>>,
84 pub transport_inspector: Option<Box<dyn TransportEventInspector>>,
85}
86
87impl AppStateInner {
88 pub fn auto_learn_public_address_enabled(&self) -> bool {
89 self.config.auto_learn_public_address.unwrap_or(false)
90 }
91
92 pub fn get_dump_events_file(&self, session_id: &String) -> String {
93 let recorder_root = self.config.recorder_path();
94 let root = Path::new(&recorder_root);
95 if !root.exists() {
96 match std::fs::create_dir_all(root) {
97 Ok(_) => {
98 info!("created dump events root: {}", root.to_string_lossy());
99 }
100 Err(e) => {
101 warn!(
102 "Failed to create dump events root: {} {}",
103 e,
104 root.to_string_lossy()
105 );
106 }
107 }
108 }
109 root.join(format!("{}.events.jsonl", session_id))
110 .to_string_lossy()
111 .to_string()
112 }
113
114 pub fn get_recorder_file(&self, session_id: &String) -> String {
115 let recorder_root = self.config.recorder_path();
116 let root = Path::new(&recorder_root);
117 if !root.exists() {
118 match std::fs::create_dir_all(root) {
119 Ok(_) => {
120 info!("created recorder root: {}", root.to_string_lossy());
121 }
122 Err(e) => {
123 warn!(
124 "Failed to create recorder root: {} {}",
125 e,
126 root.to_string_lossy()
127 );
128 }
129 }
130 }
131 let desired_ext = self.config.recorder_format().extension();
132 let mut filename = session_id.clone();
133 if !filename
134 .to_lowercase()
135 .ends_with(&format!(".{}", desired_ext.to_lowercase()))
136 {
137 filename = format!("{}.{}", filename, desired_ext);
138 }
139 root.join(filename).to_string_lossy().to_string()
140 }
141
142 pub async fn serve(self: Arc<Self>) -> Result<()> {
143 let incoming_txs = self.endpoint.incoming_transactions()?;
144 let token = self.token.child_token();
145 let endpoint_inner = self.endpoint.inner.clone();
146 let dialog_layer = self.dialog_layer.clone();
147 let app_state_clone = self.clone();
148
149 match self.start_registration().await {
150 Ok(count) => {
151 info!("registration started, count: {}", count);
152 }
153 Err(e) => {
154 warn!("failed to start registration: {:?}", e);
155 }
156 }
157
158 let pending_cleanup_state = self.clone();
159 let pending_cleanup_token = token.clone();
160 crate::spawn(async move {
161 let mut interval = tokio::time::interval(Duration::from_secs(60));
162 let ttl = Duration::from_secs(300);
163 loop {
164 tokio::select! {
165 _ = pending_cleanup_token.cancelled() => break,
166 _ = interval.tick() => {
167 let mut pending = pending_cleanup_state.pending_playbooks.lock().await;
168 let before = pending.len();
169 pending.retain(|_, (_, created_at)| created_at.elapsed() < ttl);
170 let removed = before - pending.len();
171 if removed > 0 {
172 info!(removed, remaining = pending.len(), "cleaned up stale pending_playbooks entries");
173 }
174 }
175 }
176 }
177 });
178
179 tokio::select! {
180 _ = token.cancelled() => {
181 info!("cancelled");
182 }
183 result = endpoint_inner.serve() => {
184 if let Err(e) = result {
185 info!("endpoint serve error: {:?}", e);
186 }
187 }
188 result = app_state_clone.process_incoming_request(dialog_layer.clone(), incoming_txs) => {
189 if let Err(e) = result {
190 info!("process incoming request error: {:?}", e);
191 }
192 },
193 }
194
195 let timeout = self
198 .config
199 .graceful_shutdown
200 .map(|_| Duration::from_secs(10));
201
202 match self.stop_registration(timeout).await {
203 Ok(_) => {
204 info!("registration stopped, waiting for clear");
205 }
206 Err(e) => {
207 warn!("failed to stop registration: {:?}", e);
208 }
209 }
210 info!("stopping");
211 Ok(())
212 }
213
214 async fn process_incoming_request(
215 self: Arc<Self>,
216 dialog_layer: Arc<DialogLayer>,
217 mut incoming: TransactionReceiver,
218 ) -> Result<()> {
219 while let Some(mut tx) = incoming.recv().await {
220 let key: &rsipstack::transaction::key::TransactionKey = &tx.key;
221 info!(?key, "received transaction");
222 if tx.original.to_header()?.tag()?.as_ref().is_some() {
223 match dialog_layer.match_dialog(&tx) {
224 Some(mut d) => {
225 crate::spawn(async move {
226 match d.handle(&mut tx).await {
227 Ok(_) => (),
228 Err(e) => {
229 info!("error handling transaction: {:?}", e);
230 }
231 }
232 });
233 continue;
234 }
235 None => {
236 info!("dialog not found: {}", tx.original);
237 match tx
238 .reply(rsipstack::rsip::StatusCode::CallTransactionDoesNotExist)
239 .await
240 {
241 Ok(_) => (),
242 Err(e) => {
243 info!("error replying to request: {:?}", e);
244 }
245 }
246 continue;
247 }
248 }
249 }
250 let (state_sender, state_receiver) = dialog_layer.new_dialog_state_channel();
252 match tx.original.method {
253 rsipstack::rsip::Method::Invite | rsipstack::rsip::Method::Ack => {
254 if self.shutting_down.load(Ordering::Relaxed) {
256 info!(?key, "rejecting INVITE during graceful shutdown");
257 match tx
258 .reply_with(
259 rsipstack::rsip::StatusCode::ServiceUnavailable,
260 vec![rsipstack::rsip::Header::Other(
261 "Reason".into(),
262 "SIP;cause=503;text=\"Server shutting down\"".into(),
263 )],
264 None,
265 )
266 .await
267 {
268 Ok(_) => (),
269 Err(e) => {
270 info!("error replying to request: {:?}", e);
271 }
272 }
273 continue;
274 }
275
276 let invitation_handler = match self.create_invitation_handler {
277 Some(ref create_invitation_handler) => {
278 create_invitation_handler(self.config.handler.as_ref()).ok()
279 }
280 _ => default_create_invite_handler(
281 self.config.handler.as_ref(),
282 Some(self.clone()),
283 ),
284 };
285 let invitation_handler = match invitation_handler {
286 Some(h) => h,
287 None => {
288 info!(?key, "no invite handler configured, rejecting INVITE");
289 match tx
290 .reply_with(
291 rsipstack::rsip::StatusCode::ServiceUnavailable,
292 vec![rsipstack::rsip::Header::Other(
293 "Reason".into(),
294 "SIP;cause=503;text=\"No invite handler configured\""
295 .into(),
296 )],
297 None,
298 )
299 .await
300 {
301 Ok(_) => (),
302 Err(e) => {
303 info!("error replying to request: {:?}", e);
304 }
305 }
306 continue;
307 }
308 };
309 let local_addr = tx
310 .connection
311 .as_ref()
312 .map(|connection| connection.get_addr().clone())
313 .or_else(|| dialog_layer.endpoint.get_addrs().first().cloned());
314 let contact_username =
315 tx.original.uri.auth.as_ref().map(|auth| auth.user.as_str());
316 let contact = local_addr.as_ref().map(|addr| {
317 build_public_contact_uri(
318 &self.learned_public_address,
319 self.auto_learn_public_address_enabled(),
320 addr,
321 contact_username,
322 None,
323 )
324 });
325
326 let dialog = match dialog_layer.get_or_create_server_invite(
327 &tx,
328 state_sender,
329 None,
330 contact,
331 ) {
332 Ok(d) => d,
333 Err(e) => {
334 info!("failed to obtain dialog: {:?}", e);
336 match tx
337 .reply(rsipstack::rsip::StatusCode::CallTransactionDoesNotExist)
338 .await
339 {
340 Ok(_) => (),
341 Err(e) => {
342 info!("error replying to request: {:?}", e);
343 }
344 }
345 continue;
346 }
347 };
348
349 let dialog_id = dialog.id();
350 let dialog_id_str = dialog_id.to_string();
351 let token = self.token.child_token();
352 let pending_dialog = PendingDialog {
353 token: token.clone(),
354 dialog: dialog.clone(),
355 state_receiver,
356 };
357
358 let guard = Arc::new(PendingDialogGuard::new(
359 self.invitation.clone(),
360 dialog_id,
361 pending_dialog,
362 ));
363
364 let accept_timeout = self
365 .config
366 .accept_timeout
367 .as_ref()
368 .and_then(|t| parse_duration(t).ok())
369 .unwrap_or_else(|| Duration::from_secs(60));
370
371 let token_ref = token.clone();
372 let guard_ref = guard.clone();
373 crate::spawn(async move {
374 select! {
375 _ = token_ref.cancelled() => {}
376 _ = tokio::time::sleep(accept_timeout) => {}
377 }
378 guard_ref.drop_async().await;
379 });
380
381 let mut dialog_ref = dialog.clone();
382 let token_ref = token.clone();
383 let routing_state = self.routing_state.clone();
384 let dialog_for_reject = dialog.clone();
385 let guard_ref = guard.clone();
386 crate::spawn(async move {
387 let invite_loop = async {
388 match invitation_handler
389 .on_invite(
390 dialog_id_str.clone(),
391 token.clone(),
392 dialog.clone(),
393 routing_state,
394 )
395 .await
396 {
397 Ok(_) => (),
398 Err(e) => {
399 info!(id = dialog_id_str, "error handling invite: {:?}", e);
401 let reason = format!("Failed to process invite: {}", e);
402 if let Err(reject_err) = dialog_for_reject.reject(
403 Some(rsipstack::rsip::StatusCode::ServiceUnavailable),
404 Some(reason),
405 ) {
406 info!(
407 id = dialog_id_str,
408 "error rejecting call: {:?}", reject_err
409 );
410 }
411 token.cancel();
413 guard_ref.drop_async().await;
414 }
415 }
416 };
417 select! {
418 _ = token_ref.cancelled() => {}
419 _ = async {
420 let (_,_ ) = tokio::join!(dialog_ref.handle(&mut tx), invite_loop);
421 } => {}
422 }
423 });
424 }
425 rsipstack::rsip::Method::Options => {
426 info!(?key, "ignoring out-of-dialog OPTIONS request");
427 continue;
428 }
429 rsipstack::rsip::Method::Refer => {
430 info!(?key, "ignoring out-of-dialog REFER");
431 match tx.reply(rsipstack::rsip::StatusCode::BadRequest).await {
432 Ok(_) => (),
433 Err(e) => {
434 info!("error replying to out-of-dialog REFER: {:?}", e);
435 }
436 }
437 continue;
438 }
439 _ => {
440 info!(?key, "received request: {:?}", tx.original.method);
441 match tx.reply(rsipstack::rsip::StatusCode::OK).await {
442 Ok(_) => (),
443 Err(e) => {
444 info!("error replying to request: {:?}", e);
445 }
446 }
447 }
448 }
449 }
450 Ok(())
451 }
452
453 pub fn stop(&self) {
454 if self.shutting_down.swap(true, Ordering::Relaxed) {
455 return;
456 }
457 info!("stopping, marking as shutting down");
458 self.token.cancel();
459 }
460
461 pub async fn graceful_stop(&self) -> Result<()> {
462 if self.shutting_down.swap(true, Ordering::Relaxed) {
463 return Ok(());
464 }
465
466 info!("graceful stopping, marking as shutting down");
467 let timeout = self
468 .config
469 .graceful_shutdown
470 .map(|_| Duration::from_secs(10));
471
472 self.stop_registration(timeout).await?;
473 self.token.cancel();
474 Ok(())
475 }
476
477 pub async fn start_registration(&self) -> Result<usize> {
478 let mut count = 0;
479 if let Some(register_users) = &self.config.register_users {
480 for option in register_users.iter() {
481 match self.register(option.clone()).await {
482 Ok(_) => {
483 count += 1;
484 }
485 Err(e) => {
486 warn!("failed to register user: {:?} {:?}", e, option);
487 }
488 }
489 }
490 }
491 Ok(count)
492 }
493
494 pub fn find_credentials_for_callee(&self, callee: &str) -> Option<UserCredential> {
495 let callee_uri = callee
496 .strip_prefix("sip:")
497 .or_else(|| callee.strip_prefix("sips:"))
498 .unwrap_or(callee);
499 let callee_uri = if !callee_uri.starts_with("sip:") && !callee_uri.starts_with("sips:") {
500 format!("sip:{}", callee_uri)
501 } else {
502 callee_uri.to_string()
503 };
504
505 let parsed_callee = match rsipstack::rsip::Uri::try_from(callee_uri.as_str()) {
506 Ok(uri) => uri,
507 Err(e) => {
508 warn!("failed to parse callee URI: {} {:?}", callee, e);
509 return None;
510 }
511 };
512
513 let callee_host = match &parsed_callee.host_with_port.host {
514 rsipstack::rsip::Host::Domain(domain) => domain.to_string(),
515 rsipstack::rsip::Host::IpAddr(ip) => return self.find_credentials_by_ip(ip),
516 };
517
518 if let Some(register_users) = &self.config.register_users {
520 for option in register_users.iter() {
521 let mut server = option.server.clone();
522 if !server.starts_with("sip:") && !server.starts_with("sips:") {
523 server = format!("sip:{}", server);
524 }
525
526 let parsed_server = match rsipstack::rsip::Uri::try_from(server.as_str()) {
527 Ok(uri) => uri,
528 Err(e) => {
529 warn!("failed to parse server URI: {} {:?}", option.server, e);
530 continue;
531 }
532 };
533
534 let server_host = match &parsed_server.host_with_port.host {
535 rsipstack::rsip::Host::Domain(domain) => domain.to_string(),
536 rsipstack::rsip::Host::IpAddr(ip) => {
537 if let rsipstack::rsip::Host::IpAddr(callee_ip) = &parsed_callee.host_with_port.host {
539 if ip == callee_ip {
540 if let Some(cred) = &option.credential {
541 info!(
542 callee,
543 username = cred.username,
544 server = option.server,
545 "Auto-injecting credentials from registered user for outbound call (IP match)"
546 );
547 return Some(cred.clone());
548 }
549 }
550 }
551 continue;
552 }
553 };
554
555 if server_host == callee_host {
556 if let Some(cred) = &option.credential {
557 info!(
558 callee,
559 username = cred.username,
560 server = option.server,
561 "Auto-injecting credentials from registered user for outbound call"
562 );
563 return Some(cred.clone());
564 }
565 }
566 }
567 }
568
569 None
570 }
571
572 fn find_credentials_by_ip(
574 &self,
575 callee_ip: &std::net::IpAddr,
576 ) -> Option<crate::useragent::registration::UserCredential> {
577 if let Some(register_users) = &self.config.register_users {
578 for option in register_users.iter() {
579 let mut server = option.server.clone();
580 if !server.starts_with("sip:") && !server.starts_with("sips:") {
581 server = format!("sip:{}", server);
582 }
583
584 if let Ok(parsed_server) = rsipstack::rsip::Uri::try_from(server.as_str()) {
585 if let rsipstack::rsip::Host::IpAddr(server_ip) = &parsed_server.host_with_port.host {
586 if server_ip == callee_ip {
587 if let Some(cred) = &option.credential {
588 info!(
589 callee_ip = %callee_ip,
590 username = cred.username,
591 server = option.server,
592 "Auto-injecting credentials from registered user for outbound call (IP match)"
593 );
594 return Some(cred.clone());
595 }
596 }
597 }
598 }
599 }
600 }
601 None
602 }
603
604 pub async fn stop_registration(&self, wait_for_clear: Option<Duration>) -> Result<()> {
605 {
606 let mut handles = self.registration_handles.lock().await;
607 for (_, cancel_token) in handles.drain() {
608 cancel_token.cancel();
609 }
610 }
611
612 if let Some(duration) = wait_for_clear {
613 let live_users = self.alive_users.clone();
614 let check_loop = async move {
615 loop {
616 let is_empty = {
617 let users = live_users
618 .read()
619 .map_err(|_| anyhow::anyhow!("Lock poisoned"))?;
620 users.is_empty()
621 };
622 if is_empty {
623 break;
624 }
625 tokio::time::sleep(Duration::from_millis(50)).await;
626 }
627 Ok::<(), anyhow::Error>(())
628 };
629 match tokio::time::timeout(duration, check_loop).await {
630 Ok(_) => {}
631 Err(e) => {
632 warn!("failed to wait for clear: {}", e);
633 return Err(anyhow::anyhow!("failed to wait for clear: {}", e));
634 }
635 }
636 }
637 Ok(())
638 }
639
640 pub async fn register(&self, option: RegisterOption) -> Result<()> {
641 let user = option.aor();
642 let mut server = option.server.clone();
643 if !server.starts_with("sip:") && !server.starts_with("sips:") {
644 server = format!("sip:{}", server);
645 }
646 let sip_server = match rsipstack::rsip::Uri::try_from(server) {
647 Ok(uri) => uri,
648 Err(e) => {
649 warn!("failed to parse server: {} {:?}", e, option.server);
650 return Err(anyhow::anyhow!("failed to parse server: {}", e));
651 }
652 };
653 let cancel_token = self.token.child_token();
654 let credential = option.credential.clone().map(|c| c.into());
655 let registration = rsipstack::dialog::registration::Registration::new(
656 self.endpoint.inner.clone(),
657 credential,
658 );
659 let mut handle = RegistrationHandle {
660 registration,
661 option,
662 cancel_token: cancel_token.clone(),
663 start_time: Instant::now(),
664 last_update: Instant::now(),
665 last_response: None,
666 };
667 self.registration_handles
668 .lock()
669 .await
670 .insert(user.clone(), cancel_token);
671 tracing::debug!(user = user.as_str(), "starting registration task");
672 let alive_users = self.alive_users.clone();
673
674 crate::spawn(async move {
675 handle.start_time = Instant::now();
676 let cancel_token = handle.cancel_token.clone();
677 let addrs = handle.registration.endpoint.get_addrs();
678 let local_bind_addr = if let Some(addr) = find_local_addr_for_uri(&addrs, &sip_server) {
679 addr
680 } else {
681 warn!(
682 user = user.as_str(),
683 server = %sip_server,
684 "failed to get local bind address for registration transport"
685 );
686 alive_users.write().unwrap().remove(&user);
687 return;
688 };
689 let user = handle.option.aor();
690 alive_users.write().unwrap().remove(&user);
691 let mut contact_address = local_bind_addr.addr.clone();
692 let mut contact = build_contact(
693 &local_bind_addr,
694 Some(contact_address.clone()),
695 Some(handle.option.username.as_str()),
696 None,
697 );
698 let mut should_register = true;
699 let mut timer = pending().boxed();
700
701 loop {
702 select! {
703 _ = cancel_token.cancelled() => {
704 break;
705 }
706 _ = timer.as_mut(), if !should_register => {
707 should_register = true;
708 timer = Box::pin(pending());
709 }
710 result = handle.do_register(&sip_server, None, &contact), if should_register => {
711 match result {
712 Ok((expires, new_addr)) => {
713 if handle
714 .should_retry_registration_now(
715 &local_bind_addr,
716 &contact_address,
717 new_addr.as_ref(),
718 )
719 {
720 if let Some(next_contact_address) = new_addr {
721 info!(
722 user = user.as_str(),
723 current_contact = %contact_address,
724 next_contact = %next_contact_address,
725 "public address changed, retrying registration immediately",
726 );
727 contact_address = next_contact_address;
728 contact = build_contact(
729 &local_bind_addr,
730 Some(contact_address.clone()),
731 Some(handle.option.username.as_str()),
732 None,
733 );
734 continue;
735 }
736 }
737 info!(
738 user = user.as_str(),
739 expires = expires,
740 contact = %contact_address,
741 alive_users = alive_users.read().unwrap().len(),
742 "registration refreshed",
743 );
744 alive_users.write().unwrap().insert(user.clone());
745 should_register = false;
746 timer = Box::pin(tokio::time::sleep(Duration::from_secs(
747 (expires * 3 / 4) as u64,
748 )));
749 }
750 Err(e) => {
751 warn!(
752 user = user.as_str(),
753 alive_users = alive_users.read().unwrap().len(),
754 "registration failed: {:?}", e
755 );
756 should_register = false;
757 timer = Box::pin(tokio::time::sleep(Duration::from_secs(60)));
758 }
759 }
760 }
761 }
762 }
763 handle
764 .do_register(&sip_server, Some(0), &contact)
765 .await
766 .ok();
767 alive_users.write().unwrap().remove(&user);
768 });
769 Ok(())
770 }
771}
772
773impl Drop for AppStateInner {
774 fn drop(&mut self) {
775 self.stop();
776 }
777}
778
779impl AppStateBuilder {
780 pub fn new() -> Self {
781 Self {
782 config: None,
783 stream_engine: None,
784 callrecord_sender: None,
785 callrecord_formatter: None,
786 cancel_token: None,
787 create_invitation_handler: None,
788 config_path: None,
789 message_inspector: None,
790 target_locator: None,
791 transport_inspector: None,
792 }
793 }
794
795 pub fn with_config(mut self, config: Config) -> Self {
796 self.config = Some(config);
797 self
798 }
799
800 pub fn with_stream_engine(mut self, stream_engine: Arc<StreamEngine>) -> Self {
801 self.stream_engine = Some(stream_engine);
802 self
803 }
804
805 pub fn with_callrecord_sender(mut self, sender: CallRecordSender) -> Self {
806 self.callrecord_sender = Some(sender);
807 self
808 }
809
810 pub fn with_cancel_token(mut self, token: CancellationToken) -> Self {
811 self.cancel_token = Some(token);
812 self
813 }
814
815 pub fn with_config_metadata(mut self, path: Option<String>) -> Self {
816 self.config_path = path;
817 self
818 }
819
820 pub fn with_inspector(&mut self, inspector: Box<dyn MessageInspector>) -> &mut Self {
821 self.message_inspector = Some(inspector);
822 self
823 }
824 pub fn with_target_locator(&mut self, locator: Box<dyn TargetLocator>) -> &mut Self {
825 self.target_locator = Some(locator);
826 self
827 }
828
829 pub fn with_transport_inspector(
830 &mut self,
831 inspector: Box<dyn TransportEventInspector>,
832 ) -> &mut Self {
833 self.transport_inspector = Some(inspector);
834 self
835 }
836
837 pub async fn build(self) -> Result<AppState> {
838 let config: Arc<Config> = Arc::new(self.config.unwrap_or_default());
839 let token = self
840 .cancel_token
841 .unwrap_or_else(|| CancellationToken::new());
842 let _ = set_cache_dir(&config.media_cache_path);
843 let local_ip = if !config.addr.is_empty() {
844 std::net::IpAddr::from_str(config.addr.as_str())?
845 } else {
846 crate::net_tool::get_first_non_loopback_interface()?
847 };
848 let transport_layer = rsipstack::transport::TransportLayer::new(token.clone());
849 let local_addr: SocketAddr = format!("{}:{}", local_ip, config.udp_port).parse()?;
850
851 #[cfg(unix)]
853 let std_socket = {
854 use socket2::{Domain, Protocol, SockAddr, Socket, Type};
855
856 let domain = if local_addr.is_ipv4() {
857 Domain::IPV4
858 } else {
859 Domain::IPV6
860 };
861 let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))
862 .map_err(|err| anyhow::anyhow!("Failed to create UDP socket: {}", err))?;
863
864 socket
865 .set_reuse_address(true)
866 .map_err(|err| anyhow::anyhow!("Failed to set SO_REUSEADDR: {}", err))?;
867
868 #[cfg(not(any(target_os = "solaris", target_os = "illumos", target_os = "cygwin")))]
870 socket
871 .set_reuse_port(true)
872 .map_err(|err| anyhow::anyhow!("Failed to set SO_REUSEPORT: {}", err))?;
873
874 socket
875 .bind(&SockAddr::from(local_addr))
876 .map_err(|err| anyhow::anyhow!("Failed to bind UDP socket: {}", err))?;
877
878 let std_socket: std::net::UdpSocket = socket.into();
879 std_socket
880 };
881
882 #[cfg(not(unix))]
883 let std_socket = std::net::UdpSocket::bind(local_addr)?;
884
885 std_socket.set_nonblocking(true)?;
886 let tokio_socket = tokio::net::UdpSocket::from_std(std_socket)?;
887 let actual_addr = tokio_socket.local_addr()?;
889 let bind_addr = rsipstack::transport::SipConnection::resolve_bind_address(actual_addr);
890 let mut learned_public_address: SharedPublicAddress =
891 Arc::new(ArcSwap::from_pointee(bind_addr.into()));
892
893 let udp_inner = rsipstack::transport::udp::UdpInner {
894 conn: tokio_socket,
895 addr: rsipstack::transport::SipAddr {
896 r#type: Some(rsipstack::rsip::transport::Transport::Udp),
897 addr: bind_addr.into(),
898 },
899 };
900
901 let external = config
902 .external_ip
903 .as_ref()
904 .map(|ip| {
905 format!("{}:{}", ip, actual_addr.port())
906 .parse()
907 .map_err(|e| anyhow::anyhow!("Failed to parse external address: {}", e))
908 })
909 .transpose()?;
910
911 let udp_conn = rsipstack::transport::udp::UdpConnection::attach(
912 udp_inner,
913 external,
914 Some(token.child_token()),
915 )
916 .await;
917
918 info!(
919 "start useragent, addr: {} (SO_REUSEPORT enabled)",
920 udp_conn.get_addr()
921 );
922
923 transport_layer.add_transport(udp_conn.into());
924
925 if let Some(tls_port) = config.tls_port {
927 let tls_addr: std::net::SocketAddr = format!("{}:{}", local_ip, tls_port).parse()?;
928 let tls_sip_addr = rsipstack::transport::SipAddr {
929 r#type: Some(rsipstack::rsip::transport::Transport::Tls),
930 addr: tls_addr.into(),
931 };
932 let mut tls_cfg = rsipstack::transport::tls::TlsConfig::default();
933 if let Some(ref cert_path) = config.tls_cert_file {
934 tls_cfg.cert = Some(
935 std::fs::read(cert_path)
936 .map_err(|e| anyhow::anyhow!("tls_cert_file: {}", e))?,
937 );
938 }
939 if let Some(ref key_path) = config.tls_key_file {
940 tls_cfg.key = Some(
941 std::fs::read(key_path).map_err(|e| anyhow::anyhow!("tls_key_file: {}", e))?,
942 );
943 }
944 let external_tls_addr = config
945 .external_ip
946 .as_ref()
947 .and_then(|ip| format!("{}:{}", ip, tls_port).parse().ok());
948 match rsipstack::transport::tls::TlsListenerConnection::new(
949 tls_sip_addr,
950 external_tls_addr,
951 tls_cfg,
952 )
953 .await
954 {
955 Ok(tls_conn) => {
956 transport_layer.add_transport(tls_conn.into());
957 info!("TLS SIP transport started on {}:{}", local_ip, tls_port);
958 }
959 Err(e) => {
960 return Err(anyhow::anyhow!("Failed to start TLS SIP transport: {}", e));
961 }
962 }
963 }
964
965 let endpoint_option = rsipstack::transaction::endpoint::EndpointOption::default();
966 let mut endpoint_builder = rsipstack::EndpointBuilder::new();
967 if let Some(ref user_agent) = config.useragent {
968 endpoint_builder.with_user_agent(user_agent.as_str());
969 }
970
971 let mut endpoint_builder = endpoint_builder
972 .with_cancel_token(token.child_token())
973 .with_transport_layer(transport_layer)
974 .with_option(endpoint_option);
975
976 if config.auto_learn_public_address.unwrap_or_default() {
977 let inspector = LearningMessageInspector::new(bind_addr.into(), self.message_inspector);
978 learned_public_address = inspector.shared_public_address();
979 endpoint_builder = endpoint_builder.with_inspector(Box::new(inspector));
980 } else if let Some(inspector) = self.message_inspector {
981 endpoint_builder = endpoint_builder.with_inspector(inspector);
982 }
983
984 if let Some(locator) = self.target_locator {
985 endpoint_builder.with_target_locator(locator);
986 } else if let Some(ref rules) = config.rewrites {
987 endpoint_builder
988 .with_target_locator(Box::new(RewriteTargetLocator::new(rules.clone())));
989 }
990
991 if let Some(inspector) = self.transport_inspector {
992 endpoint_builder = endpoint_builder.with_transport_inspector(inspector);
993 }
994
995 let endpoint = endpoint_builder.build();
996 let dialog_layer = Arc::new(DialogLayer::new(endpoint.inner.clone()));
997
998 let stream_engine = self.stream_engine.unwrap_or_default();
999
1000 let callrecord_formatter = if let Some(formatter) = self.callrecord_formatter {
1001 formatter
1002 } else {
1003 let formatter = if let Some(ref callrecord) = config.callrecord {
1004 DefaultCallRecordFormatter::new_with_config(callrecord)
1005 } else {
1006 DefaultCallRecordFormatter::default()
1007 };
1008 Arc::new(formatter)
1009 };
1010
1011 let callrecord_sender = if let Some(sender) = self.callrecord_sender {
1012 Some(sender)
1013 } else if let Some(ref callrecord) = config.callrecord {
1014 let builder = CallRecordManagerBuilder::new()
1015 .with_cancel_token(token.child_token())
1016 .with_config(callrecord.clone())
1017 .with_max_concurrent(32)
1018 .with_formatter(callrecord_formatter.clone());
1019
1020 let mut callrecord_manager = builder.build();
1021 let sender = callrecord_manager.sender.clone();
1022 crate::spawn(async move {
1023 callrecord_manager.serve().await;
1024 });
1025 Some(sender)
1026 } else {
1027 None
1028 };
1029
1030 let app_state = Arc::new(AppStateInner {
1031 config,
1032 token,
1033 stream_engine,
1034 callrecord_sender,
1035 endpoint,
1036 registration_handles: Mutex::new(HashMap::new()),
1037 alive_users: Arc::new(RwLock::new(HashSet::new())),
1038 dialog_layer: dialog_layer.clone(),
1039 create_invitation_handler: self.create_invitation_handler,
1040 invitation: Invitation::new(dialog_layer),
1041 routing_state: Arc::new(crate::call::RoutingState::new()),
1042 pending_playbooks: Arc::new(Mutex::new(HashMap::new())),
1043 learned_public_address,
1044 active_calls: Arc::new(std::sync::Mutex::new(HashMap::new())),
1045 total_calls: AtomicU64::new(0),
1046 total_failed_calls: AtomicU64::new(0),
1047 uptime: Local::now(),
1048 shutting_down: Arc::new(AtomicBool::new(false)),
1049 });
1050
1051 Ok(app_state)
1052 }
1053}