sipx_ua/
subscribe.rs

1//! The subscriptions a notifier is serving (RFC 6665).
2//!
3//! [`sipx_sip::event`] decides what a `Subscription-State` means. This holds the ones that exist:
4//! establishing them, refreshing them, expiring them, and — the part that matters most — making
5//! sure a terminated one stays terminated.
6//!
7//! Time is a parameter, not a call to a clock, for the same reason it is in the timer queue: a
8//! notifier driven by a scheduler somebody else owns has to be able to say what "now" is, and a
9//! test that wants to watch a subscription expire should not have to wait an hour.
10//! **Supported** (`S-35`): `sipx-call::Notifier` drives this exact store from the live endpoint
11//! dispatcher. Breaking changes receive migration guidance while sipx remains pre-1.0.
12//!
13
14use std::time::Duration;
15
16use sipx_sip::event::{
17    BAD_EVENT, Packages, Reason, State, Subscription, granted_expiry, is_unsubscribe,
18};
19use sipx_sip::headers::{CSeq, Expires, From as FromHeader};
20use sipx_sip::{HeaderName, Method, Request};
21
22/// What identifies one subscription (RFC 6665 §4.4.1).
23///
24/// The dialog plus the event package — *not* the dialog alone. §4.4.1 allows several subscriptions
25/// in one dialog as long as their `Event` differs, so keying on the dialog would have a second
26/// subscription silently replace the first.
27#[derive(Debug, Clone, PartialEq, Eq, Hash)]
28pub struct Id {
29    /// The dialog's `Call-ID`.
30    pub call_id: String,
31    /// The subscriber's tag.
32    pub from_tag: String,
33    /// The exact `Event` type, with only its `id` parameter if it has one.
34    pub event: String,
35}
36
37impl Id {
38    /// Read the identity out of a SUBSCRIBE.
39    #[must_use]
40    pub fn from_request(request: &Request) -> Option<Self> {
41        if request.headers.count(&HeaderName::CallId) != 1
42            || request.headers.count(&HeaderName::From) != 1
43            || request.headers.count(&HeaderName::Event) != 1
44        {
45            return None;
46        }
47        let call_id = request.headers.value(&HeaderName::CallId)?;
48        let call_id = std::str::from_utf8(&call_id).ok()?;
49        if call_id.is_empty() {
50            return None;
51        }
52        let from = request.headers.typed::<FromHeader>()?.ok()?;
53        let mut tags = from.params.iter().filter(|parameter| parameter.is("tag"));
54        let tag = tags.next()?.value.as_deref()?;
55        if tags.next().is_some() || tag.is_empty() || !tag.iter().copied().all(is_token_char) {
56            return None;
57        }
58        Some(Self {
59            call_id: call_id.to_owned(),
60            from_tag: std::str::from_utf8(tag).ok()?.to_owned(),
61            event: event_identity(&request.headers.value(&HeaderName::Event)?)?,
62        })
63    }
64}
65
66fn event_identity(value: &[u8]) -> Option<String> {
67    let segments = event_segments(value)?;
68    let event = trim_ows(segments.first()?);
69    if !valid_event_type(event) {
70        return None;
71    }
72
73    let mut id = None;
74    for segment in segments.iter().skip(1) {
75        let parameter = trim_ows(segment);
76        if parameter.is_empty() {
77            return None;
78        }
79        let (name, value) =
80            parameter
81                .iter()
82                .position(|byte| *byte == b'=')
83                .map_or((parameter, None), |equals| {
84                    (
85                        trim_ows(parameter.get(..equals).unwrap_or_default()),
86                        Some(trim_ows(
87                            parameter
88                                .get(equals.saturating_add(1)..)
89                                .unwrap_or_default(),
90                        )),
91                    )
92                });
93        if name.is_empty() || !name.iter().copied().all(is_token_char) {
94            return None;
95        }
96        if name.eq_ignore_ascii_case(b"id") {
97            let value = value?;
98            if id.is_some() || value.is_empty() || !value.iter().copied().all(is_token_char) {
99                return None;
100            }
101            id = Some(value);
102        } else if !value.is_none_or(valid_generic_value) {
103            return None;
104        }
105    }
106
107    let event = std::str::from_utf8(event).ok()?;
108    match id {
109        Some(id) => Some(format!("{event};id={}", std::str::from_utf8(id).ok()?)),
110        None => Some(event.to_owned()),
111    }
112}
113
114fn event_segments(value: &[u8]) -> Option<Vec<&[u8]>> {
115    let mut segments = Vec::new();
116    let mut start = 0;
117    let mut quoted = false;
118    let mut escaped = false;
119    for (index, byte) in value.iter().copied().enumerate() {
120        if quoted && escaped {
121            escaped = false;
122            continue;
123        }
124        match byte {
125            b'\\' if quoted => escaped = true,
126            b'"' => quoted = !quoted,
127            b';' if !quoted => {
128                segments.push(value.get(start..index)?);
129                start = index.saturating_add(1);
130            }
131            _ => {}
132        }
133    }
134    if quoted || escaped {
135        return None;
136    }
137    segments.push(value.get(start..)?);
138    Some(segments)
139}
140
141fn valid_event_type(value: &[u8]) -> bool {
142    !value.is_empty()
143        && value
144            .split(|byte| *byte == b'.')
145            .all(|token| !token.is_empty() && token.iter().copied().all(is_token_nodot_char))
146}
147
148fn valid_generic_value(value: &[u8]) -> bool {
149    if value.len() >= 2 && value.first() == Some(&b'"') && value.last() == Some(&b'"') {
150        return true;
151    }
152    !value.is_empty()
153        && value.iter().all(|byte| {
154            byte.is_ascii_graphic() && !matches!(byte, b';' | b',' | b'"' | b'<' | b'>')
155        })
156}
157
158fn trim_ows(mut value: &[u8]) -> &[u8] {
159    while matches!(value.first(), Some(b' ' | b'\t')) {
160        value = value.get(1..).unwrap_or_default();
161    }
162    while matches!(value.last(), Some(b' ' | b'\t')) {
163        value = value
164            .get(..value.len().saturating_sub(1))
165            .unwrap_or_default();
166    }
167    value
168}
169
170fn is_token_char(byte: u8) -> bool {
171    is_token_nodot_char(byte) || byte == b'.'
172}
173
174fn is_token_nodot_char(byte: u8) -> bool {
175    byte.is_ascii_alphanumeric()
176        || matches!(
177            byte,
178            b'-' | b'!' | b'%' | b'*' | b'_' | b'+' | b'`' | b'\'' | b'~'
179        )
180}
181
182/// One subscription being served.
183#[derive(Debug, Clone, PartialEq, Eq)]
184pub struct Served {
185    /// Which subscription.
186    pub id: Id,
187    /// Its state.
188    pub state: State,
189    /// When it expires, in seconds on the caller's clock.
190    pub expires_at: u64,
191    /// The last accepted remote SUBSCRIBE sequence number.
192    pub remote_cseq: u32,
193}
194
195/// What answering a SUBSCRIBE concluded.
196#[derive(Debug, Clone, PartialEq, Eq)]
197pub enum Answer {
198    /// A new subscription. Answer 2xx, then send the first NOTIFY.
199    Established {
200        /// Which one.
201        id: Id,
202        /// The expiry granted, which may be shorter than the one asked for.
203        expires: Duration,
204    },
205    /// An existing subscription's timer was pushed out.
206    Refreshed {
207        /// Which one.
208        id: Id,
209        /// The expiry granted.
210        expires: Duration,
211    },
212    /// `Expires: 0` — the subscriber is leaving (§3.1.1).
213    ///
214    /// The notifier still owes a terminating NOTIFY (§4.2.1.4), which is the part that is easy to
215    /// miss: an unsubscribe is not a subscription of no duration, it is an ending.
216    Unsubscribed {
217        /// Which one.
218        id: Id,
219    },
220    /// The `Event` names a package this notifier does not serve. Answer [`BAD_EVENT`].
221    Unserved {
222        /// The status to answer with — 489, and not 400 or 501.
223        status: u16,
224    },
225    /// A new subscription would exceed this notifier's configured peer-driven resource bound.
226    AtCapacity,
227    /// The request could not be read as a SUBSCRIBE at all.
228    Malformed,
229    /// A matching subscription received an equal or lower remote `CSeq`.
230    OutOfOrder {
231        /// The subscription that was deliberately left unchanged.
232        id: Id,
233    },
234}
235
236/// The subscriptions one notifier is serving.
237#[derive(Debug)]
238pub struct Subscriptions {
239    packages: Packages,
240    policy_maximum: Duration,
241    capacity: usize,
242    held: Vec<Served>,
243}
244
245impl Subscriptions {
246    /// A notifier serving these packages, granting at most this long.
247    #[must_use]
248    pub fn new(packages: Packages, policy_maximum: Duration) -> Self {
249        Self {
250            packages,
251            policy_maximum,
252            capacity: 1024,
253            held: Vec::new(),
254        }
255    }
256
257    /// Apply a finite concurrent-subscription bound.
258    ///
259    /// Zero is raised to one: a notifier configured with no capacity could advertise packages but
260    /// never serve one, which is almost certainly a configuration mistake rather than policy.
261    #[must_use]
262    pub fn with_capacity(mut self, capacity: usize) -> Self {
263        self.capacity = capacity.max(1);
264        self
265    }
266
267    /// The concurrent-subscription bound.
268    #[must_use]
269    pub fn capacity(&self) -> usize {
270        self.capacity
271    }
272
273    /// The packages served, for an `Allow-Events` header.
274    #[must_use]
275    pub fn packages(&self) -> &Packages {
276        &self.packages
277    }
278
279    /// How many subscriptions are live.
280    #[must_use]
281    pub fn active(&self) -> usize {
282        self.held
283            .iter()
284            .filter(|held| held.state != State::Terminated)
285            .count()
286    }
287
288    /// Every subscription being served, terminated ones included until they are swept.
289    #[must_use]
290    pub fn all(&self) -> &[Served] {
291        &self.held
292    }
293
294    /// Answer a SUBSCRIBE.
295    pub fn on_subscribe(&mut self, request: &Request, now: u64) -> Answer {
296        let cseq = if request.headers.count(&HeaderName::CSeq) == 1 {
297            match request.headers.typed::<CSeq>() {
298                Some(Ok(CSeq {
299                    sequence,
300                    method: Method::Subscribe,
301                })) => sequence,
302                _ => return Answer::Malformed,
303            }
304        } else {
305            return Answer::Malformed;
306        };
307        if request.headers.count(&HeaderName::Expires) > 1 {
308            return Answer::Malformed;
309        }
310        let Some(id) = Id::from_request(request) else {
311            return Answer::Malformed;
312        };
313        if !self.packages.serves(&id.event) {
314            // §4.2.1.1. Refused by name rather than accepted and then never notified — a
315            // subscriber left waiting for a notification cannot tell that from a slow notifier.
316            return Answer::Unserved { status: BAD_EVENT };
317        }
318
319        let requested = match request.headers.typed::<Expires>() {
320            None => self.policy_maximum,
321            Some(Ok(expires)) => Duration::from_secs(u64::from(expires.0)),
322            Some(Err(_)) => return Answer::Malformed,
323        };
324
325        if is_unsubscribe(requested) {
326            // Marked terminated rather than removed, so a NOTIFY that crosses it on the wire finds
327            // a terminated subscription rather than no subscription at all — which is the
328            // difference between "this is over" and "this never existed".
329            if let Some(held) = self.held.iter_mut().find(|held| held.id == id) {
330                if cseq <= held.remote_cseq {
331                    return Answer::OutOfOrder { id };
332                }
333                held.remote_cseq = cseq;
334                held.state = State::Terminated;
335            }
336            return Answer::Unsubscribed { id };
337        }
338
339        let expires = granted_expiry(requested, self.policy_maximum);
340        let expires_at = now.saturating_add(expires.as_secs());
341
342        if let Some(held) = self.held.iter_mut().find(|held| held.id == id) {
343            // §4.1.2.2: a refresh on an existing dialog pushes the timer out. A *terminated*
344            // subscription is not refreshed back to life — §4.1.3 makes termination final, and a
345            // subscriber that wants another one sends a SUBSCRIBE in a new dialog.
346            if cseq <= held.remote_cseq {
347                return Answer::OutOfOrder { id };
348            }
349            if held.state == State::Terminated {
350                return Answer::Unserved { status: BAD_EVENT };
351            }
352            held.remote_cseq = cseq;
353            held.expires_at = expires_at;
354            held.state = State::Active;
355            return Answer::Refreshed { id, expires };
356        }
357
358        if self.active() >= self.capacity {
359            return Answer::AtCapacity;
360        }
361
362        self.held.push(Served {
363            id: id.clone(),
364            state: State::Active,
365            expires_at,
366            remote_cseq: cseq,
367        });
368        Answer::Established { id, expires }
369    }
370
371    /// End a subscription deliberately.
372    ///
373    /// Returns the state to put in the terminating NOTIFY, or `None` if there was nothing to end.
374    pub fn terminate(&mut self, id: &Id, reason: Reason) -> Option<Subscription> {
375        let held = self.held.iter_mut().find(|held| &held.id == id)?;
376        if held.state == State::Terminated {
377            // Already over. Reporting it again would send a second terminating NOTIFY for one
378            // subscription, which a subscriber is entitled to find confusing.
379            return None;
380        }
381        held.state = State::Terminated;
382        Some(Subscription::terminated(reason))
383    }
384
385    /// Terminate everything that has run out of time, and say which (§4.1.3, `reason=timeout`).
386    pub fn expire(&mut self, now: u64) -> Vec<Id> {
387        let mut expired = Vec::new();
388        for held in &mut self.held {
389            if held.state != State::Terminated && held.expires_at <= now {
390                held.state = State::Terminated;
391                expired.push(held.id.clone());
392            }
393        }
394        expired
395    }
396
397    /// The `Subscription-State` a NOTIFY for this subscription should carry.
398    ///
399    /// `None` when the subscription is terminated or unknown — **which is what stops a terminated
400    /// subscription being notified**. A notifier that produced an `active` state here for a
401    /// subscription it had ended would resurrect it, and the subscriber would go on believing it
402    /// was watching something.
403    #[must_use]
404    pub fn notify_state(&self, id: &Id, now: u64) -> Option<Subscription> {
405        let held = self.held.iter().find(|held| &held.id == id)?;
406        match held.state {
407            State::Terminated => None,
408            state => Some(Subscription {
409                state,
410                expires: Some(Duration::from_secs(held.expires_at.saturating_sub(now))),
411                reason: None,
412                retry_after: None,
413            }),
414        }
415    }
416
417    /// Forget terminated subscriptions.
418    ///
419    /// Separate from terminating them, and deliberately so: a terminated subscription has to stay
420    /// findable long enough for its terminating NOTIFY to be sent and for a crossing refresh to be
421    /// refused rather than treated as new.
422    pub fn sweep(&mut self) -> usize {
423        let before = self.held.len();
424        self.held.retain(|held| held.state != State::Terminated);
425        before - self.held.len()
426    }
427}
428
429#[cfg(test)]
430#[allow(
431    clippy::unwrap_used,
432    clippy::expect_used,
433    clippy::panic,
434    clippy::indexing_slicing
435)]
436mod tests {
437    use super::*;
438    use bytes::Bytes;
439    use sipx_sip::{Limits, Message, parse_datagram};
440
441    const NOW: u64 = 1_700_000_000;
442
443    fn subscribe(event: &str, expires: Option<u64>, tag: &str) -> Request {
444        subscribe_with_cseq(event, expires, tag, 1)
445    }
446
447    fn subscribe_with_cseq(event: &str, expires: Option<u64>, tag: &str, cseq: u32) -> Request {
448        let expires_line =
449            expires.map_or_else(String::new, |seconds| format!("Expires: {seconds}\r\n"));
450        let text = format!(
451            "SUBSCRIBE sip:alice@sipx.test SIP/2.0\r\n\
452             Via: SIP/2.0/UDP watcher.example;branch=z9hG4bKx\r\n\
453             To: <sip:alice@sipx.test>\r\n\
454             From: <sip:watcher@example.net>;tag={tag}\r\n\
455             Call-ID: sub-1@watcher\r\n\
456             CSeq: {cseq} SUBSCRIBE\r\n\
457             Event: {event}\r\n\
458             {expires_line}\
459             Max-Forwards: 70\r\n\
460             Content-Length: 0\r\n\r\n"
461        );
462        match parse_datagram(Bytes::from(text), &Limits::datagram()).expect("parses") {
463            Message::Request(request) => request,
464            Message::Response(_) => panic!("a request"),
465        }
466    }
467
468    fn notifier() -> Subscriptions {
469        Subscriptions::new(
470            Packages::new().with("dialog").with("presence"),
471            Duration::from_secs(3600),
472        )
473    }
474
475    #[test]
476    fn a_new_subscription_is_refused_at_the_bound_but_a_refresh_is_not() {
477        let mut notifier = notifier().with_capacity(1);
478        let first = subscribe("dialog", Some(600), "w1");
479        assert!(matches!(
480            notifier.on_subscribe(&first, NOW),
481            Answer::Established { .. }
482        ));
483
484        let second = subscribe("presence", Some(600), "w2");
485        assert_eq!(notifier.on_subscribe(&second, NOW), Answer::AtCapacity);
486        assert_eq!(notifier.active(), 1);
487
488        let refresh = subscribe_with_cseq("dialog", Some(600), "w1", 2);
489        assert!(matches!(
490            notifier.on_subscribe(&refresh, NOW + 1),
491            Answer::Refreshed { .. }
492        ));
493    }
494
495    #[test]
496    fn malformed_expiry_and_cseq_do_not_mutate_the_store() {
497        let mut notifier = notifier();
498        for (name, value) in [
499            (HeaderName::Expires, "4294967296"),
500            (HeaderName::CSeq, "not-a-cseq"),
501            (HeaderName::CSeq, "2 MESSAGE"),
502        ] {
503            let mut request = subscribe("dialog", Some(600), "w1");
504            request.headers.remove_all(&name);
505            request
506                .headers
507                .push(sipx_sip::Header::build(name, value).expect("syntactic header"));
508            assert_eq!(notifier.on_subscribe(&request, NOW), Answer::Malformed);
509            assert_eq!(notifier.active(), 0);
510            assert!(notifier.all().is_empty());
511        }
512    }
513
514    /// The story's failing-first test.
515    ///
516    /// Once a subscription is over it stays over. A notifier that produced an `active` state for a
517    /// terminated subscription would resurrect it, and the subscriber would go on believing it was
518    /// watching something that has stopped being watched.
519    #[test]
520    fn a_terminated_subscription_stops_notifying() {
521        let mut notifier = notifier();
522        let request = subscribe("dialog", Some(600), "w1");
523        let Answer::Established { id, .. } = notifier.on_subscribe(&request, NOW) else {
524            panic!("a new subscription");
525        };
526
527        assert!(
528            notifier.notify_state(&id, NOW).is_some(),
529            "an active subscription is notified"
530        );
531
532        notifier
533            .terminate(&id, Reason::NoResource)
534            .expect("the terminating state");
535
536        assert!(
537            notifier.notify_state(&id, NOW).is_none(),
538            "a terminated subscription must produce no further notifications"
539        );
540        assert_eq!(notifier.active(), 0);
541
542        // And a refresh does not bring it back. §4.1.3 makes termination final; a subscriber that
543        // wants another subscription starts a new dialog.
544        let refresh = subscribe_with_cseq("dialog", Some(600), "w1", 2);
545        assert_eq!(
546            notifier.on_subscribe(&refresh, NOW),
547            Answer::Unserved { status: BAD_EVENT },
548            "a terminated subscription must not be refreshed back to life"
549        );
550        assert!(notifier.notify_state(&id, NOW).is_none());
551    }
552
553    #[test]
554    fn a_subscribe_establishes_and_a_second_one_refreshes() {
555        let mut notifier = notifier();
556        let first = notifier.on_subscribe(&subscribe("dialog", Some(600), "w1"), NOW);
557        let Answer::Established { id, expires } = first else {
558            panic!("a new subscription, got {first:?}");
559        };
560        assert_eq!(expires, Duration::from_secs(600));
561        assert_eq!(notifier.active(), 1);
562
563        let again = notifier.on_subscribe(
564            &subscribe_with_cseq("dialog", Some(900), "w1", 2),
565            NOW + 300,
566        );
567        assert_eq!(
568            again,
569            Answer::Refreshed {
570                id: id.clone(),
571                expires: Duration::from_secs(900)
572            }
573        );
574        assert_eq!(
575            notifier.active(),
576            1,
577            "a refresh is not a second subscription"
578        );
579        assert_eq!(
580            notifier
581                .notify_state(&id, NOW + 300)
582                .expect("active")
583                .expires,
584            Some(Duration::from_secs(900)),
585            "the timer was pushed out"
586        );
587    }
588
589    /// §4.2.1.1: "the server MAY shorten the interval but MUST NOT lengthen it".
590    #[test]
591    fn a_notifier_shortens_a_generous_request_to_its_policy() {
592        let mut notifier = notifier();
593        let answer = notifier.on_subscribe(&subscribe("dialog", Some(86400), "w1"), NOW);
594        let Answer::Established { expires, .. } = answer else {
595            panic!("a new subscription");
596        };
597        assert_eq!(expires, Duration::from_secs(3600), "the policy maximum");
598    }
599
600    /// §3.1.1: `Expires: 0` unsubscribes, and §4.2.1.4 still owes a terminating NOTIFY.
601    #[test]
602    fn an_expires_of_zero_ends_the_subscription() {
603        let mut notifier = notifier();
604        let Answer::Established { id, .. } =
605            notifier.on_subscribe(&subscribe("dialog", Some(600), "w1"), NOW)
606        else {
607            panic!("a new subscription");
608        };
609
610        assert_eq!(
611            notifier.on_subscribe(&subscribe_with_cseq("dialog", Some(0), "w1", 2), NOW,),
612            Answer::Unsubscribed { id: id.clone() }
613        );
614        assert_eq!(notifier.active(), 0);
615        assert!(
616            notifier.notify_state(&id, NOW).is_none(),
617            "nothing further is notified after an unsubscribe"
618        );
619        assert!(
620            notifier.all().iter().any(|held| held.id == id),
621            "and it is still findable, so a NOTIFY crossing it finds a terminated subscription \
622             rather than none at all"
623        );
624    }
625
626    /// §4.2.1.1: an unserved package is refused 489 — not accepted and then never notified, which
627    /// a subscriber cannot tell from a slow notifier.
628    #[test]
629    fn a_package_this_notifier_does_not_serve_is_refused_with_489() {
630        let mut notifier = notifier();
631        assert_eq!(
632            notifier.on_subscribe(&subscribe("message-summary", Some(600), "w1"), NOW),
633            Answer::Unserved { status: 489 }
634        );
635        assert_eq!(notifier.active(), 0, "nothing was established");
636    }
637
638    #[test]
639    fn a_subscription_that_runs_out_of_time_is_terminated() {
640        let mut notifier = notifier();
641        let Answer::Established { id, .. } =
642            notifier.on_subscribe(&subscribe("dialog", Some(60), "w1"), NOW)
643        else {
644            panic!("a new subscription");
645        };
646
647        assert!(notifier.expire(NOW + 59).is_empty(), "not yet");
648        assert_eq!(notifier.expire(NOW + 60), vec![id.clone()]);
649        assert_eq!(notifier.active(), 0);
650        assert!(notifier.notify_state(&id, NOW + 60).is_none());
651    }
652
653    /// §4.4.1: several subscriptions may share a dialog if their `Event` differs, so the package
654    /// is part of the identity. Keying on the dialog alone lets a second subscription silently
655    /// replace the first.
656    #[test]
657    fn two_packages_in_one_dialog_are_two_subscriptions() {
658        let mut notifier = notifier();
659        let first = notifier.on_subscribe(&subscribe("dialog", Some(600), "w1"), NOW);
660        let second = notifier.on_subscribe(&subscribe("presence", Some(600), "w1"), NOW);
661        assert!(matches!(first, Answer::Established { .. }));
662        assert!(
663            matches!(second, Answer::Established { .. }),
664            "a different Event in the same dialog is a new subscription, not a refresh"
665        );
666        assert_eq!(notifier.active(), 2);
667    }
668
669    #[test]
670    fn terminating_twice_reports_the_ending_once() {
671        let mut notifier = notifier();
672        let Answer::Established { id, .. } =
673            notifier.on_subscribe(&subscribe("dialog", Some(600), "w1"), NOW)
674        else {
675            panic!("a new subscription");
676        };
677        assert!(notifier.terminate(&id, Reason::GiveUp).is_some());
678        assert!(
679            notifier.terminate(&id, Reason::GiveUp).is_none(),
680            "one subscription gets one terminating NOTIFY"
681        );
682    }
683
684    #[test]
685    fn sweeping_forgets_only_what_has_ended() {
686        let mut notifier = notifier();
687        let Answer::Established { id, .. } =
688            notifier.on_subscribe(&subscribe("dialog", Some(600), "w1"), NOW)
689        else {
690            panic!("a new subscription");
691        };
692        let _ = notifier.on_subscribe(&subscribe("presence", Some(600), "w1"), NOW);
693        notifier.terminate(&id, Reason::Timeout);
694
695        assert_eq!(notifier.sweep(), 1);
696        assert_eq!(notifier.all().len(), 1);
697        assert_eq!(notifier.active(), 1);
698    }
699
700    #[test]
701    fn a_subscribe_without_an_event_is_malformed() {
702        let text = "SUBSCRIBE sip:alice@sipx.test SIP/2.0\r\n\
703             Via: SIP/2.0/UDP watcher.example;branch=z9hG4bKx\r\n\
704             To: <sip:alice@sipx.test>\r\n\
705             From: <sip:watcher@example.net>;tag=w1\r\n\
706             Call-ID: sub-1@watcher\r\n\
707             CSeq: 1 SUBSCRIBE\r\n\
708             Max-Forwards: 70\r\n\
709             Content-Length: 0\r\n\r\n";
710        let request = match parse_datagram(Bytes::from(text), &Limits::datagram()).expect("parses")
711        {
712            Message::Request(request) => request,
713            Message::Response(_) => panic!("a request"),
714        };
715        assert_eq!(notifier().on_subscribe(&request, NOW), Answer::Malformed);
716    }
717
718    #[test]
719    fn a_subscribe_with_no_expires_gets_the_policy_maximum() {
720        let mut notifier = notifier();
721        let answer = notifier.on_subscribe(&subscribe("dialog", None, "w1"), NOW);
722        let Answer::Established { expires, .. } = answer else {
723            panic!("a new subscription");
724        };
725        assert_eq!(expires, Duration::from_secs(3600));
726    }
727
728    #[test]
729    fn replayed_subscribe_cseq_cannot_refresh_or_terminate() {
730        let mut notifier = notifier();
731        let Answer::Established { id, .. } =
732            notifier.on_subscribe(&subscribe_with_cseq("dialog", Some(600), "w1", 20), NOW)
733        else {
734            panic!("a new subscription");
735        };
736        let before = notifier.notify_state(&id, NOW).expect("active");
737
738        for request in [
739            subscribe_with_cseq("dialog", Some(900), "w1", 20),
740            subscribe_with_cseq("dialog", Some(0), "w1", 19),
741        ] {
742            assert_eq!(
743                notifier.on_subscribe(&request, NOW + 30),
744                Answer::OutOfOrder { id: id.clone() }
745            );
746            assert_eq!(notifier.notify_state(&id, NOW), Some(before.clone()));
747            assert_eq!(notifier.active(), 1);
748        }
749
750        assert!(matches!(
751            notifier.on_subscribe(
752                &subscribe_with_cseq("dialog", Some(900), "w1", 21),
753                NOW + 30,
754            ),
755            Answer::Refreshed { .. }
756        ));
757        assert_eq!(
758            notifier
759                .all()
760                .iter()
761                .find(|served| served.id == id)
762                .expect("subscription remains")
763                .remote_cseq,
764            21
765        );
766    }
767
768    #[test]
769    fn event_identity_uses_only_exact_type_and_id_tokens() {
770        let reordered = Id::from_request(&subscribe(
771            "dialog;vendor=one;ID=Opaque-A;mode=full",
772            Some(600),
773            "w1",
774        ))
775        .expect("identity");
776        let differently_ordered = Id::from_request(&subscribe(
777            "dialog;mode=partial;id=Opaque-A;vendor=two",
778            Some(600),
779            "w1",
780        ))
781        .expect("identity");
782        assert_eq!(reordered, differently_ordered);
783        assert_eq!(reordered.event, "dialog;id=Opaque-A");
784
785        let changed_type_case =
786            Id::from_request(&subscribe("Dialog;id=Opaque-A", Some(600), "w1")).expect("identity");
787        let changed_id_case =
788            Id::from_request(&subscribe("dialog;id=opaque-a", Some(600), "w1")).expect("identity");
789        assert_ne!(reordered, changed_type_case, "event-type is byte matched");
790        assert_ne!(reordered, changed_id_case, "id is an opaque token");
791
792        assert!(
793            Id::from_request(&subscribe("dialog;id=one;ID=two", Some(600), "w1")).is_none(),
794            "duplicate identity parameters fail closed"
795        );
796    }
797
798    #[test]
799    fn duplicate_identity_headers_fail_before_first_value_selection() {
800        for name in [HeaderName::CallId, HeaderName::From, HeaderName::Event] {
801            let mut request = subscribe("dialog", Some(600), "w1");
802            let value = request.headers.value(&name).expect("header").into_owned();
803            request
804                .headers
805                .push(sipx_sip::Header::build(name, value).expect("syntactic header"));
806            assert!(Id::from_request(&request).is_none());
807            assert_eq!(notifier().on_subscribe(&request, NOW), Answer::Malformed);
808        }
809    }
810}
sipx_ua/subscribe.rs

sipx_ua/
subscribe.rs
