sipx_call/
subscriber.rs

1//! Runtime driver for the sans-I/O RFC 6665 event client.
2//!
3//! [`sipx_ua::event_client::EventClient`] owns protocol decisions. This module owns the bounded
4//! socket transactions, timers and application channels that apply those decisions.
5
6use std::collections::HashMap;
7use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
8use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
9use std::time::Duration;
10
11use bytes::Bytes;
12use sipx_sip::build::ResponseBuilder;
13use sipx_sip::{HeaderName, Response, StatusCode};
14use sipx_transport::{Handle, Incoming, Target, TransportKind};
15use sipx_ua::event_client::{
16    Config, EventClient, NotificationMeta, Output, PackageConsumer, Peer, Start, StartError,
17    StateChange, SubscriptionId, Timer, Transport,
18};
19use thiserror::Error;
20use tokio::sync::mpsc;
21use tokio::task::JoinHandle;
22use tokio_util::sync::CancellationToken;
23
24const DRIVER_QUEUE: usize = 64;
25
26/// Runtime resource measurements for outbound event subscriptions.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub struct EventSubscriptionCounts {
29    /// Live lifecycle tasks.
30    pub active_tasks: usize,
31    /// Live timer tasks.
32    pub active_timers: usize,
33    /// SUBSCRIBE transactions whose final result is still being observed.
34    pub active_transactions: usize,
35    /// Lifecycle tasks started since construction.
36    pub started_tasks: u64,
37    /// Lifecycle tasks which exited.
38    pub finished_tasks: u64,
39    /// NOTIFY requests refused because a bounded driver queue was full.
40    pub shed: u64,
41}
42
43#[derive(Debug, Default)]
44struct Counters {
45    tasks: AtomicUsize,
46    timers: AtomicUsize,
47    transactions: AtomicUsize,
48    started: AtomicU64,
49    finished: AtomicU64,
50    shed: AtomicU64,
51}
52
53/// Failure before an outbound subscription owns runtime work.
54#[derive(Debug, Error)]
55#[non_exhaustive]
56pub enum EventSubscriptionError {
57    /// The client has not been attached to a dispatcher yet.
58    #[error("event subscriptions are not attached to a dispatcher")]
59    NotAttached,
60    /// Dispatcher shutdown has atomically closed admission.
61    #[error("event subscription shutdown has closed admission")]
62    ShuttingDown,
63    /// A peer-driven or configured client bound was reached.
64    #[error("event subscription capacity exceeded")]
65    CapacityExceeded,
66    /// The caller reused an active Call-ID.
67    #[error("an event subscription already owns this Call-ID")]
68    DuplicateIdentity,
69    /// The pure client rejected the start fields.
70    #[error(transparent)]
71    Start(#[from] StartError),
72}
73
74/// One application delivery from a package consumer.
75#[derive(Debug)]
76pub struct EventNotification<V> {
77    /// Local monotonic instant when the runtime accepted this package value.
78    pub received_at: tokio::time::Instant,
79    /// Framework state and remote sequence; absent for the initial neutral value.
80    pub metadata: Option<NotificationMeta>,
81    /// Parsed package value.
82    pub value: V,
83}
84
85/// One application-visible event from a running subscription.
86#[derive(Debug)]
87pub enum EventSubscriptionEvent<V> {
88    /// A package value was accepted from a NOTIFY.
89    Notification(EventNotification<V>),
90    /// The framework lifecycle changed.
91    State(StateChange),
92}
93
94/// Application ownership of one running subscription.
95#[derive(Debug)]
96pub struct EventSubscription<V> {
97    id: SubscriptionId,
98    deliveries: mpsc::Receiver<EventNotification<V>>,
99    states: mpsc::Receiver<StateChange>,
100    commands: mpsc::Sender<Command>,
101}
102
103impl<V> EventSubscription<V> {
104    /// Opaque identity allocated by the pure client.
105    #[must_use]
106    pub fn id(&self) -> SubscriptionId {
107        self.id
108    }
109
110    /// Receive one package value and release its bounded queue slot.
111    pub async fn recv(&mut self) -> Option<EventNotification<V>> {
112        let delivery = self.deliveries.recv().await?;
113        // discard: the bounded command queue cannot fill before its equally bounded deliveries.
114        let _ = self.commands.try_send(Command::Drained(1));
115        Some(delivery)
116    }
117
118    /// Receive one lifecycle fact.
119    pub async fn next_state(&mut self) -> Option<StateChange> {
120        self.states.recv().await
121    }
122
123    /// Receive the next package value or lifecycle fact without favoring either channel.
124    ///
125    /// This is the cancellation-safe choice for applications which must distinguish an initial
126    /// refusal from the first package snapshot. [`Self::recv`] and [`Self::next_state`] remain
127    /// available when an application intentionally observes only one side.
128    pub async fn next_event(&mut self) -> Option<EventSubscriptionEvent<V>> {
129        tokio::select! {
130            Some(delivery) = self.deliveries.recv() => {
131                // discard: the bounded command queue cannot fill before its bounded deliveries.
132                let _ = self.commands.try_send(Command::Drained(1));
133                Some(EventSubscriptionEvent::Notification(delivery))
134            }
135            Some(change) = self.states.recv() => Some(EventSubscriptionEvent::State(change)),
136            else => None,
137        }
138    }
139
140    /// Send Expires 0 and wait only for command admission. Terminal NOTIFY or Timer N completes
141    /// the protocol operation and is observable through [`Self::next_state`].
142    pub async fn unsubscribe(&self) -> Result<(), EventSubscriptionError> {
143        self.commands
144            .send(Command::Unsubscribe)
145            .await
146            .map_err(|_| EventSubscriptionError::NotAttached)
147    }
148}
149
150impl<V> Drop for EventSubscription<V> {
151    fn drop(&mut self) {
152        // discard: Drop cannot wait; finite expiry and dispatcher shutdown remain backstops.
153        let _ = self.commands.try_send(Command::Unsubscribe);
154    }
155}
156
157#[derive(Debug)]
158struct Shared {
159    endpoint: Mutex<Option<Handle>>,
160    routes: Mutex<HashMap<Vec<u8>, mpsc::Sender<Incoming>>>,
161    drivers: Mutex<HashMap<Vec<u8>, JoinHandle<()>>>,
162    config: Config,
163    counters: Arc<Counters>,
164    shutdown: CancellationToken,
165}
166
167/// Cloneable application handle. It does not keep dispatcher-owned tasks alive.
168#[derive(Debug, Clone)]
169pub struct EventSubscriptionsHandle {
170    shared: Arc<Shared>,
171}
172
173impl EventSubscriptionsHandle {
174    /// Start one package-generic subscription through the attached endpoint.
175    pub fn subscribe<C: PackageConsumer>(
176        &self,
177        start: Start<C>,
178    ) -> Result<EventSubscription<C::Value>, EventSubscriptionError> {
179        let call_id = start.call_id.as_bytes().to_vec();
180        let mut drivers = lock(&self.shared.drivers);
181        drivers.retain(|_, task| !task.is_finished());
182        if self.shared.shutdown.is_cancelled() {
183            return Err(EventSubscriptionError::ShuttingDown);
184        }
185        if drivers.contains_key(&call_id) {
186            return Err(EventSubscriptionError::DuplicateIdentity);
187        }
188        let endpoint = lock(&self.shared.endpoint)
189            .clone()
190            .ok_or(EventSubscriptionError::NotAttached)?;
191        reserve(&self.shared)?;
192        let mut core = match EventClient::new(self.shared.config.clone()) {
193            Ok(core) => core,
194            Err(error) => {
195                release(&self.shared.counters);
196                return Err(error.into());
197            }
198        };
199        let (id, initial) = match core.start(start) {
200            Ok(started) => started,
201            Err(error) => {
202                release(&self.shared.counters);
203                return Err(error.into());
204            }
205        };
206        let (incoming_tx, incoming_rx) = mpsc::channel(DRIVER_QUEUE);
207        let mut routes = lock(&self.shared.routes);
208        if routes.contains_key(&call_id) {
209            release(&self.shared.counters);
210            return Err(EventSubscriptionError::DuplicateIdentity);
211        }
212        routes.insert(call_id.clone(), incoming_tx);
213        drop(routes);
214        let (delivery_tx, deliveries) = mpsc::channel(self.shared.config.delivery_capacity);
215        let (state_tx, states) = mpsc::channel(DRIVER_QUEUE);
216        let (command_tx, command_rx) = mpsc::channel(DRIVER_QUEUE);
217        let driver = Driver {
218            id,
219            call_id: call_id.clone(),
220            endpoint,
221            core,
222            incoming: incoming_rx,
223            commands: command_rx,
224            delivery: delivery_tx,
225            states: state_tx,
226            events: None,
227            response: None,
228            timers: HashMap::new(),
229            shared: Arc::clone(&self.shared),
230        };
231        self.shared.counters.started.fetch_add(1, Ordering::Relaxed);
232        let task = tokio::spawn(driver.run(initial));
233        drivers.insert(call_id, task);
234        drop(drivers);
235        Ok(EventSubscription {
236            id,
237            deliveries,
238            states,
239            commands: command_tx,
240        })
241    }
242
243    /// Point-in-time owned-work counts.
244    #[must_use]
245    pub fn counts(&self) -> EventSubscriptionCounts {
246        EventSubscriptionCounts {
247            active_tasks: self.shared.counters.tasks.load(Ordering::Relaxed),
248            active_timers: self.shared.counters.timers.load(Ordering::Relaxed),
249            active_transactions: self.shared.counters.transactions.load(Ordering::Relaxed),
250            started_tasks: self.shared.counters.started.load(Ordering::Relaxed),
251            finished_tasks: self.shared.counters.finished.load(Ordering::Relaxed),
252            shed: self.shared.counters.shed.load(Ordering::Relaxed),
253        }
254    }
255}
256
257/// Dispatcher-owned outbound event subscription runtime.
258#[derive(Debug)]
259pub struct EventSubscriptions {
260    shared: Arc<Shared>,
261}
262
263impl EventSubscriptions {
264    /// Construct a bounded runtime. Attach it with
265    /// [`crate::Dispatcher::with_event_subscriptions`] before starting work.
266    pub fn new(config: Config) -> Result<Self, EventSubscriptionError> {
267        config.validate()?;
268        Ok(Self {
269            shared: Arc::new(Shared {
270                endpoint: Mutex::new(None),
271                routes: Mutex::new(HashMap::new()),
272                drivers: Mutex::new(HashMap::new()),
273                config,
274                counters: Arc::new(Counters::default()),
275                shutdown: CancellationToken::new(),
276            }),
277        })
278    }
279
280    /// Application handle which can start subscriptions after dispatcher attachment.
281    #[must_use]
282    pub fn handle(&self) -> EventSubscriptionsHandle {
283        EventSubscriptionsHandle {
284            shared: Arc::clone(&self.shared),
285        }
286    }
287
288    pub(crate) fn attach(&self, endpoint: Handle) {
289        *lock(&self.shared.endpoint) = Some(endpoint);
290    }
291
292    /// Route one NOTIFY without moving it away from the dispatcher's fallback path.
293    pub(crate) async fn receive(&self, incoming: &Incoming) -> bool {
294        let Some(call_id) = incoming.request.headers.value(&HeaderName::CallId) else {
295            return false;
296        };
297        let sender = lock(&self.shared.routes).get(call_id.as_ref()).cloned();
298        let Some(sender) = sender else {
299            return false;
300        };
301        let cloned = Incoming {
302            key: incoming.key.clone(),
303            request: incoming.request.clone(),
304            source: incoming.source,
305            transport: incoming.transport,
306            connection_generation: incoming.connection_generation,
307        };
308        match sender.try_send(cloned) {
309            Ok(()) => true,
310            Err(mpsc::error::TrySendError::Full(_)) => {
311                self.shared.counters.shed.fetch_add(1, Ordering::Relaxed);
312                let endpoint = lock(&self.shared.endpoint).clone();
313                answer_notify(
314                    endpoint.as_ref(),
315                    incoming,
316                    503,
317                    Some(Duration::from_secs(1)),
318                )
319                .await;
320                true
321            }
322            Err(mpsc::error::TrySendError::Closed(_)) => false,
323        }
324    }
325
326    pub(crate) async fn shutdown(&mut self) {
327        let drivers: Vec<_> = {
328            let mut drivers = lock(&self.shared.drivers);
329            self.shared.shutdown.cancel();
330            drivers.drain().map(|(_, task)| task).collect()
331        };
332        for task in drivers {
333            if let Err(error) = task.await {
334                tracing::warn!(%error, "event subscription driver did not join cleanly");
335            }
336        }
337    }
338}
339
340impl Drop for EventSubscriptions {
341    fn drop(&mut self) {
342        let _drivers = lock(&self.shared.drivers);
343        self.shared.shutdown.cancel();
344    }
345}
346
347#[derive(Debug)]
348enum Command {
349    Drained(usize),
350    Unsubscribe,
351}
352
353#[derive(Debug)]
354enum RuntimeEvent {
355    Response(Option<Response>),
356    Timer(Timer, u64),
357}
358
359struct Driver<C: PackageConsumer> {
360    id: SubscriptionId,
361    call_id: Vec<u8>,
362    endpoint: Handle,
363    core: EventClient<C>,
364    incoming: mpsc::Receiver<Incoming>,
365    commands: mpsc::Receiver<Command>,
366    delivery: mpsc::Sender<EventNotification<C::Value>>,
367    states: mpsc::Sender<StateChange>,
368    events: Option<(mpsc::Sender<RuntimeEvent>, mpsc::Receiver<RuntimeEvent>)>,
369    response: Option<JoinHandle<()>>,
370    timers: HashMap<Timer, JoinHandle<()>>,
371    shared: Arc<Shared>,
372}
373
374impl<C: PackageConsumer> Driver<C> {
375    async fn run(mut self, initial: Vec<Output<C::Value>>) {
376        let guard = WorkGuard::task(Arc::clone(&self.shared.counters));
377        let (event_tx, event_rx) = mpsc::channel(DRIVER_QUEUE);
378        self.events = Some((event_tx, event_rx));
379        self.apply(initial, None).await;
380        loop {
381            if !self.core.contains(self.id) {
382                break;
383            }
384            let event = {
385                let Some((_, events)) = self.events.as_mut() else {
386                    break;
387                };
388                tokio::select! {
389                    () = self.shared.shutdown.cancelled() => DriverInput::Shutdown,
390                    incoming = self.incoming.recv() => incoming.map_or(DriverInput::Shutdown, |incoming| DriverInput::Incoming(Box::new(incoming))),
391                    command = self.commands.recv() => DriverInput::Command(command),
392                    event = events.recv() => DriverInput::Event(event),
393                    () = self.delivery.closed() => DriverInput::Shutdown,
394                }
395            };
396            let (outputs, incoming) = match event {
397                DriverInput::Incoming(incoming) => {
398                    let incoming = *incoming;
399                    let source = peer_from_incoming(&incoming);
400                    let outputs = self.core.notify(1, &incoming.request, source);
401                    (outputs, Some(incoming))
402                }
403                DriverInput::Command(Some(Command::Drained(count))) => {
404                    self.core.consumer_drained(self.id, count);
405                    (Vec::new(), None)
406                }
407                DriverInput::Command(Some(Command::Unsubscribe)) => {
408                    (self.core.unsubscribe(self.id), None)
409                }
410                DriverInput::Event(Some(RuntimeEvent::Response(response))) => (
411                    self.core
412                        .response(self.id, response.as_ref(), &sipx_ua::auth::new_cnonce()),
413                    None,
414                ),
415                DriverInput::Event(Some(RuntimeEvent::Timer(timer, generation))) => {
416                    (self.core.timer_fired(self.id, timer, generation), None)
417                }
418                DriverInput::Shutdown | DriverInput::Command(None) | DriverInput::Event(None) => {
419                    (self.core.shutdown_deadline(), None)
420                }
421            };
422            self.apply(outputs, incoming.as_ref()).await;
423        }
424        if self.core.contains(self.id) {
425            let outputs = self.core.shutdown_deadline();
426            self.apply(outputs, None).await;
427        }
428        if let Some(response) = self.response.take() {
429            abort_and_join(response).await;
430        }
431        for (_, timer) in self.timers.drain() {
432            abort_and_join(timer).await;
433        }
434        lock(&self.shared.routes).remove(&self.call_id);
435        drop(guard);
436    }
437
438    async fn apply(&mut self, outputs: Vec<Output<C::Value>>, incoming: Option<&Incoming>) {
439        for output in outputs {
440            match output {
441                Output::SendSubscribe {
442                    request, target, ..
443                } => self.send(*request, target).await,
444                Output::RespondNotify {
445                    status,
446                    retry_after,
447                    ..
448                } => {
449                    if let Some(incoming) = incoming {
450                        answer_notify(Some(&self.endpoint), incoming, status, retry_after).await;
451                    }
452                }
453                Output::Deliver {
454                    metadata, value, ..
455                } => {
456                    // discard: failure requires concurrent consumer closure; the driver then stops.
457                    let _ = self.delivery.try_send(EventNotification {
458                        received_at: tokio::time::Instant::now(),
459                        metadata,
460                        value,
461                    });
462                }
463                Output::ArmTimer {
464                    timer,
465                    generation,
466                    after,
467                    ..
468                } => self.arm(timer, generation, after).await,
469                Output::CancelTimer { timer, .. } => {
470                    if let Some(task) = self.timers.remove(&timer) {
471                        abort_and_join(task).await;
472                    }
473                }
474                Output::StateChanged { change, .. } => {
475                    // discard: a full or closed state channel means its consumer chose not to read.
476                    let _ = self.states.try_send(change);
477                }
478                Output::Stopped => {}
479            }
480        }
481    }
482
483    async fn send(&mut self, request: sipx_sip::Request, peer: Peer) {
484        if let Some(response) = self.response.take() {
485            abort_and_join(response).await;
486        }
487        let Some((events, _)) = self.events.as_ref() else {
488            return;
489        };
490        match self.endpoint.send(request, target(peer)).await {
491            Ok(mut responses) => {
492                self.core
493                    .connection_selected(self.id, responses.connection_generation());
494                let events = events.clone();
495                let counters = Arc::clone(&self.shared.counters);
496                self.response = Some(tokio::spawn(async move {
497                    let _guard = WorkGuard::transaction(counters);
498                    let response = responses.final_response().await;
499                    // discard: closure means the owning driver is already tearing down.
500                    let _ = events.send(RuntimeEvent::Response(response)).await;
501                }));
502            }
503            Err(error) => {
504                tracing::warn!(%error, "could not send event SUBSCRIBE");
505                // discard: closure means the owning driver is already tearing down.
506                let _ = events.try_send(RuntimeEvent::Response(None));
507            }
508        }
509    }
510
511    async fn arm(&mut self, timer: Timer, generation: u64, after: Duration) {
512        if let Some(previous) = self.timers.remove(&timer) {
513            abort_and_join(previous).await;
514        }
515        let Some((events, _)) = self.events.as_ref() else {
516            return;
517        };
518        let events = events.clone();
519        let counters = Arc::clone(&self.shared.counters);
520        self.timers.insert(
521            timer,
522            tokio::spawn(async move {
523                let _guard = WorkGuard::timer(counters);
524                // Protocol timer: the duration is the state-machine input this task represents.
525                tokio::time::sleep(after).await;
526                // discard: closure means the owning driver has cancelled this timer's state.
527                let _ = events.send(RuntimeEvent::Timer(timer, generation)).await;
528            }),
529        );
530    }
531}
532
533async fn abort_and_join(task: JoinHandle<()>) {
534    task.abort();
535    // discard: cancellation is the requested outcome; the await is solely the ownership barrier.
536    let _ = task.await;
537}
538
539enum DriverInput {
540    Incoming(Box<Incoming>),
541    Command(Option<Command>),
542    Event(Option<RuntimeEvent>),
543    Shutdown,
544}
545
546enum WorkKind {
547    Task,
548    Timer,
549    Transaction,
550}
551
552struct WorkGuard {
553    counters: Arc<Counters>,
554    kind: WorkKind,
555}
556
557impl WorkGuard {
558    fn task(counters: Arc<Counters>) -> Self {
559        Self {
560            counters,
561            kind: WorkKind::Task,
562        }
563    }
564
565    fn timer(counters: Arc<Counters>) -> Self {
566        counters.timers.fetch_add(1, Ordering::Relaxed);
567        Self {
568            counters,
569            kind: WorkKind::Timer,
570        }
571    }
572
573    fn transaction(counters: Arc<Counters>) -> Self {
574        counters.transactions.fetch_add(1, Ordering::Relaxed);
575        Self {
576            counters,
577            kind: WorkKind::Transaction,
578        }
579    }
580}
581
582impl Drop for WorkGuard {
583    fn drop(&mut self) {
584        match self.kind {
585            WorkKind::Task => {
586                self.counters.tasks.fetch_sub(1, Ordering::Relaxed);
587                self.counters.finished.fetch_add(1, Ordering::Relaxed);
588            }
589            WorkKind::Timer => {
590                self.counters.timers.fetch_sub(1, Ordering::Relaxed);
591            }
592            WorkKind::Transaction => {
593                self.counters.transactions.fetch_sub(1, Ordering::Relaxed);
594            }
595        }
596    }
597}
598
599// `fetch_update` remains the spelling available at the workspace MSRV; current nightly deprecates
600// it before the replacement is available on that supported toolchain.
601#[allow(deprecated)]
602fn reserve(shared: &Shared) -> Result<(), EventSubscriptionError> {
603    shared
604        .counters
605        .tasks
606        .fetch_update(Ordering::AcqRel, Ordering::Relaxed, |current| {
607            (current < shared.config.capacity).then_some(current.saturating_add(1))
608        })
609        .map_err(|_| EventSubscriptionError::CapacityExceeded)?;
610    Ok(())
611}
612
613fn release(counters: &Counters) {
614    counters.tasks.fetch_sub(1, Ordering::Relaxed);
615}
616
617fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
618    mutex.lock().unwrap_or_else(PoisonError::into_inner)
619}
620
621fn peer_from_incoming(incoming: &Incoming) -> Peer {
622    let mut peer = Peer::new(incoming.source, from_transport(incoming.transport));
623    peer.connection = incoming.connection_generation;
624    peer
625}
626
627fn target(peer: Peer) -> Target {
628    let mut target = Target::new(peer.address, to_transport(peer.transport));
629    if let Some(identity) = peer.identity {
630        target = target.verifying(identity);
631    }
632    if let Some(path) = peer.path {
633        target = target.at_path(path);
634    }
635    target
636}
637
638fn from_transport(transport: TransportKind) -> Transport {
639    match transport {
640        TransportKind::Udp => Transport::Udp,
641        TransportKind::Tcp => Transport::Tcp,
642        TransportKind::Tls => Transport::Tls,
643        TransportKind::Ws => Transport::Ws,
644        TransportKind::Wss => Transport::Wss,
645        TransportKind::Quic => Transport::Quic,
646    }
647}
648
649fn to_transport(transport: Transport) -> TransportKind {
650    match transport {
651        Transport::Udp => TransportKind::Udp,
652        Transport::Tcp => TransportKind::Tcp,
653        Transport::Tls => TransportKind::Tls,
654        Transport::Ws => TransportKind::Ws,
655        Transport::Wss => TransportKind::Wss,
656        Transport::Quic => TransportKind::Quic,
657    }
658}
659
660async fn answer_notify(
661    endpoint: Option<&Handle>,
662    incoming: &Incoming,
663    status: u16,
664    retry_after: Option<Duration>,
665) {
666    let (Some(endpoint), Some(status)) = (endpoint, StatusCode::new(status)) else {
667        return;
668    };
669    let built = ResponseBuilder::to_request(&incoming.request, status, "Event notification")
670        .and_then(|builder| match retry_after {
671            Some(value) => builder.header(
672                HeaderName::RetryAfter,
673                Bytes::from(value.as_secs().to_string()),
674            ),
675            None => Ok(builder),
676        });
677    let Ok(builder) = built else {
678        return;
679    };
680    if let Err(error) = endpoint.respond(&incoming.key, builder.build()).await {
681        tracing::warn!(%error, "could not answer event NOTIFY");
682    }
683}
684
685#[cfg(test)]
686#[allow(
687    clippy::unwrap_used,
688    clippy::expect_used,
689    clippy::panic,
690    clippy::indexing_slicing
691)]
692mod admission_tests {
693    use std::sync::{Arc, Barrier};
694    use std::time::Duration;
695
696    use bytes::Bytes;
697    use sipx_transport::{Config as TransportConfig, bind};
698    use sipx_ua::event_client::{PackageRejection, SamePeer, Start, Transport};
699
700    use super::*;
701
702    #[derive(Debug)]
703    struct Package;
704
705    impl PackageConsumer for Package {
706        type Value = ();
707        fn event(&self) -> &'static str {
708            "admission"
709        }
710        fn accept(&self) -> &[String] {
711            &[]
712        }
713        fn neutral(&mut self) -> Option<()> {
714            None
715        }
716        fn consume(&mut self, _: Option<&[u8]>, _: &[u8]) -> Result<(), PackageRejection> {
717            Ok(())
718        }
719    }
720
721    fn start(target: std::net::SocketAddr) -> Start<Package> {
722        Start {
723            resource: sipx_sip::Uri::parse(Bytes::from_static(b"sip:resource@example.test"))
724                .expect("URI"),
725            local_identity: "<sip:client@example.test>".to_owned(),
726            contact: "<sip:client@127.0.0.1>".to_owned(),
727            target: Peer::new(target, Transport::Udp),
728            expires: Duration::from_secs(60),
729            body: Bytes::new(),
730            content_type: None,
731            credentials: None,
732            call_id: "admission@example.test".to_owned(),
733            from_tag: "admission".to_owned(),
734            initial_cseq: 1,
735            consumer: Package,
736            trust: Arc::new(SamePeer),
737        }
738    }
739
740    #[tokio::test]
741    async fn racing_shutdown_closes_admission_before_any_spawn() {
742        let (endpoint, _) = bind(TransportConfig::new(
743            "127.0.0.1:0".parse().expect("address"),
744        ))
745        .await
746        .expect("endpoint");
747        let runtime = EventSubscriptions::new(Config::default()).expect("runtime");
748        runtime.attach(endpoint.clone());
749        let handle = runtime.handle();
750        let post_shutdown = handle.clone();
751        let shared = Arc::clone(&runtime.shared);
752        let drivers = lock(&shared.drivers);
753        let barrier = Arc::new(Barrier::new(2));
754        let contender = Arc::clone(&barrier);
755        let target = endpoint.local_addr();
756        let attempt = std::thread::spawn(move || {
757            contender.wait();
758            handle.subscribe(start(target))
759        });
760        barrier.wait();
761        shared.shutdown.cancel();
762        drop(drivers);
763        assert!(matches!(
764            attempt.join().expect("thread"),
765            Err(EventSubscriptionError::ShuttingDown)
766        ));
767        assert!(matches!(
768            post_shutdown.subscribe(start(target)),
769            Err(EventSubscriptionError::ShuttingDown)
770        ));
771        assert!(lock(&shared.drivers).is_empty());
772        endpoint.shutdown().await;
773    }
774}
sipx_call/subscriber.rs

sipx_call/
subscriber.rs
