Method start

flow-rs/src/node/shared.rs:136–244

(
        mut self: Box<Self>,
        ctx: Context,
        resources: ResourceCollection,
    )

impl Actor for Shared {
    fn start(
        mut self: Box<Self>,
        ctx: Context,
        resources: ResourceCollection,
    ) -> rt::task::JoinHandle<Result<()>> {
        let (s1, r1) = unbounded::<u64>();
        let mut s2s = vec![];
        let mut r2s = vec![];
        for _ in 0..self.outputs.len() {
            let (s2, r2) = unbounded::<u64>();
            s2s.push(s2);
            r2s.push(r2);
        }
        let inputs = self.inputs.clone();
        if !inputs.is_empty() {
            // close in order
            crate::rt::task::spawn(async move {
                while let Ok(id) = r1.recv().await {
                    for input in inputs.values() {
                        input.send_any(DummyEnvelope {}.seal()).await.ok();
                    }
                    for s2 in &s2s {
                        s2.send(id).await.ok();
                    }
                }
            });
        }

        let mut outputs: HashMap<_, Arc<RwLock<HashMap<_, Sender>>>> = HashMap::new();
        for (k, input) in std::mem::take(&mut self.outputs) {
            let demux = Arc::new(RwLock::new(HashMap::new()));
            outputs.insert(k, demux.clone());
            let r2 = r2s.pop().unwrap();
            crate::rt::task::spawn(async move {
                loop {
                    match input.recv_any().await {
                        Ok(msg) => {
                            let outputs = demux.read().await;
                            let output = outputs
                                .get(
                                    msg.info()
                                        .transfer_addr
                                        .as_ref()
                                        .expect("lost transfer_address"),
                                )
                                .expect("unexpected transfer_address");
                            output.send_any(msg).await.ok();
                        }
                        _ if !input.is_closed() => {
                            if let Ok(id) = r2.recv().await {
                                demux.write().await.remove(&id);
                            }
                        }
                        _ => break,
                    }
                }
            });
        }