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 try_begin(&mut self, kind: PendingRequestKind) -> Option<PendingRequestStart> {
48        if self
49            .pending_requests
50            .iter()
51            .any(|pending_request| pending_request.kind == kind)
52        {
53            return None;
54        }
55        let request_id = self.allocate_request_id();
56        let timeout_timer_id = self.allocate_timeout_timer_id();
57        self.pending_requests.push(PendingRequestState {
58            request_id: request_id.clone(),
59            kind,
60            timeout_timer_id,
61        });
62        Some(PendingRequestStart {
63            request_id,
64            timeout_timer_id,
65        })
66    }
67
68    pub(super) fn resolve_response(
69        &mut self,
70        response_to: &RequestId,
71        expected_kind: PendingRequestKind,
72        ok: bool,
73    ) -> Commands {
74        let Some(index) = self.pending_requests.iter().position(|pending_request| {
75            pending_request.request_id == *response_to && pending_request.kind == expected_kind
76        }) else {
77            return Vec::new();
78        };
79        vec![complete_pending_request(
80            self.pending_requests.remove(index),
81            ok,
82        )]
83    }
84
85    pub(super) fn resolve_timeout(&mut self, timeout_id: RequestTimeoutId) -> Option<Commands> {
86        let index = self
87            .pending_requests
88            .iter()
89            .position(|pending_request| pending_request.timeout_timer_id == timeout_id)?;
90        Some(vec![complete_pending_request(
91            self.pending_requests.remove(index),
92            false,
93        )])
94    }
95
96    pub(super) fn clear(&mut self) {
97        self.pending_requests.clear();
98    }
99
100    /// Completes pending requests in begin order.
101    pub(super) fn fail_all(&mut self) -> Commands {
102        mem::take(&mut self.pending_requests)
103            .into_iter()
104            .map(|pending_request| complete_pending_request(pending_request, false))
105            .collect()
106    }
107
108    fn allocate_request_id(&mut self) -> RequestId {
109        let request_id = RequestId::new(self.next_request_counter.to_string());
110        self.next_request_counter = self.next_request_counter.saturating_add(1);
111        request_id
112    }
113
114    fn allocate_timeout_timer_id(&mut self) -> RequestTimeoutId {
115        let timer_id = self.next_timeout_timer_id;
116        self.next_timeout_timer_id = timer_id.next();
117        timer_id
118    }
119}
120
121fn complete_pending_request(pending_request: PendingRequestState, ok: bool) -> Command {
122    Command::CompletePendingRequest {
123        request_id: pending_request.request_id,
124        timeout_timer_id: pending_request.timeout_timer_id.raw(),
125        ok,
126    }
127}
o_sfu_protocol/core/request_tracker.rs

o_sfu_protocol/core/
request_tracker.rs
