rx_rust/subject/
publish_subject.rs

1//! A multicast subject that forwards what it receives to every observer.
2//!
3//! It is [`SerializedMulticast`] with nothing of its own to guard: every method here is one call
4//! into it. The subject is terminated as soon as [`Observer::on_termination`] is *called*, not when
5//! the termination reaches the observers — see the multicast's module documentation for why, and
6//! for everything else that governs the ordering here.
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_multicast::{MulticastDisposal, SerializedMulticast};
15use crate::utils::types::MaybeSend;
16use crate::{
17    observable::Observable,
18    observer::{Flow, Observer, Termination},
19};
20use educe::Educe;
21
22/// Basic multicast subject that forwards events to all observers.
23///
24/// Observers are notified in subscription order, and an observer that unsubscribes does not
25/// disturb the order of the others.
26#[derive(Educe)]
27#[educe(Debug, Clone)]
28pub struct PublishSubject<'or, T, E>(SerializedMulticast<'or, T, E>);
29
30impl<T, E> PublishSubject<'_, T, E> {
31    pub fn new() -> Self {
32        Self(SerializedMulticast::idle(()))
33    }
34}
35
36impl<T, E> Default for PublishSubject<'_, T, E> {
37    fn default() -> Self {
38        Self::new()
39    }
40}
41
42delegate_disposal!(
43    Disposal<'or, T, E>,
44    OptionDisposal<MulticastDisposal<'or, T, E, ()>>,
45);
46
47impl<'or, T, E> Observable<'or, T, E> for PublishSubject<'or, T, E>
48where
49    T: Clone,
50    E: Clone,
51{
52    type D = Disposal<'or, T, E>;
53
54    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
55        match self.0.subscribe(observer) {
56            Some(disposal) => OptionDisposal::some(disposal),
57            None => OptionDisposal::none(),
58        }
59        .into_subscription()
60    }
61}
62
63impl<T, E> Observer<T, E> for PublishSubject<'_, T, E>
64where
65    T: Clone,
66    E: Clone,
67{
68    fn on_next(&mut self, value: T) -> Flow {
69        // A value that arrives after the termination is dropped outside the lock.
70        self.0.send(EventBatch::Next(value))
71    }
72
73    fn on_termination(self, termination: Termination<E>) {
74        // Only the first termination is ever queued, and nothing joins the subject after it.
75        let _ = self.0.send(EventBatch::Termination(termination));
76    }
77}
78
79impl<'or, T, E> Subject<'or, T, E> for PublishSubject<'or, T, E>
80where
81    T: Clone,
82    E: Clone,
83{
84    fn terminated(&self) -> Option<Termination<E>> {
85        self.0.terminated()
86    }
87}
rx_rust/subject/publish_subject.rs

rx_rust/subject/
publish_subject.rs
