o_sfu_protocol/core/
request_tracker.rs

1//! tracks in-flight recording requests and their timeout timers
2
3use std::mem;
4
5use super::{
6    Command, Commands, PendingRequestKind,
7    timers::{REQUEST_TIMEOUT_TIMER_BASE, RequestTimeoutId},
8};
9use crate::signaling::RequestId;
10
11#[derive(Debug, Clone, PartialEq, Eq)]
12struct PendingRequestState {
13    request_id: RequestId,
14    kind: PendingRequestKind,
15    timeout_timer_id: RequestTimeoutId,
16}
17
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub(super) struct PendingRequestStart {
20    pub(super) request_id: RequestId,
21    pub(super) timeout_timer_id: RequestTimeoutId,
22}
23
24/// Owns recording request identities, timeout identities and resolve-once state.
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub(super) struct RequestTracker {
27    next_request_counter: u64,
28    next_timeout_timer_id: RequestTimeoutId,
29    pending_requests: Vec<PendingRequestState>,
30}
31
32impl Default for RequestTracker {
33    fn default() -> Self {
34        Self::new()
35    }
36}
37
38impl RequestTracker {
39    pub(super) const fn new() -> Self {
40        Self {
41            next_request_counter: 0,
42            next_timeout_timer_id: REQUEST_TIMEOUT_TIMER_BASE,
43            pending_requests: Vec::new(),
44        }
45    }
46
47    pub(super) fn begin(&mut self, kind: PendingRequestKind) -> PendingRequestStart {
48        let request_id = self.allocate_request_id();
49        let timeout_timer_id = self.allocate_timeout_timer_id();
50        self.pending_requests.push(PendingRequestState {
51            request_id: request_id.clone(),
52            kind,
53            timeout_timer_id,
54        });
55        PendingRequestStart {
56            request_id,
57            timeout_timer_id,
58        }
59    }
60
61    pub(super) fn resolve_response(
62        &mut self,
63        response_to: &RequestId,
64        expected_kind: PendingRequestKind,
65        ok: bool,
66    ) -> Commands {
67        let Some(index) = self.pending_requests.iter().position(|pending_request| {
68            pending_request.request_id == *response_to && pending_request.kind == expected_kind
69        }) else {
70            return Vec::new();
71        };
72        // Remove before settling so duplicate responses and stale timers cannot settle twice.
73        vec![complete_pending_request(
74            self.pending_requests.remove(index),
75            ok,
76        )]
77    }
78
79    pub(super) fn resolve_timeout(&mut self, timeout_id: RequestTimeoutId) -> Option<Commands> {
80        let index = self
81            .pending_requests
82            .iter()
83            .position(|pending_request| pending_request.timeout_timer_id == timeout_id)?;
84        Some(vec![reject_pending_request(
85            self.pending_requests.remove(index),
86            "bus request timed out",
87        )])
88    }
89
90    pub(super) fn clear(&mut self) {
91        self.pending_requests.clear();
92    }
93
94    /// Fails pending requests in begin order.
95    pub(super) fn fail_all(&mut self) -> Commands {
96        mem::take(&mut self.pending_requests)
97            .into_iter()
98            .map(|pending_request| reject_pending_request(pending_request, "bus closed"))
99            .collect()
100    }
101
102    fn allocate_request_id(&mut self) -> RequestId {
103        let request_id = RequestId::new(self.next_request_counter.to_string());
104        self.next_request_counter = self.next_request_counter.saturating_add(1);
105        request_id
106    }
107
108    fn allocate_timeout_timer_id(&mut self) -> RequestTimeoutId {
109        let timer_id = self.next_timeout_timer_id;
110        self.next_timeout_timer_id = timer_id.next();
111        timer_id
112    }
113}
114
115fn complete_pending_request(pending_request: PendingRequestState, ok: bool) -> Command {
116    Command::CompletePendingRequest {
117        request_id: pending_request.request_id,
118        timeout_timer_id: pending_request.timeout_timer_id.raw(),
119        ok,
120    }
121}
122
123fn reject_pending_request(pending_request: PendingRequestState, error: &str) -> Command {
124    Command::RejectPendingRequest {
125        request_id: pending_request.request_id,
126        timeout_timer_id: pending_request.timeout_timer_id.raw(),
127        error: error.to_owned(),
128    }
129}
o_sfu_protocol/core/request_tracker.rs

o_sfu_protocol/core/
request_tracker.rs
