rx_rust/subject/
async_subject.rs

1//! A subject that remembers only its last value and replays it on completion.
2//!
3//! The last value lives in the [`SerializedMulticast`]'s resources, so reading it, reading the
4//! termination and emitting are one atomic step: the value that is replayed on completion is
5//! exactly the one every later subscriber observes, and a value that arrives once the completion
6//! was queued is dropped rather than replacing it.
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;
22
23/// Remembers only the last emission and replays it on completion.
24///
25/// Unlike [`PublishSubject`](super::publish_subject::PublishSubject), an observer that subscribes after the subject completed still
26/// observes that last value, followed by the completion. An error replays nothing.
27#[derive(Educe)]
28#[educe(Debug, Clone)]
29pub struct AsyncSubject<'or, T, E>(SerializedMulticast<'or, T, E, Option<T>>);
30
31impl<T, E> AsyncSubject<'_, T, E> {
32    pub fn new() -> Self {
33        Self(SerializedMulticast::idle(None))
34    }
35}
36
37impl<T, E> Default for AsyncSubject<'_, T, E> {
38    fn default() -> Self {
39        Self::new()
40    }
41}
42
43delegate_disposal!(
44    Disposal<'or, T, E>,
45    OptionDisposal<MulticastDisposal<'or, T, E, Option<T>>>,
46);
47
48impl<'or, T, E> Observable<'or, T, E> for AsyncSubject<'or, T, E>
49where
50    T: Clone,
51    E: Clone,
52{
53    type D = Disposal<'or, T, E>;
54
55    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
56        match self
57            .0
58            .subscribe_with(observer, |last, terminated| match terminated {
59                // Reading the last value and the termination is one step, so a subscriber can never
60                // observe one without the other.
61                None => Admission::Join(Vec::new()),
62                Some(Termination::Completed) => Admission::Terminated(
63                    last.clone().into_iter().collect(),
64                    Termination::Completed,
65                ),
66                Some(termination) => Admission::Terminated(Vec::new(), termination.clone()),
67            }) {
68            Some(disposal) => OptionDisposal::some(disposal),
69            None => OptionDisposal::none(),
70        }
71        .into_subscription()
72    }
73}
74
75impl<T, E> Observer<T, E> for AsyncSubject<'_, T, E>
76where
77    T: Clone,
78    E: Clone,
79{
80    fn on_next(&mut self, value: T) -> Flow {
81        self.0
82            .update(|last, terminated| {
83                if terminated.is_some() {
84                    // Replacing the value now would change what the later subscribers observe,
85                    // after the completion already replayed the previous one.
86                    return UpdateOutcome::new(Flow::Stop)
87                        .with_drop_outside(Some(value))
88                        .without_events();
89                }
90                // The replaced value is dropped outside the lock: dropping it can run arbitrary
91                // code.
92                let previous = last.replace(value);
93                UpdateOutcome::new(Flow::Continue)
94                    .with_drop_outside(previous)
95                    .without_events()
96            })
97            .unwrap_or(Flow::Stop)
98    }
99
100    fn on_termination(self, termination: Termination<E>) {
101        let _ = self.0.update(|last, terminated| {
102            if terminated.is_some() {
103                return UpdateOutcome::empty()
104                    .with_drop_outside(Some(termination))
105                    .without_events();
106            }
107            // The last value and the completion are queued together, and the subject is terminated
108            // by that very step: nothing can slip between them.
109            let events = if matches!(termination, Termination::Completed) {
110                match last.clone() {
111                    Some(value) => EventBatch::NextAndTermination(value, termination),
112                    None => EventBatch::Termination(termination),
113                }
114            } else {
115                // An error replays nothing, so the last value is not even cloned.
116                EventBatch::Termination(termination)
117            };
118            UpdateOutcome::empty()
119                .without_drop_outside()
120                .with_events(events)
121        });
122    }
123}
124
125impl<'or, T, E> Subject<'or, T, E> for AsyncSubject<'or, T, E>
126where
127    T: Clone,
128    E: Clone,
129{
130    fn terminated(&self) -> Option<Termination<E>>
131    where
132        E: Clone,
133    {
134        self.0.terminated()
135    }
136}
rx_rust/subject/async_subject.rs

rx_rust/subject/
async_subject.rs
