rx_rust/subject/
behavior_subject.rs

1//! A subject that keeps a current value and hands it to every new subscriber.
2//!
3//! The current value lives in the [`SerializedMulticast`]'s resources, so replacing it and
4//! forwarding it are one atomic step, and so are reading it and joining the multicast: a
5//! subscriber can neither miss a value emitted while it was subscribing nor observe one twice.
6
7use super::Subject;
8use crate::delegate_disposal;
9use crate::disposable::DisposableExt;
10use crate::disposable::option_disposal::OptionDisposal;
11use crate::observable::Subscription;
12use crate::utils::pending_events::EventBatch;
13use crate::utils::serialized_delivery::UpdateOutcome;
14use crate::utils::serialized_multicast::{Admission, MulticastDisposal, SerializedMulticast};
15use crate::utils::types::MaybeSend;
16use crate::{
17    observable::Observable,
18    observer::{Flow, Observer, Termination},
19};
20use educe::Educe;
21
22/// Keeps the latest value and emits it immediately to new subscribers.
23#[derive(Educe)]
24#[educe(Debug, Clone)]
25pub struct BehaviorSubject<'or, T, E>(SerializedMulticast<'or, T, E, T>);
26
27impl<T, E> BehaviorSubject<'_, T, E> {
28    pub fn new(value: T) -> Self {
29        Self(SerializedMulticast::idle(value))
30    }
31}
32
33impl<T, E> BehaviorSubject<'_, T, E>
34where
35    T: Clone,
36    E: Clone,
37{
38    /// The current value.
39    ///
40    /// # Panics
41    ///
42    /// Panics once an observer's callback has panicked, which takes the subject's whole state with
43    /// it and leaves nothing to return.
44    pub fn value(&self) -> T {
45        self.0
46            .read(|value, _| value.clone())
47            .expect("the subject is dead because an observer panicked")
48    }
49}
50
51delegate_disposal!(
52    Disposal<'or, T, E>,
53    OptionDisposal<MulticastDisposal<'or, T, E, T>>,
54);
55
56impl<'or, T, E> Observable<'or, T, E> for BehaviorSubject<'or, T, E>
57where
58    T: Clone,
59    E: Clone,
60{
61    type D = Disposal<'or, T, E>;
62
63    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
64        match self
65            .0
66            .subscribe_with(observer, |value, terminated| match terminated {
67                // The current value is snapshotted under the very lock that queues the subscription,
68                // so it is followed by exactly the values emitted after it.
69                None => Admission::Join(vec![value.clone()]),
70                Some(termination) => Admission::Terminated(Vec::new(), termination.clone()),
71            }) {
72            Some(disposal) => OptionDisposal::some(disposal),
73            None => OptionDisposal::none(),
74        }
75        .into_subscription()
76    }
77}
78
79impl<T, E> Observer<T, E> for BehaviorSubject<'_, T, E>
80where
81    T: Clone,
82    E: Clone,
83{
84    fn on_next(&mut self, value: T) -> Flow {
85        self.0
86            .update(|current, terminated| {
87                if terminated.is_some() {
88                    return UpdateOutcome::new(Flow::Stop)
89                        .with_drop_outside(Some(value))
90                        .without_events();
91                }
92                // Replacing the current value and forwarding it are one step, so the value a
93                // subscriber is given always matches the values it then receives. The replaced
94                // value is dropped outside the lock: dropping it can run arbitrary code.
95                let previous = std::mem::replace(current, value.clone());
96                UpdateOutcome::new(Flow::Continue)
97                    .with_drop_outside(Some(previous))
98                    .with_next_event(value)
99            })
100            .unwrap_or(Flow::Stop)
101    }
102
103    fn on_termination(self, termination: Termination<E>) {
104        let _ = self.0.send(EventBatch::Termination(termination));
105    }
106}
107
108impl<'or, T, E> Subject<'or, T, E> for BehaviorSubject<'or, T, E>
109where
110    T: Clone,
111    E: Clone,
112{
113    fn terminated(&self) -> Option<Termination<E>>
114    where
115        E: Clone,
116    {
117        self.0.terminated()
118    }
119}
rx_rust/subject/behavior_subject.rs

rx_rust/subject/
behavior_subject.rs
