o_sfu_protocol/core/
outbound_batch.rs

1use std::mem::take;
2
3use super::{
4    BATCH_FLUSH_DELAY_MS, BATCH_FLUSH_TIMER_ID, Command, Commands, MAX_OUTBOUND_BATCH_LEN,
5};
6use crate::signaling::{Envelope, EnvelopeBatch};
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub(super) enum FlushMode {
10    Immediate,
11    Batched,
12}
13
14/// buffers control-plane envelopes behind one host flush timer
15///
16/// the first batched envelope schedules [`BATCH_FLUSH_TIMER_ID`]
17/// later batched envelopes join the same frame until the timer fires or the batch
18/// reaches [`MAX_OUTBOUND_BATCH_LEN`]
19#[derive(Debug, Clone, Default, PartialEq, Eq)]
20pub(super) struct OutboundBatcher {
21    pending_batch: EnvelopeBatch,
22    flush_scheduled: bool,
23}
24
25impl OutboundBatcher {
26    pub(super) fn new() -> Self {
27        Self::default()
28    }
29
30    pub(super) fn enqueue(&mut self, envelope: Envelope, mode: FlushMode) -> Commands {
31        match mode {
32            FlushMode::Immediate => {
33                self.pending_batch.push(envelope);
34                self.flush(true)
35            }
36            FlushMode::Batched => {
37                self.pending_batch.push(envelope);
38                if self.pending_batch.len() >= MAX_OUTBOUND_BATCH_LEN {
39                    self.flush(true)
40                } else if self.flush_scheduled {
41                    Vec::new()
42                } else {
43                    self.flush_scheduled = true;
44                    vec![Command::ScheduleTimer {
45                        id: BATCH_FLUSH_TIMER_ID,
46                        ms: BATCH_FLUSH_DELAY_MS,
47                    }]
48                }
49            }
50        }
51    }
52
53    pub(super) fn extend(&mut self, envelopes: EnvelopeBatch) {
54        self.pending_batch.extend(envelopes);
55    }
56
57    pub(super) fn flush(&mut self, cancel_timer: bool) -> Commands {
58        if self.pending_batch.is_empty() {
59            self.flush_scheduled = false;
60            return Vec::new();
61        }
62        let batch = take(&mut self.pending_batch);
63        let Ok(frame) = serde_json::to_string(&batch) else {
64            self.flush_scheduled = false;
65            return Vec::new();
66        };
67        let had_timer = self.flush_scheduled;
68        self.flush_scheduled = false;
69        let mut commands = Vec::new();
70        if cancel_timer && had_timer {
71            commands.push(Command::CancelTimer {
72                id: BATCH_FLUSH_TIMER_ID,
73            });
74        }
75        commands.push(Command::SendWebSocket { frame });
76        commands
77    }
78
79    pub(super) fn clear(&mut self) {
80        self.pending_batch.clear();
81        self.flush_scheduled = false;
82    }
83
84    pub(super) fn discard_pending(&mut self) -> Commands {
85        let commands = if self.flush_scheduled {
86            vec![Command::CancelTimer {
87                id: BATCH_FLUSH_TIMER_ID,
88            }]
89        } else {
90            Vec::new()
91        };
92        self.clear();
93        commands
94    }
95}
o_sfu_protocol/core/outbound_batch.rs

o_sfu_protocol/core/
outbound_batch.rs
