ftth_rsipstack/transaction/
transaction.rs

1use super::endpoint::EndpointInnerRef;
2use super::key::TransactionKey;
3use super::{SipConnection, TransactionState, TransactionTimer, TransactionType};
4use crate::dialog::DialogId;
5use crate::rsip;
6use crate::transaction::make_tag;
7use crate::transport::SipAddr;
8use crate::{Error, Result};
9use rsip::headers::ContentLength;
10use rsip::message::HasHeaders;
11use rsip::prelude::HeadersExt;
12use rsip::{Header, Method, Request, Response, SipMessage, StatusCode, StatusCodeKind};
13use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
14use tracing::{debug, info, trace};
15
16pub type TransactionEventReceiver = UnboundedReceiver<TransactionEvent>;
17pub type TransactionEventSender = UnboundedSender<TransactionEvent>;
18
19/// SIP Transaction Events
20///
21/// `TransactionEvent` represents the various events that can occur during
22/// a SIP transaction's lifecycle. These events drive the transaction state machine
23/// and coordinate between the transaction layer and transaction users.
24///
25/// # Events
26///
27/// * `Received` - A SIP message was received for this transaction
28/// * `Timer` - A transaction timer has fired
29/// * `Respond` - Request to send a response (server transactions only)
30/// * `Terminate` - Request to terminate the transaction
31///
32/// # Examples
33///
34/// ```rust,no_run
35/// use ftth_rsipstack::transaction::transaction::TransactionEvent;
36/// use rsip::SipMessage;
37///
38/// # fn handle_event(event: TransactionEvent) {
39/// match event {
40///     TransactionEvent::Received(msg, conn) => {
41///         // Process received SIP message
42///     },
43///     TransactionEvent::Timer(timer) => {
44///         // Handle timer expiration
45///     },
46///     TransactionEvent::Respond(response) => {
47///         // Send response
48///     },
49///     TransactionEvent::Terminate => {
50///         // Clean up transaction
51///     }
52/// }
53/// # }
54/// ```
55pub enum TransactionEvent {
56    Received(SipMessage, Option<SipConnection>),
57    Timer(TransactionTimer),
58    Respond(Response),
59    Terminate,
60}
61
62/// SIP Transaction
63///
64/// `Transaction` implements the SIP transaction layer as defined in RFC 3261.
65/// A transaction consists of a client transaction (sends requests) or server
66/// transaction (receives requests) that handles the reliable delivery of SIP
67/// messages and manages retransmissions and timeouts.
68///
69/// # Key Features
70///
71/// * Automatic retransmission handling
72/// * Timer management per RFC 3261
73/// * State machine implementation
74/// * Reliable message delivery
75/// * Connection management
76///
77/// # Transaction Types
78///
79/// * `ClientInvite` - Client INVITE transaction
80/// * `ClientNonInvite` - Client non-INVITE transaction  
81/// * `ServerInvite` - Server INVITE transaction
82/// * `ServerNonInvite` - Server non-INVITE transaction
83///
84/// # State Machine
85///
86/// Transactions follow the state machines defined in RFC 3261:
87/// * Calling → Trying → Proceeding → Completed → Terminated
88/// * Additional states for INVITE transactions: Confirmed
89///
90/// # Examples
91///
92/// ```rust,no_run
93/// use ftth_rsipstack::rsip;
94/// use ftth_rsipstack::transaction::{
95///     transaction::Transaction,
96///     key::{TransactionKey, TransactionRole},
97/// };
98/// use rsip::SipMessage;
99///
100/// # async fn example() -> ftth_rsipstack::Result<()> {
101/// # let endpoint_inner = todo!();
102/// # let connection = None;
103/// // Create a mock request
104/// let request = rsip::Request {
105///     method: rsip::Method::Register,
106///     uri: rsip::Uri::try_from("sip:example.com")?,
107///     headers: vec![
108///         rsip::Header::Via("SIP/2.0/UDP example.com:5060;branch=z9hG4bKnashds".into()),
109///         rsip::Header::CSeq("1 REGISTER".into()),
110///         rsip::Header::From("Alice <sip:alice@example.com>;tag=1928301774".into()),
111///         rsip::Header::CallId("a84b4c76e66710@pc33.atlanta.com".into()),
112///     ].into(),
113///     version: rsip::Version::V2,
114///     body: Default::default(),
115/// };
116/// let key = TransactionKey::from_request(&request, TransactionRole::Client)?;
117///
118/// // Create a client transaction
119/// let mut transaction = Transaction::new_client(
120///     key,
121///     request,
122///     endpoint_inner,
123///     connection
124/// );
125///
126/// // Send the request
127/// transaction.send().await?;
128///
129/// // Receive responses
130/// while let Some(message) = transaction.receive().await {
131///     match message {
132///         SipMessage::Response(response) => {
133///             // Handle response
134///         },
135///         _ => {}
136///     }
137/// }
138/// # Ok(())
139/// # }
140/// ```
141///
142/// # Timer Handling
143///
144/// The transaction automatically manages SIP timers:
145/// * Timer A: Retransmission timer for unreliable transports
146/// * Timer B: Transaction timeout timer
147/// * Timer D: Wait time for response retransmissions
148/// * Timer E: Non-INVITE retransmission timer
149/// * Timer F: Non-INVITE transaction timeout
150/// * Timer G: INVITE response retransmission timer
151/// * Timer K: Wait time for ACK
152pub struct Transaction {
153    pub transaction_type: TransactionType,
154    pub key: TransactionKey,
155    pub original: Request,
156    pub destination: Option<SipAddr>,
157    pub state: TransactionState,
158    pub endpoint_inner: EndpointInnerRef,
159    pub connection: Option<SipConnection>,
160    pub last_response: Option<Response>,
161    pub last_ack: Option<Request>,
162    pub tu_receiver: TransactionEventReceiver,
163    pub tu_sender: TransactionEventSender,
164    pub timer_a: Option<u64>,
165    pub timer_b: Option<u64>,
166    pub timer_c: Option<u64>,
167    pub timer_d: Option<u64>,
168    pub timer_k: Option<u64>, // server invite only
169    pub timer_g: Option<u64>, // server invite only
170    is_cleaned_up: bool,
171}
172
173impl Transaction {
174    fn new(
175        transaction_type: TransactionType,
176        key: TransactionKey,
177        original: Request,
178        connection: Option<SipConnection>,
179        endpoint_inner: EndpointInnerRef,
180    ) -> Self {
181        let (tu_sender, tu_receiver) = unbounded_channel();
182        let state = if matches!(
183            transaction_type,
184            TransactionType::ServerInvite | TransactionType::ServerNonInvite
185        ) {
186            TransactionState::Trying
187        } else {
188            TransactionState::Nothing
189        };
190        trace!(%key, %state, "transaction created");
191        let tx = Self {
192            transaction_type,
193            endpoint_inner,
194            connection,
195            key,
196            original,
197            destination: None,
198            state,
199            last_response: None,
200            last_ack: None,
201            timer_a: None,
202            timer_b: None,
203            timer_c: None,
204            timer_d: None,
205            timer_k: None,
206            timer_g: None,
207            tu_receiver,
208            tu_sender,
209            is_cleaned_up: false,
210        };
211        tx.endpoint_inner
212            .attach_transaction(&tx.key, tx.tu_sender.clone());
213        tx
214    }
215
216    pub fn new_client(
217        key: TransactionKey,
218        original: Request,
219        endpoint_inner: EndpointInnerRef,
220        connection: Option<SipConnection>,
221    ) -> Self {
222        let tx_type = match original.method {
223            Method::Invite => TransactionType::ClientInvite,
224            _ => TransactionType::ClientNonInvite,
225        };
226        Transaction::new(tx_type, key, original, connection, endpoint_inner)
227    }
228
229    pub fn new_server(
230        key: TransactionKey,
231        original: Request,
232        endpoint_inner: EndpointInnerRef,
233        connection: Option<SipConnection>,
234    ) -> Self {
235        let tx_type = match original.method {
236            Method::Invite | Method::Ack => TransactionType::ServerInvite,
237            _ => TransactionType::ServerNonInvite,
238        };
239        Transaction::new(tx_type, key, original, connection, endpoint_inner)
240    }
241    // send client request
242    pub async fn send(&mut self) -> Result<()> {
243        match self.transaction_type {
244            TransactionType::ClientInvite | TransactionType::ClientNonInvite => {}
245            _ => {
246                return Err(Error::TransactionError(
247                    "send is only valid for client transactions".to_string(),
248                    self.key.clone(),
249                ));
250            }
251        }
252        if self.connection.is_none() {
253            let target_uri = match &self.destination {
254                Some(addr) => addr,
255                None => &SipAddr::try_from(&self.original.uri)?,
256            };
257            let (connection, resolved_addr) = self
258                .endpoint_inner
259                .transport_layer
260                .lookup(target_uri, Some(&self.key))
261                .await?;
262            // For UDP, we need to store the resolved destination address
263            if !connection.is_reliable() {
264                self.destination.replace(resolved_addr);
265            }
266            self.connection.replace(connection);
267        }
268
269        let connection = self.connection.as_ref().ok_or(Error::TransactionError(
270            "no connection found".to_string(),
271            self.key.clone(),
272        ))?;
273        let content_length_header =
274            Header::ContentLength(ContentLength::from(self.original.body().len() as u32));
275        self.original
276            .headers_mut()
277            .unique_push(content_length_header);
278
279        let message = if let Some(ref inspector) = self.endpoint_inner.inspector {
280            inspector.before_send(self.original.to_owned().into())
281        } else {
282            self.original.to_owned().into()
283        };
284
285        connection.send(message, self.destination.as_ref()).await?;
286        self.transition(TransactionState::Calling).map(|_| ())
287    }
288
289    pub async fn reply_with(
290        &mut self,
291        status_code: StatusCode,
292        headers: Vec<rsip::Header>,
293        body: Option<Vec<u8>>,
294    ) -> Result<()> {
295        match status_code.kind() {
296            rsip::StatusCodeKind::Provisional => {}
297            _ => {
298                let to = self.original.to_header()?;
299                if to.tag()?.is_none() {
300                    self.original
301                        .headers
302                        .unique_push(to.clone().with_tag(make_tag())?.into());
303                }
304            }
305        }
306        let mut resp = self
307            .endpoint_inner
308            .make_response(&self.original, status_code, body);
309        resp.headers.extend(headers);
310        self.respond(resp).await
311    }
312    /// Quick reply with status code
313    pub async fn reply(&mut self, status_code: StatusCode) -> Result<()> {
314        self.reply_with(status_code, vec![], None).await
315    }
316    // send server response
317    pub async fn respond(&mut self, response: Response) -> Result<()> {
318        match self.transaction_type {
319            TransactionType::ServerInvite | TransactionType::ServerNonInvite => {}
320            _ => {
321                return Err(Error::TransactionError(
322                    "respond is only valid for server transactions".to_string(),
323                    self.key.clone(),
324                ));
325            }
326        }
327
328        let new_state = match response.status_code.kind() {
329            rsip::StatusCodeKind::Provisional => match response.status_code {
330                rsip::StatusCode::Trying => TransactionState::Trying,
331                _ => TransactionState::Proceeding,
332            },
333            _ => match self.transaction_type {
334                TransactionType::ServerInvite => TransactionState::Completed,
335                _ => TransactionState::Terminated,
336            },
337        };
338        // check an transition to new state
339        self.can_transition(&new_state)?;
340
341        let connection = self.connection.as_ref().ok_or(Error::TransactionError(
342            "no connection found".to_string(),
343            self.key.clone(),
344        ))?;
345
346        let response = if let Some(ref inspector) = self.endpoint_inner.inspector {
347            inspector.before_send(response.clone().to_owned().into())
348        } else {
349            response.to_owned().into()
350        };
351        trace!(key = %self.key, "responding with {}", response);
352
353        match response.clone() {
354            SipMessage::Response(resp) => self.last_response.replace(resp),
355            _ => None,
356        };
357        connection.send(response, self.destination.as_ref()).await?;
358        self.transition(new_state).map(|_| ())
359    }
360
361    fn can_transition(&self, target: &TransactionState) -> Result<()> {
362        match (&self.state, target) {
363            (&TransactionState::Nothing, &TransactionState::Calling)
364            | (&TransactionState::Nothing, &TransactionState::Trying)
365            | (&TransactionState::Nothing, &TransactionState::Proceeding)
366            | (&TransactionState::Nothing, &TransactionState::Terminated)
367            | (&TransactionState::Calling, &TransactionState::Trying)
368            | (&TransactionState::Calling, &TransactionState::Proceeding)
369            | (&TransactionState::Calling, &TransactionState::Completed)
370            | (&TransactionState::Calling, &TransactionState::Terminated)
371            | (&TransactionState::Trying, &TransactionState::Trying) // retransmission
372            | (&TransactionState::Trying, &TransactionState::Proceeding)
373            | (&TransactionState::Trying, &TransactionState::Completed)
374            | (&TransactionState::Trying, &TransactionState::Confirmed)
375            | (&TransactionState::Trying, &TransactionState::Terminated)
376            | (&TransactionState::Proceeding, &TransactionState::Completed)
377            | (&TransactionState::Proceeding, &TransactionState::Confirmed)
378            | (&TransactionState::Proceeding, &TransactionState::Terminated)
379            | (&TransactionState::Completed, &TransactionState::Confirmed)
380            | (&TransactionState::Completed, &TransactionState::Terminated)
381            | (&TransactionState::Confirmed, &TransactionState::Terminated) => Ok(()),
382            _ => {
383                Err(Error::TransactionError(
384                    format!(
385                        "invalid state transition from {} to {}",
386                        self.state, target
387                    ),
388                    self.key.clone(),
389                ))
390            }
391        }
392    }
393    pub async fn send_cancel(&mut self, cancel: Request) -> Result<()> {
394        if self.transaction_type != TransactionType::ClientInvite {
395            return Err(Error::TransactionError(
396                "send_cancel is only valid for client invite transactions".to_string(),
397                self.key.clone(),
398            ));
399        }
400
401        match self.state {
402            TransactionState::Calling | TransactionState::Trying | TransactionState::Proceeding => {
403                if let Some(connection) = &self.connection {
404                    let cancel = if let Some(ref inspector) = self.endpoint_inner.inspector {
405                        inspector.before_send(cancel.to_owned().into())
406                    } else {
407                        cancel.to_owned().into()
408                    };
409
410                    connection.send(cancel, self.destination.as_ref()).await?;
411                }
412                self.transition(TransactionState::Completed).map(|_| ())
413            }
414            _ => Err(Error::TransactionError(
415                format!("invalid state for sending CANCEL {:?}", self.state),
416                self.key.clone(),
417            )),
418        }
419    }
420
421    pub async fn send_ack(&mut self, connection: Option<SipConnection>) -> Result<()> {
422        if self.transaction_type != TransactionType::ClientInvite {
423            return Err(Error::TransactionError(
424                "send_ack is only valid for client invite transactions".to_string(),
425                self.key.clone(),
426            ));
427        }
428
429        match self.state {
430            TransactionState::Completed => {} // must be in completed state, to send ACK
431            _ => {
432                return Err(Error::TransactionError(
433                    format!("invalid state for sending ACK {:?}", self.state),
434                    self.key.clone(),
435                ));
436            }
437        }
438        let ack = match self.last_ack.clone() {
439            Some(ack) => ack,
440            None => match self.last_response {
441                Some(ref resp) => self
442                    .endpoint_inner
443                    .make_ack(self.original.uri.clone(), resp)?,
444                None => {
445                    return Err(Error::TransactionError(
446                        "no last response found to send ACK".to_string(),
447                        self.key.clone(),
448                    ));
449                }
450            },
451        };
452
453        let ack = if let Some(ref inspector) = self.endpoint_inner.inspector {
454            inspector.before_send(ack.to_owned().into())
455        } else {
456            ack.to_owned().into()
457        };
458        match ack.clone() {
459            SipMessage::Request(ack) => self.last_ack.replace(ack),
460            _ => None,
461        };
462        if let Some(conn) = connection {
463            conn.send(ack, self.destination.as_ref()).await?;
464        }
465        // client send ack and transition to Terminated
466        self.transition(TransactionState::Terminated).map(|_| ())
467    }
468
469    pub async fn receive(&mut self) -> Option<SipMessage> {
470        while let Some(event) = self.tu_receiver.recv().await {
471            match event {
472                TransactionEvent::Received(msg, connection) => {
473                    if let Some(msg) = match msg {
474                        SipMessage::Request(req) => self.on_received_request(req, connection).await,
475                        SipMessage::Response(resp) => {
476                            self.on_received_response(resp, connection).await
477                        }
478                    } {
479                        if let Some(ref inspector) = self.endpoint_inner.inspector {
480                            return Some(inspector.after_received(msg));
481                        }
482                        return Some(msg);
483                    }
484                }
485                TransactionEvent::Timer(t) => {
486                    self.on_timer(t).await.ok();
487                }
488                TransactionEvent::Respond(response) => {
489                    self.respond(response).await.ok();
490                }
491                TransactionEvent::Terminate => {
492                    info!("received terminate event");
493                    return None;
494                }
495            }
496        }
497        None
498    }
499
500    pub async fn send_trying(&mut self) -> Result<()> {
501        let response =
502            self.endpoint_inner
503                .make_response(&self.original, rsip::StatusCode::Trying, None);
504        self.respond(response).await
505    }
506
507    pub fn is_terminated(&self) -> bool {
508        self.state == TransactionState::Terminated
509    }
510}
511
512impl Transaction {
513    fn inform_tu_response(&mut self, response: Response) -> Result<()> {
514        self.tu_sender
515            .send(TransactionEvent::Received(
516                SipMessage::Response(response),
517                None,
518            ))
519            .map_err(|e| Error::TransactionError(e.to_string(), self.key.clone()))
520    }
521
522    async fn on_received_request(
523        &mut self,
524        req: Request,
525        connection: Option<SipConnection>,
526    ) -> Option<SipMessage> {
527        match self.transaction_type {
528            TransactionType::ClientInvite | TransactionType::ClientNonInvite => return None,
529            _ => {}
530        }
531
532        if self.connection.is_none() && connection.is_some() {
533            self.connection = connection;
534        }
535        if req.method == Method::Cancel {
536            match self.state {
537                TransactionState::Proceeding
538                | TransactionState::Trying
539                | TransactionState::Completed => {
540                    if let Some(connection) = &self.connection {
541                        let resp = self
542                            .endpoint_inner
543                            .make_response(&req, StatusCode::OK, None);
544
545                        let resp = if let Some(ref inspector) = self.endpoint_inner.inspector {
546                            inspector.before_send(resp.into())
547                        } else {
548                            resp.into()
549                        };
550
551                        connection.send(resp, self.destination.as_ref()).await.ok();
552                    }
553                    return Some(req.into()); // into dialog
554                }
555                _ => {
556                    if let Some(connection) = &self.connection {
557                        let resp = self.endpoint_inner.make_response(
558                            &req,
559                            StatusCode::CallTransactionDoesNotExist,
560                            None,
561                        );
562                        let resp = if let Some(ref inspector) = self.endpoint_inner.inspector {
563                            inspector.before_send(resp.into())
564                        } else {
565                            resp.into()
566                        };
567                        connection.send(resp, self.destination.as_ref()).await.ok();
568                    }
569                }
570            };
571            return None;
572        }
573
574        match self.state {
575            TransactionState::Trying | TransactionState::Proceeding => {
576                // retransmission of last response
577                if let Some(last_response) = &self.last_response {
578                    self.respond(last_response.to_owned()).await.ok();
579                }
580            }
581            TransactionState::Completed | TransactionState::Confirmed => {
582                if req.method == Method::Ack {
583                    self.transition(TransactionState::Confirmed).ok();
584                    return Some(req.into());
585                }
586            }
587            _ => {}
588        }
589        None
590    }
591
592    async fn on_received_response(
593        &mut self,
594        resp: Response,
595        connection: Option<SipConnection>,
596    ) -> Option<SipMessage> {
597        match self.transaction_type {
598            TransactionType::ServerInvite | TransactionType::ServerNonInvite => return None,
599            _ => {}
600        }
601        let new_state = match resp.status_code.kind() {
602            StatusCodeKind::Provisional => {
603                if resp.status_code == rsip::StatusCode::Trying {
604                    TransactionState::Trying
605                } else {
606                    TransactionState::Proceeding
607                }
608            }
609            _ => {
610                if self.transaction_type == TransactionType::ClientInvite {
611                    TransactionState::Completed
612                } else {
613                    TransactionState::Terminated
614                }
615            }
616        };
617
618        self.can_transition(&new_state).ok()?;
619        if self.state == new_state {
620            // ignore duplicate response
621            return None;
622        }
623
624        self.last_response.replace(resp.clone());
625        self.transition(new_state).ok();
626        self.send_ack(connection).await.ok(); // send ACK for client invite
627        Some(SipMessage::Response(resp))
628    }
629
630    async fn on_timer(&mut self, timer: TransactionTimer) -> Result<()> {
631        match self.state {
632            TransactionState::Calling | TransactionState::Trying => {
633                if matches!(
634                    self.transaction_type,
635                    TransactionType::ClientInvite | TransactionType::ClientNonInvite
636                ) {
637                    if let TransactionTimer::TimerA(key, duration) = timer {
638                        // Resend the INVITE request
639                        if let Some(connection) = &self.connection {
640                            let retry_message =
641                                if let Some(ref inspector) = self.endpoint_inner.inspector {
642                                    inspector.before_send(self.original.to_owned().into())
643                                } else {
644                                    self.original.to_owned().into()
645                                };
646                            connection
647                                .send(retry_message, self.destination.as_ref())
648                                .await?;
649                        }
650                        // Restart Timer A with an upper limit
651                        let duration = (duration * 2).min(self.endpoint_inner.option.t1x64);
652                        let timer_a = self
653                            .endpoint_inner
654                            .timers
655                            .timeout(duration, TransactionTimer::TimerA(key, duration));
656                        self.timer_a.replace(timer_a);
657                    } else if let TransactionTimer::TimerB(_) = timer {
658                        let timeout_response = self.endpoint_inner.make_response(
659                            &self.original,
660                            rsip::StatusCode::RequestTimeout,
661                            None,
662                        );
663                        self.inform_tu_response(timeout_response)?;
664                        self.transition(TransactionState::Terminated)?;
665                    }
666                }
667            }
668            TransactionState::Proceeding => {
669                if let TransactionTimer::TimerC(_) = timer {
670                    // Inform TU about timeout
671                    let timeout_response = self.endpoint_inner.make_response(
672                        &self.original,
673                        rsip::StatusCode::RequestTimeout,
674                        None,
675                    );
676                    self.inform_tu_response(timeout_response)?;
677                    self.transition(TransactionState::Terminated)?;
678                }
679            }
680            TransactionState::Completed => {
681                if let TransactionTimer::TimerG(key, duration) = timer {
682                    // resend the response
683                    if let Some(last_response) = &self.last_response {
684                        if let Some(connection) = &self.connection {
685                            let last_response =
686                                if let Some(ref inspector) = self.endpoint_inner.inspector {
687                                    inspector.before_send(last_response.to_owned().into())
688                                } else {
689                                    last_response.to_owned().into()
690                                };
691                            connection
692                                .send(last_response, self.destination.as_ref())
693                                .await?;
694                        }
695                    }
696                    // restart Timer G with an upper limit
697                    let duration = (duration * 2).min(self.endpoint_inner.option.t1x64);
698                    let timer_g = self
699                        .endpoint_inner
700                        .timers
701                        .timeout(duration, TransactionTimer::TimerG(key, duration));
702                    self.timer_g.replace(timer_g);
703                } else if let TransactionTimer::TimerD(_) = timer {
704                    self.transition(TransactionState::Terminated)?;
705                } else if let TransactionTimer::TimerK(_) = timer {
706                    self.transition(TransactionState::Terminated)?;
707                }
708            }
709            TransactionState::Confirmed => {
710                if let TransactionTimer::TimerK(_) = timer {
711                    self.transition(TransactionState::Terminated)?;
712                }
713            }
714            _ => {}
715        }
716        Ok(())
717    }
718
719    fn transition(&mut self, state: TransactionState) -> Result<TransactionState> {
720        if self.state == state {
721            return Ok(self.state.clone());
722        }
723        match state {
724            TransactionState::Nothing => {}
725            TransactionState::Calling => {
726                let connection = self.connection.as_ref().ok_or(Error::TransactionError(
727                    "no connection found".to_string(),
728                    self.key.clone(),
729                ))?;
730
731                if matches!(
732                    self.transaction_type,
733                    TransactionType::ClientInvite | TransactionType::ClientNonInvite
734                ) {
735                    if !connection.is_reliable() {
736                        let timer_a = self.endpoint_inner.timers.timeout(
737                            self.endpoint_inner.option.t1,
738                            TransactionTimer::TimerA(
739                                self.key.clone(),
740                                self.endpoint_inner.option.t1,
741                            ),
742                        );
743                        self.timer_a.replace(timer_a);
744                    }
745                    self.timer_b.replace(self.endpoint_inner.timers.timeout(
746                        self.endpoint_inner.option.t1x64,
747                        TransactionTimer::TimerB(self.key.clone()),
748                    ));
749                }
750            }
751            TransactionState::Trying | TransactionState::Proceeding => {
752                self.timer_a
753                    .take()
754                    .map(|id| self.endpoint_inner.timers.cancel(id));
755                if matches!(self.transaction_type, TransactionType::ClientInvite) {
756                    self.timer_b
757                        .take()
758                        .map(|id| self.endpoint_inner.timers.cancel(id));
759                    if self.timer_c.is_none() {
760                        // start Timer C for client invite only
761                        let timer_c = self.endpoint_inner.timers.timeout(
762                            self.endpoint_inner.option.timerc,
763                            TransactionTimer::TimerC(self.key.clone()),
764                        );
765                        self.timer_c.replace(timer_c);
766                    }
767                }
768            }
769            TransactionState::Completed => {
770                self.timer_a
771                    .take()
772                    .map(|id| self.endpoint_inner.timers.cancel(id));
773                self.timer_b
774                    .take()
775                    .map(|id| self.endpoint_inner.timers.cancel(id));
776                self.timer_c
777                    .take()
778                    .map(|id| self.endpoint_inner.timers.cancel(id));
779
780                if self.transaction_type == TransactionType::ServerInvite {
781                    // start Timer G for server invite only
782                    let connection = self.connection.as_ref().ok_or(Error::TransactionError(
783                        "no connection found".to_string(),
784                        self.key.clone(),
785                    ))?;
786                    if !connection.is_reliable() {
787                        let timer_g = self.endpoint_inner.timers.timeout(
788                            self.endpoint_inner.option.t1,
789                            TransactionTimer::TimerG(
790                                self.key.clone(),
791                                self.endpoint_inner.option.t1,
792                            ),
793                        );
794                        self.timer_g.replace(timer_g);
795                    }
796                    info!(key=%self.key, last = self.last_response.is_none(), "entered confirmed state, waiting for ACK");
797                    match self.last_response {
798                        Some(ref resp) => {
799                            let dialog_id = DialogId::try_from(resp)?;
800                            self.endpoint_inner
801                                .waiting_ack
802                                .write()
803                                .as_mut()
804                                .map(|wa| wa.insert(dialog_id, self.key.clone()))
805                                .ok();
806                        }
807                        _ => {}
808                    }
809                    // start Timer K, wait for ACK
810                    let timer_k = self.endpoint_inner.timers.timeout(
811                        self.endpoint_inner.option.t4,
812                        TransactionTimer::TimerK(self.key.clone()),
813                    );
814                    self.timer_k.replace(timer_k);
815                }
816                // start Timer D
817                let timer_d = self.endpoint_inner.timers.timeout(
818                    self.endpoint_inner.option.t1x64,
819                    TransactionTimer::TimerD(self.key.clone()),
820                );
821                self.timer_d.replace(timer_d);
822            }
823            TransactionState::Confirmed => {
824                self.cleanup_timer();
825                let timer_k = self.endpoint_inner.timers.timeout(
826                    self.endpoint_inner.option.t4,
827                    TransactionTimer::TimerK(self.key.clone()),
828                );
829                self.timer_k.replace(timer_k);
830            }
831            TransactionState::Terminated => {
832                self.cleanup();
833                self.tu_sender.send(TransactionEvent::Terminate).ok(); // tell TU to terminate
834            }
835        }
836        debug!(
837            key = %self.key,
838            "transition: {:?} -> {:?}", self.state, state
839        );
840        self.state = state;
841        Ok(self.state.clone())
842    }
843
844    fn cleanup_timer(&mut self) {
845        self.timer_a
846            .take()
847            .map(|id| self.endpoint_inner.timers.cancel(id));
848        self.timer_b
849            .take()
850            .map(|id| self.endpoint_inner.timers.cancel(id));
851        self.timer_c
852            .take()
853            .map(|id| self.endpoint_inner.timers.cancel(id));
854        self.timer_d
855            .take()
856            .map(|id| self.endpoint_inner.timers.cancel(id));
857        self.timer_k
858            .take()
859            .map(|id| self.endpoint_inner.timers.cancel(id));
860        self.timer_g
861            .take()
862            .map(|id| self.endpoint_inner.timers.cancel(id));
863    }
864
865    fn cleanup(&mut self) {
866        if self.is_cleaned_up {
867            return;
868        }
869        self.is_cleaned_up = true;
870        self.cleanup_timer();
871
872        match self.last_response {
873            Some(ref resp) => match DialogId::try_from(resp) {
874                Ok(dialog_id) => self
875                    .endpoint_inner
876                    .waiting_ack
877                    .write()
878                    .as_mut()
879                    .map(|wa| wa.remove(&dialog_id))
880                    .ok(),
881                Err(_) => None,
882            },
883            _ => None,
884        };
885
886        let last_message = {
887            match self.transaction_type {
888                TransactionType::ClientInvite => {
889                    //
890                    // For client invite, make a placeholder ACK if in proceeding or trying state
891                    if matches!(
892                        self.state,
893                        TransactionState::Proceeding | TransactionState::Trying
894                    ) {
895                        if self.last_ack.is_none() {
896                            if let Some(ref resp) = self.last_response {
897                                if let Ok(ack) = self
898                                    .endpoint_inner
899                                    .make_ack(self.original.uri.clone(), resp)
900                                {
901                                    self.last_ack.replace(ack);
902                                }
903                            }
904                        }
905                    }
906                    self.last_ack.take().map(SipMessage::Request)
907                }
908                TransactionType::ServerNonInvite => {
909                    self.last_response.take().map(SipMessage::Response)
910                }
911                _ => None,
912            }
913        };
914        self.endpoint_inner
915            .detach_transaction(&self.key, last_message);
916    }
917}
918
919impl Drop for Transaction {
920    fn drop(&mut self) {
921        self.cleanup();
922        trace!(key=%self.key, state=%self.state, "transaction dropped");
923    }
924}
ftth_rsipstack/transaction/transaction.rs

ftth_rsipstack/transaction/
transaction.rs
