1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
use {
crate::streamer::ChannelSend,
crossbeam_channel::{Receiver, SendError, Sender, TryRecvError, TrySendError, bounded},
};
/// A sender implementation that evicts the oldest message when the channel is full.
#[derive(Clone)]
pub struct EvictingSender<T> {
sender: Sender<T>,
receiver: Receiver<T>,
}
impl<T> EvictingSender<T> {
/// Create a new evicting sender with provided sender, receiver.
#[inline]
pub fn new(sender: Sender<T>, receiver: Receiver<T>) -> Self {
Self { sender, receiver }
}
/// Create a new `EvictingSender` with a bounded channel of the specified capacity.
#[inline]
pub fn new_bounded(capacity: usize) -> (Self, Receiver<T>) {
let (sender, receiver) = bounded(capacity);
(Self::new(sender, receiver.clone()), receiver)
}
}
impl<T> ChannelSend<T> for EvictingSender<T>
where
T: Send + 'static,
{
#[inline]
fn send(&self, msg: T) -> std::result::Result<(), SendError<T>> {
self.sender.send(msg)
}
fn try_send(&self, msg: T) -> std::result::Result<(), TrySendError<T>> {
let Err(e) = self.sender.try_send(msg) else {
return Ok(());
};
match e {
// Prefer newer messages over older messages.
TrySendError::Full(msg) => match self.receiver.try_recv() {
Ok(older) => {
// Attempt to requeue the newer message.
// NB: if multiple senders are used, and another sender is faster than us to send() after we've popped `older`,
// our try_send() will fail with Full(msg), in which case we drop the new message.
self.sender.try_send(msg)?;
// Propagate the error _with the older message_.
Err(TrySendError::Full(older))
}
// Unlikely race condition -- it was just indicated that the channel is full.
// Attempt to requeue the message.
Err(TryRecvError::Empty) => self.sender.try_send(msg),
// Unreachable in practice since we maintain a reference to both the sender and receiver.
Err(TryRecvError::Disconnected) => unreachable!(),
},
// Unreachable in practice since we maintain a reference to both the sender and receiver.
TrySendError::Disconnected(_) => unreachable!(),
}
}
#[inline]
fn is_empty(&self) -> bool {
self.receiver.is_empty()
}
#[inline]
fn len(&self) -> usize {
self.receiver.len()
}
}