servo_base/generic_channel/
buffered.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
4
5use std::cell::RefCell;
6use std::mem;
7use std::panic::Location;
8
9use malloc_size_of_derive::MallocSizeOf;
10use serde::Serialize;
11
12use super::{GenericSender, SendResult};
13
14/// A buffered sender that collects individual messages (`U`) and sends them
15/// as a single batched message (`T`) via a user-provided packing closure.
16///
17/// The buffer is flushed automatically when it reaches `max_buffer` items,
18/// or explicitly via [`flush`](GenericBufferedSender::flush).
19/// [`send_immediate`](GenericBufferedSender::send_immediate)
20/// combines the current buffer contents with the new message into a single
21/// packed message, ensuring ordering without an extra flush step.
22#[derive(MallocSizeOf)]
23pub struct GenericBufferedSender<T, U>
24where
25    T: Serialize,
26{
27    sender: GenericSender<T>,
28    buffer: RefCell<Vec<U>>,
29    #[ignore_malloc_size_of = "dyn are difficult to measure"]
30    buffering: Box<dyn Fn(Vec<U>) -> T>,
31    max_buffer: usize,
32}
33
34impl<T: Serialize, U> GenericBufferedSender<T, U> {
35    /// Create a new buffered sender.
36    ///
37    /// * `sender` — the underlying `GenericSender<T>` that delivers packed messages.
38    /// * `buffering` — closure that packs a `Vec<U>` into a single `T`.
39    /// * `max_buffer` — automatic flush is triggered when the buffer reaches this size.
40    pub fn new(
41        sender: GenericSender<T>,
42        buffering: Box<dyn Fn(Vec<U>) -> T>,
43        max_buffer: usize,
44    ) -> Self {
45        Self {
46            sender,
47            buffer: RefCell::new(Vec::new()),
48            buffering,
49            max_buffer,
50        }
51    }
52
53    /// Returns `true` if the buffer contains no pending messages.
54    pub fn is_empty(&self) -> bool {
55        self.buffer.borrow().is_empty()
56    }
57
58    /// Returns the number of pending messages in the buffer.
59    pub fn len(&self) -> usize {
60        self.buffer.borrow().len()
61    }
62
63    /// Buffer a message for later batched delivery.
64    ///
65    /// If the buffer reaches `max_buffer` items an automatic flush is triggered.
66    pub fn send(&self, msg: U) -> SendResult {
67        if self.buffer.borrow().len() + 1 >= self.max_buffer {
68            self.send_immediate(msg)
69        } else {
70            self.buffer.borrow_mut().push(msg);
71            Ok(())
72        }
73    }
74
75    #[inline]
76    #[track_caller]
77    /// Buffer a message for later batched delivery.
78    ///
79    /// If the buffer reaches `max_buffer` items an automatic flush is triggered.
80    /// Errors on automatic flush are logged.
81    pub fn send_or_warn(&self, msg: U) {
82        if let Err(error) = self.send(msg) {
83            let location = Location::caller();
84            log::warn!("Failed to send buffered messages due to `{error}` at {location:?}");
85        }
86    }
87
88    /// Deliver a message immediately, combining it with any
89    /// buffered messages into a single packed `T`.
90    pub fn send_immediate(&self, msg: U) -> SendResult {
91        let mut buffer = self.buffer.borrow_mut();
92        buffer.push(msg);
93        let msgs = mem::take(&mut *buffer);
94        drop(buffer);
95        let packed = (self.buffering)(msgs);
96        self.sender.send(packed)
97    }
98
99    #[inline]
100    #[track_caller]
101    /// Deliver a message immediately, combining it with any
102    /// buffered messages into a single packed `T`.
103    ///
104    /// Errors are logged.
105    pub fn send_immediate_or_warn(&self, msg: U) {
106        if let Err(error) = self.send_immediate(msg) {
107            let location = Location::caller();
108            log::warn!(
109                "Failed to send (immediate) buffered messages due to `{error}` at {location:?}"
110            );
111        }
112    }
113
114    /// Flush all buffered messages by packing them into a single `T` and
115    /// sending it.
116    pub fn flush(&self) -> SendResult {
117        let mut buffer = self.buffer.borrow_mut();
118        if buffer.is_empty() {
119            return Ok(());
120        }
121        let msgs = mem::take(&mut *buffer);
122        drop(buffer);
123        let packed = (self.buffering)(msgs);
124        self.sender.send(packed)
125    }
126
127    #[inline]
128    #[track_caller]
129    /// Flush all buffered messages by packing them into a single `T` and sending it.
130    /// Errors are logged
131    pub fn flush_or_warn(&self) {
132        if let Err(error) = self.flush() {
133            let location = Location::caller();
134            log::warn!("Failed to flush buffered messages due to `{error}` at {location:?}");
135        }
136    }
137
138    /// Discard all buffered messages without sending them.
139    pub fn discard(&self) {
140        self.buffer.borrow_mut().clear();
141    }
142}
143
144impl<T: Serialize, U> Drop for GenericBufferedSender<T, U> {
145    fn drop(&mut self) {
146        // Best-effort flush on drop. Ignore send failures.
147        if !self.buffer.borrow().is_empty() {
148            let msgs = mem::take(&mut *self.buffer.borrow_mut());
149            let packed = (self.buffering)(msgs);
150            let _ = self.sender.send(packed);
151        }
152    }
153}
servo_base/generic_channel/buffered.rs

servo_base/generic_channel/
buffered.rs
