Struct holochain::conductor::interface::SignalBroadcaster

pub struct SignalBroadcaster { /* private fields */ }

A collection of Senders to be used for emitting Signals from a Cell. There is one Sender per attached Interface

Implementations

impl SignalBroadcaster

pub fn new(senders: Vec<Sender<Signal>>) -> Self

internal constructor

Examples found in repository

src/conductor/conductor.rs (line 2049)

        pub fn signal_broadcaster(&self) -> SignalBroadcaster {
            let senders = self
                .app_interfaces
                .share_ref(|ai| ai.values().map(|i| i.signal_tx()).cloned().collect());
            SignalBroadcaster::new(senders)
        }

pub fn subscribe_separately(&self) -> Vec<Receiver<Signal>>

Get a list of Signal receivers, one per sender (per interface)

Examples found in repository

src/conductor/interface.rs (line 94)

    pub fn subscribe_merged(&self) -> impl tokio_stream::Stream<Item = Signal> {
        use tokio_stream::StreamExt;
        let mut streams = tokio_stream::StreamMap::new();
        for (i, rx) in self.subscribe_separately().into_iter().enumerate() {
            streams.insert(i, tokio_stream::wrappers::BroadcastStream::new(rx));
        }
        streams.map(|(_, signal)| signal.expect("Couldn't receive a signal"))
    }

pub fn subscribe_merged(&self) -> impl Stream<Item = Signal>

Get a single merged stream of all Signals from all broadcasters

Examples found in repository

src/sweettest/sweet_conductor_handle.rs (line 101)

    pub async fn signal_stream(&self) -> impl tokio_stream::Stream<Item = Signal> {
        self.0.signal_broadcaster().subscribe_merged()
    }

src/sweettest/sweet_conductor.rs (line 89)

    pub async fn new(
        handle: ConductorHandle,
        env_dir: TestDir,
        config: ConductorConfig,
    ) -> SweetConductor {
        // Automatically add a test app interface
        handle
            .add_test_app_interface(AppInterfaceId::default())
            .await
            .expect("Couldn't set up test app interface");

        // Get a stream of all signals since conductor startup
        let signal_stream = handle.signal_broadcaster().subscribe_merged();

        // XXX: this is a bit wonky.
        // We create a Spaces instance here purely because it's easier to initialize
        // the per-space databases this way. However, we actually use the TestEnvs
        // to actually access those databases.
        // As a TODO, we can remove the need for TestEnvs in sweettest or have
        // some other better integration between the two.
        let spaces = Spaces::new(&ConductorConfig {
            environment_path: env_dir.to_path_buf().into(),
            ..Default::default()
        })
        .unwrap();

        let keystore = handle.keystore().clone();

        Self {
            handle: Some(SweetConductorHandle(handle)),
            db_dir: env_dir,
            keystore,
            spaces,
            config,
            dnas: Vec::new(),
            signal_stream: Some(Box::new(signal_stream)),
        }
    }

Trait Implementations

impl Clone for SignalBroadcaster

fn clone(&self) -> SignalBroadcaster

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for SignalBroadcaster

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

Formats the value using the given formatter.