rx_rust/operators/connectable/
connectable_controller.rs

1use super::ref_count::RefCount;
2use crate::disposable::Disposable;
3use crate::observable::{Observable, Subscription};
4use crate::observer::Observer;
5use crate::subject::subject_observable::SubjectObservable;
6use crate::utils::types::MaybeSend;
7use educe::Educe;
8
9/// Marker for a connectable controller that is not connected to its source.
10#[derive(Debug, Clone, Copy, Default)]
11pub struct Disconnected;
12
13/// State carried by a connected controller. Dropping it disconnects the source.
14#[derive(Educe)]
15#[educe(Debug)]
16pub struct Connected<D: Disposable>(Subscription<D>);
17
18/// Multicasts a source `Observable` through a `Subject`, but waits until its [`connect`](ConnectableController::connect)
19/// method is called before subscribing to the source and emitting items to its observers.
20/// Subscribe to the multicast output through [`observable`](ConnectableController::observable).
21/// See <https://reactivex.io/documentation/operators/connect.html>
22///
23/// # Examples
24/// ```rust
25/// use rx_rust::{
26///     observable::ObservableExt,
27///     observer::Termination,
28///     operators::{
29///         connectable::connectable_controller::ConnectableController,
30///         creating::from_iter::FromIter,
31///     },
32///     subject::publish_subject::PublishSubject,
33/// };
34///
35/// use std::{convert::Infallible, sync::{Arc, Mutex}};
36///
37/// let values_1 = Arc::new(Mutex::new(Vec::new()));
38/// let values_2 = Arc::new(Mutex::new(Vec::new()));
39/// let terminations = Arc::new(Mutex::new(Vec::new()));
40///
41/// let subject: PublishSubject<'_, i32, Infallible> = PublishSubject::default();
42/// let controller = ConnectableController::new(FromIter::new(vec![1, 2]), subject);
43/// let observable = controller.observable();
44/// let values_1_observer = Arc::clone(&values_1);
45/// let values_2_observer = Arc::clone(&values_2);
46/// let terminations_observer = Arc::clone(&terminations);
47///
48/// let subscription_1 = observable.clone().subscribe_with_callback(
49///     move |value| values_1_observer.lock().unwrap().push(value),
50///     |_| {},
51/// );
52/// let subscription_2 = observable.subscribe_with_callback(
53///     move |value| values_2_observer.lock().unwrap().push(value),
54///     move |termination| terminations_observer
55///         .lock()
56///         .unwrap()
57///         .push(termination),
58/// );
59///
60/// // Nothing is emitted until the source is connected.
61/// let connected = controller.connect();
62/// // Dropping the connected controller disconnects the source.
63/// drop(connected);
64/// drop(subscription_1);
65/// drop(subscription_2);
66///
67/// assert_eq!(&*values_1.lock().unwrap(), &[1, 2]);
68/// assert_eq!(&*values_2.lock().unwrap(), &[1, 2]);
69/// assert_eq!(
70///     &*terminations.lock().unwrap(),
71///     &[Termination::Completed]
72/// );
73/// ```
74#[derive(Educe)]
75#[educe(Debug)]
76pub struct ConnectableController<OE, S, State = Disconnected> {
77    source: OE,
78    subject: S,
79    state: State,
80}
81
82impl<OE, S> ConnectableController<OE, S, Disconnected> {
83    pub fn new(source: OE, subject: S) -> Self {
84        Self {
85            source,
86            subject,
87            state: Disconnected,
88        }
89    }
90
91    /// Connects to the source. The returned controller owns the connection and
92    /// disconnects it when dropped.
93    ///
94    /// Ignoring the returned controller would disconnect at the end of the
95    /// statement, so the compiler warns about it:
96    ///
97    /// ```compile_fail
98    /// #![deny(unused_must_use)]
99    /// use rx_rust::{
100    ///     observable::ObservableExt,
101    ///     operators::creating::from_iter::FromIter,
102    /// };
103    ///
104    /// FromIter::new([1_i32]).publish().connect();
105    /// ```
106    #[must_use = "the returned controller owns the source connection"]
107    pub fn connect<'or, T, E>(self) -> ConnectableController<OE, S, Connected<OE::D>>
108    where
109        OE: Observable<'or, T, E> + Clone,
110        S: Observer<T, E> + Clone + MaybeSend + 'or,
111    {
112        let sub = self.source.clone().subscribe(self.subject.clone());
113        ConnectableController {
114            source: self.source,
115            subject: self.subject,
116            state: Connected(sub),
117        }
118    }
119
120    pub fn ref_count<'or, T, E>(self) -> RefCount<'or, T, E, OE, S>
121    where
122        OE: Observable<'or, T, E>,
123        S: Clone,
124    {
125        RefCount::new(self)
126    }
127}
128
129impl<OE, S, D> ConnectableController<OE, S, Connected<D>>
130where
131    D: Disposable,
132{
133    /// Disconnects the source and returns the controller in its disconnected state.
134    pub fn disconnect(self) -> ConnectableController<OE, S, Disconnected> {
135        let Self {
136            source,
137            subject,
138            state,
139        } = self;
140        drop(state);
141        ConnectableController {
142            source,
143            subject,
144            state: Disconnected,
145        }
146    }
147}
148
149impl<OE, S, State> ConnectableController<OE, S, State> {
150    pub fn observable(&self) -> SubjectObservable<S>
151    where
152        S: Clone,
153    {
154        SubjectObservable::new(self.subject.clone())
155    }
156}
rx_rust/operators/connectable/connectable_controller.rs

rx_rust/operators/connectable/
connectable_controller.rs
