sipx_sip/transaction/
client.rs

1//! Client transactions: RFC 3261 §17.1, amended by RFC 6026.
2
3use std::time::Duration;
4
5use crate::message::{Header, Message, Method, Request, Response};
6use crate::name::HeaderName;
7use crate::transaction::timing::{Timer, Timers};
8use crate::transaction::{Output, Reason, Reliability, TuEvent};
9
10/// The state of a client transaction.
11///
12/// `Calling` belongs to the INVITE machine and `Trying` to the non-INVITE one; the rest are
13/// shared. `Accepted` is RFC 6026's addition and exists so that a retransmitted 2xx — which
14/// forking proxies produce as a matter of course — still has a transaction to arrive at.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum ClientState {
17    /// INVITE: the request has gone out and nothing has come back.
18    Calling,
19    /// Non-INVITE: the request has gone out and nothing has come back.
20    Trying,
21    /// A provisional response has arrived.
22    Proceeding,
23    /// A final response has arrived; waiting out retransmissions.
24    Completed,
25    /// RFC 6026: a 2xx has arrived and more may follow.
26    Accepted,
27    /// Over.
28    Terminated,
29}
30
31impl ClientState {
32    /// Whether the transaction has finished and can be dropped.
33    #[must_use]
34    pub fn is_terminated(self) -> bool {
35        matches!(self, Self::Terminated)
36    }
37}
38
39/// A client transaction.
40#[derive(Debug)]
41pub struct ClientTransaction {
42    request: Request,
43    is_invite: bool,
44    state: ClientState,
45    reliability: Reliability,
46    timers: Timers,
47    /// The current retransmission interval — Timer A for INVITE, Timer E otherwise.
48    interval: Duration,
49    /// The ACK generated for a non-2xx final response, kept so a retransmitted response can
50    /// be answered with the same ACK rather than a freshly built one.
51    ack: Option<Request>,
52}
53
54impl ClientTransaction {
55    /// Start a client transaction, returning the request to send and the timers to set.
56    #[must_use]
57    pub fn new(request: Request, reliability: Reliability, timers: Timers) -> (Self, Vec<Output>) {
58        let is_invite = request.method == Method::Invite;
59        let mut tx = Self {
60            request,
61            is_invite,
62            state: if is_invite {
63                ClientState::Calling
64            } else {
65                ClientState::Trying
66            },
67            reliability,
68            timers,
69            interval: timers.t1,
70            ack: None,
71        };
72
73        let mut out = vec![Output::send(Message::Request(tx.request.clone()))];
74        if !reliability.is_reliable() {
75            out.push(Output::SetTimer {
76                timer: if is_invite { Timer::A } else { Timer::E },
77                after: tx.interval,
78            });
79        }
80        out.push(Output::SetTimer {
81            timer: if is_invite { Timer::B } else { Timer::F },
82            after: timers.timeout(),
83        });
84        tx.state = if is_invite {
85            ClientState::Calling
86        } else {
87            ClientState::Trying
88        };
89        (tx, out)
90    }
91
92    /// The current state.
93    #[must_use]
94    pub fn state(&self) -> ClientState {
95        self.state
96    }
97
98    /// The request this transaction was created for.
99    #[must_use]
100    pub fn request(&self) -> &Request {
101        &self.request
102    }
103
104    /// Feed a response in.
105    pub fn on_response(&mut self, response: Response) -> Vec<Output> {
106        if self.is_invite {
107            self.invite_response(response)
108        } else {
109            self.non_invite_response(response)
110        }
111    }
112
113    fn invite_response(&mut self, response: Response) -> Vec<Output> {
114        let status = response.status;
115        match self.state {
116            ClientState::Calling | ClientState::Proceeding => {
117                let from_calling = self.state == ClientState::Calling;
118                let mut out = Vec::new();
119
120                if status.is_provisional() {
121                    if from_calling {
122                        // Provisional means the far end is alive: stop retransmitting, and
123                        // drop the timeout with it. RFC 3261 §17.1.1.2 fires Timer B from
124                        // Calling only, because a phone may legitimately ring for far longer
125                        // than 64*T1 and the answer must still find its transaction.
126                        out.push(Output::ClearTimer(Timer::A));
127                        out.push(Output::ClearTimer(Timer::B));
128                    }
129                    self.state = ClientState::Proceeding;
130                    out.push(Output::to_tu(TuEvent::Response(Box::new(response))));
131                    return out;
132                }
133
134                if from_calling {
135                    out.push(Output::ClearTimer(Timer::A));
136                }
137                out.push(Output::ClearTimer(Timer::B));
138
139                if status.is_success() {
140                    // No ACK from here. The ACK for a 2xx is a separate transaction the TU
141                    // must build, because only the TU knows the dialog's route set. Sending
142                    // one here would use the wrong Request-URI and the wrong route.
143                    self.state = ClientState::Accepted;
144                    out.push(Output::to_tu(TuEvent::Response(Box::new(response))));
145                    out.push(Output::SetTimer {
146                        timer: Timer::M,
147                        after: self.timers.timeout(),
148                    });
149                } else {
150                    // The ACK for a non-2xx *is* part of this transaction and reuses its
151                    // branch, so the far end matches it to the INVITE it is acknowledging.
152                    let ack = make_ack(&self.request, &response);
153                    out.push(Output::send(Message::Request(ack.clone())));
154                    self.ack = Some(ack);
155                    self.state = ClientState::Completed;
156                    out.push(Output::to_tu(TuEvent::Response(Box::new(response))));
157                    out.push(Output::SetTimer {
158                        timer: Timer::D,
159                        after: self.timers.timer_d(self.reliability),
160                    });
161                }
162                out
163            }
164            ClientState::Completed => {
165                // A retransmitted final response gets the same ACK again — and the TU is not
166                // told, because it has already dealt with this response.
167                self.ack
168                    .as_ref()
169                    .map(|ack| Output::send(Message::Request(ack.clone())))
170                    .into_iter()
171                    .collect()
172            }
173            ClientState::Accepted => {
174                if status.is_success() {
175                    // RFC 6026: another 2xx. A forking proxy produces these routinely, and
176                    // each one is a distinct answered branch the TU has to know about.
177                    vec![Output::to_tu(TuEvent::Response(Box::new(response)))]
178                } else {
179                    Vec::new()
180                }
181            }
182            ClientState::Trying | ClientState::Terminated => Vec::new(),
183        }
184    }
185
186    fn non_invite_response(&mut self, response: Response) -> Vec<Output> {
187        match self.state {
188            ClientState::Trying | ClientState::Proceeding => {
189                if response.status.is_provisional() {
190                    self.state = ClientState::Proceeding;
191                    return vec![Output::to_tu(TuEvent::Response(Box::new(response)))];
192                }
193                self.state = ClientState::Completed;
194                vec![
195                    Output::ClearTimer(Timer::E),
196                    Output::ClearTimer(Timer::F),
197                    Output::to_tu(TuEvent::Response(Box::new(response))),
198                    Output::SetTimer {
199                        timer: Timer::K,
200                        after: self.timers.absorb(self.reliability),
201                    },
202                ]
203            }
204            // Retransmissions in Completed are absorbed outright: the TU has its answer.
205            _ => Vec::new(),
206        }
207    }
208
209    /// Feed a fired timer in.
210    pub fn on_timer(&mut self, timer: Timer) -> Vec<Output> {
211        match (self.state, timer) {
212            (ClientState::Calling, Timer::A) => {
213                self.interval = self.timers.double(self.interval);
214                vec![
215                    Output::send(Message::Request(self.request.clone())),
216                    Output::SetTimer {
217                        timer: Timer::A,
218                        after: self.interval,
219                    },
220                ]
221            }
222            (ClientState::Trying, Timer::E) => {
223                self.interval = self.timers.double_capped(self.interval);
224                vec![
225                    Output::send(Message::Request(self.request.clone())),
226                    Output::SetTimer {
227                        timer: Timer::E,
228                        after: self.interval,
229                    },
230                ]
231            }
232            (ClientState::Proceeding, Timer::E) if !self.is_invite => {
233                // Once provisional responses are flowing the interval stops backing off and
234                // sits at T2: the far end is clearly alive, so this is keep-alive rather than
235                // recovery.
236                self.interval = self.timers.t2;
237                vec![
238                    Output::send(Message::Request(self.request.clone())),
239                    Output::SetTimer {
240                        timer: Timer::E,
241                        after: self.interval,
242                    },
243                ]
244            }
245            // Timer B belongs to the INVITE machine and fires only from Calling (§17.1.1.2):
246            // once a provisional has arrived the INVITE transaction has no timeout of its own,
247            // which §16.6 item 11 states outright and is the reason Timer C exists at proxies.
248            // Timer F belongs to the non-INVITE machine and does carry over into Proceeding
249            // (§17.1.2.2) — the asymmetry is deliberate, not an oversight.
250            (ClientState::Calling, Timer::B)
251            | (ClientState::Trying | ClientState::Proceeding, Timer::F) => {
252                self.state = ClientState::Terminated;
253                vec![
254                    Output::to_tu(TuEvent::Timeout),
255                    Output::Terminated(Reason::Timeout),
256                ]
257            }
258            (ClientState::Completed, Timer::D | Timer::K) | (ClientState::Accepted, Timer::M) => {
259                self.state = ClientState::Terminated;
260                vec![Output::Terminated(Reason::Completed)]
261            }
262            _ => Vec::new(),
263        }
264    }
265
266    /// The transport could not deliver the request.
267    pub fn on_transport_error(&mut self) -> Vec<Output> {
268        if self.state.is_terminated() {
269            return Vec::new();
270        }
271        self.state = ClientState::Terminated;
272        vec![
273            Output::to_tu(TuEvent::TransportError),
274            Output::Terminated(Reason::TransportError),
275        ]
276    }
277}
278
279/// Build the ACK for a non-2xx final response (RFC 3261 §17.1.1.3).
280///
281/// It reuses the INVITE's `Via` — the same branch — because it is part of the same
282/// transaction, takes `To` from the *response* so the tag the far end chose is echoed back,
283/// and copies the `Route` set from the request so it follows the same path.
284fn make_ack(request: &Request, response: &Response) -> Request {
285    let mut ack = Request::new(Method::Ack, request.uri.clone());
286
287    // Exactly one Via: the topmost one from the request.
288    if let Some(via) = request.headers.get(&HeaderName::Via) {
289        ack.headers.push(via.clone());
290    }
291    if let Some(from) = request.headers.get(&HeaderName::From) {
292        ack.headers.push(from.clone());
293    }
294    // To comes from the response: it carries the tag the far end assigned, and an ACK without
295    // it will not match anything at the far end.
296    if let Some(to) = response.headers.get(&HeaderName::To) {
297        ack.headers.push(to.clone());
298    }
299    if let Some(call_id) = request.headers.get(&HeaderName::CallId) {
300        ack.headers.push(call_id.clone());
301    }
302    for route in request.headers.get_all(&HeaderName::Route) {
303        ack.headers.push(route.clone());
304    }
305    // RFC 3261 §8.1.1: Max-Forwards is mandatory in every request, the ACK included. A proxy
306    // is entitled to reject one that lacks it, which restarts the far end's retransmissions.
307    ack.headers.push(Header::new_unchecked(
308        HeaderName::MaxForwards,
309        bytes::Bytes::from_static(b"70"),
310    ));
311
312    // Same sequence number, method ACK.
313    let sequence = request
314        .headers
315        .value(&HeaderName::CSeq)
316        .and_then(|v| {
317            let digits: Vec<u8> = v.iter().copied().take_while(u8::is_ascii_digit).collect();
318            String::from_utf8(digits).ok()
319        })
320        .unwrap_or_default();
321    let mut cseq = sequence.into_bytes();
322    cseq.extend_from_slice(b" ACK");
323    ack.headers.push(Header::new_unchecked(
324        HeaderName::CSeq,
325        bytes::Bytes::from(cseq),
326    ));
327    ack.headers.push(Header::new_unchecked(
328        HeaderName::ContentLength,
329        bytes::Bytes::from_static(b"0"),
330    ));
331
332    ack
333}
sipx_sip/transaction/client.rs

sipx_sip/transaction/
client.rs
