rx_rust/subject/
replay_subject.rs

1//! A subject that buffers what it emits and replays that buffer to every new subscriber.
2//!
3//! The buffer lives in the [`SerializedMulticast`]'s resources, so buffering a value and
4//! forwarding it are one atomic step, and so are snapshotting the buffer and joining the
5//! multicast: a subscriber can neither miss a value emitted while it was subscribing nor observe
6//! one twice.
7
8use super::Subject;
9use crate::delegate_disposal;
10use crate::disposable::DisposableExt;
11use crate::disposable::option_disposal::OptionDisposal;
12use crate::observable::Subscription;
13use crate::utils::pending_events::EventBatch;
14use crate::utils::serialized_delivery::UpdateOutcome;
15use crate::utils::serialized_multicast::{Admission, MulticastDisposal, SerializedMulticast};
16use crate::utils::types::MaybeSend;
17use crate::{
18    observable::Observable,
19    observer::{Flow, Observer, Termination},
20};
21use educe::Educe;
22use std::collections::VecDeque;
23
24/// Buffers emissions and replays them to late subscribers.
25///
26/// A subscriber that arrives after the subject terminated observes the buffered values followed by
27/// the termination, whether the subject completed or errored.
28#[derive(Educe)]
29#[educe(Debug, Clone)]
30pub struct ReplaySubject<'or, T, E>(SerializedMulticast<'or, T, E, Buffer<T>>);
31
32/// The replayed values, and how many of them are kept.
33#[derive(Educe)]
34#[educe(Debug)]
35struct Buffer<T> {
36    values: VecDeque<T>,
37    /// The number of values kept, or [`None`] to keep every one of them.
38    size: Option<usize>,
39}
40
41impl<T, E> ReplaySubject<'_, T, E> {
42    pub fn new(buffer_size: Option<usize>) -> Self {
43        let values = match buffer_size {
44            Some(size) => VecDeque::with_capacity(size),
45            None => VecDeque::new(),
46        };
47        Self(SerializedMulticast::idle(Buffer {
48            values,
49            size: buffer_size,
50        }))
51    }
52}
53
54delegate_disposal!(
55    Disposal<'or, T, E>,
56    OptionDisposal<MulticastDisposal<'or, T, E, Buffer<T>>>,
57);
58
59impl<'or, T, E> Observable<'or, T, E> for ReplaySubject<'or, T, E>
60where
61    T: Clone,
62    E: Clone,
63{
64    type D = Disposal<'or, T, E>;
65
66    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
67        match self.0.subscribe_with(observer, |buffer, terminated| {
68            // The buffer is the history of the subject, so it is replayed whichever way the
69            // subject terminated: an error does not erase what was emitted before it.
70            let values = buffer.values.iter().cloned().collect();
71            match terminated {
72                None => Admission::Join(values),
73                Some(termination) => Admission::Terminated(values, termination.clone()),
74            }
75        }) {
76            Some(disposal) => OptionDisposal::some(disposal),
77            None => OptionDisposal::none(),
78        }
79        .into_subscription()
80    }
81}
82
83impl<T, E> Observer<T, E> for ReplaySubject<'_, T, E>
84where
85    T: Clone,
86    E: Clone,
87{
88    fn on_next(&mut self, value: T) -> Flow {
89        self.0
90            .update(|buffer, terminated| {
91                if terminated.is_some() {
92                    return UpdateOutcome::new(Flow::Stop)
93                        .with_drop_outside(Some(value))
94                        .without_events();
95                }
96                // Buffering the value and forwarding it are one step, so the buffer a subscriber
97                // is replayed always matches the values it then receives. The evicted value is
98                // dropped outside the lock: dropping it can run arbitrary code.
99                let evicted = buffer.push(value.clone());
100                UpdateOutcome::new(Flow::Continue)
101                    .with_drop_outside(evicted)
102                    .with_next_event(value)
103            })
104            .unwrap_or(Flow::Stop)
105    }
106
107    fn on_termination(self, termination: Termination<E>) {
108        let _ = self.0.send(EventBatch::Termination(termination));
109    }
110}
111
112impl<T> Buffer<T> {
113    /// Buffers `value`, returning the value it evicted, if any.
114    ///
115    /// A buffer of size zero keeps nothing: the value is only forwarded.
116    fn push(&mut self, value: T) -> Option<T> {
117        let Some(size) = self.size else {
118            self.values.push_back(value);
119            return None;
120        };
121        if self.values.len() < size {
122            self.values.push_back(value);
123            return None;
124        }
125        let evicted = self.values.pop_front();
126        if evicted.is_some() {
127            self.values.push_back(value);
128        }
129        evicted
130    }
131}
132
133impl<'or, T, E> Subject<'or, T, E> for ReplaySubject<'or, T, E>
134where
135    T: Clone,
136    E: Clone,
137{
138    fn terminated(&self) -> Option<Termination<E>>
139    where
140        E: Clone,
141    {
142        self.0.terminated()
143    }
144}
rx_rust/subject/replay_subject.rs

rx_rust/subject/
replay_subject.rs
