Struct ConnectableController 

pub struct ConnectableController<OE, S, State = Disconnected> { /* private fields */ }

Multicasts a source Observable through a Subject, but waits until its connect method is called before subscribing to the source and emitting items to its observers. Subscribe to the multicast output through observable. See https://reactivex.io/documentation/operators/connect.html

Examples

use rx_rust::{
    observable::ObservableExt,
    observer::Termination,
    operators::{
        connectable::connectable_controller::ConnectableController,
        creating::from_iter::FromIter,
    },
    subject::publish_subject::PublishSubject,
};

use std::{convert::Infallible, sync::{Arc, Mutex}};

let values_1 = Arc::new(Mutex::new(Vec::new()));
let values_2 = Arc::new(Mutex::new(Vec::new()));
let terminations = Arc::new(Mutex::new(Vec::new()));

let subject: PublishSubject<'_, i32, Infallible> = PublishSubject::default();
let controller = ConnectableController::new(FromIter::new(vec![1, 2]), subject);
let observable = controller.observable();
let values_1_observer = Arc::clone(&values_1);
let values_2_observer = Arc::clone(&values_2);
let terminations_observer = Arc::clone(&terminations);

let subscription_1 = observable.clone().subscribe_with_callback(
    move |value| values_1_observer.lock().unwrap().push(value),
    |_| {},
);
let subscription_2 = observable.subscribe_with_callback(
    move |value| values_2_observer.lock().unwrap().push(value),
    move |termination| terminations_observer
        .lock()
        .unwrap()
        .push(termination),
);

// Nothing is emitted until the source is connected.
let connected = controller.connect();
// Dropping the connected controller disconnects the source.
drop(connected);
drop(subscription_1);
drop(subscription_2);

assert_eq!(&*values_1.lock().unwrap(), &[1, 2]);
assert_eq!(&*values_2.lock().unwrap(), &[1, 2]);
assert_eq!(
    &*terminations.lock().unwrap(),
    &[Termination::Completed]
);

Implementations

impl<OE, S> ConnectableController<OE, S, Disconnected>

pub fn new(source: OE, subject: S) -> Self

pub fn connect<'or, T, E>( self, ) -> ConnectableController<OE, S, Connected<OE::D>>

where OE: Observable<'or, T, E> + Clone, S: Observer<T, E> + Clone + MaybeSend + 'or,

Connects to the source. The returned controller owns the connection and disconnects it when dropped.

Ignoring the returned controller would disconnect at the end of the statement, so the compiler warns about it:

#![deny(unused_must_use)]
use rx_rust::{
    observable::ObservableExt,
    operators::creating::from_iter::FromIter,
};

FromIter::new([1_i32]).publish().connect();

pub fn ref_count<'or, T, E>(self) -> RefCount<'or, T, E, OE, S>

where OE: Observable<'or, T, E>, S: Clone,

impl<OE, S, D> ConnectableController<OE, S, Connected<D>>

where D: Disposable,

pub fn disconnect(self) -> ConnectableController<OE, S, Disconnected>

Disconnects the source and returns the controller in its disconnected state.

impl<OE, S, State> ConnectableController<OE, S, State>

pub fn observable(&self) -> SubjectObservable<S>

where S: Clone,

Trait Implementations

impl<OE, S, State> Debug for ConnectableController<OE, S, State>

where OE: Debug, S: Debug, State: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.