Method new

flow-rs/src/node/shared.rs:47–111

(
        local_key: u64,
        rx: ReceiverT<SharedConns>,
        cfg: &crate::config::interlayer::Node,
        graphs: &crate::config::interlayer::Config,
    )

impl Shared {
    pub fn new(
        local_key: u64,
        rx: ReceiverT<SharedConns>,
        cfg: &crate::config::interlayer::Node,
        graphs: &crate::config::interlayer::Config,
    ) -> Result<Shared> {
        let mut nodes = load_static(local_key, cfg)?;
        let mut inputs = HashMap::new();
        let mut outputs = HashMap::new();

        let find_cap =
            |port| {
                let mut cap = 0usize;
                for graph in graphs
                    .graphs
                    .iter()
                    .filter(|graph| !graph.nodes.keys().any(|node| node == &cfg.entity.name))
                {
                    cap =
                        std::cmp::max(
                            cap,
                            graph
                                .connections
                                .values()
                                .map(|conn| {
                                    if conn.rx.iter().chain(conn.tx.iter()).any(|p| {
                                        p.node_name == cfg.entity.name && &p.port_name == port
                                    }) {
                                        conn.cap
                                    } else {
                                        0
                                    }
                                })
                                .max()
                                .expect("unused port in shared nodes"),
                        );
                }
                cap
            };

        for input in &cfg.inputs {
            let cap = find_cap(input);
            let chan = ChannelStorage::bound(cap);
            for node in &mut nodes {
                node.set_port(input.as_str(), None, &chan);
            }
            inputs.insert(input.clone(), Arc::new(chan.sender()));
        }

        for output in &cfg.outputs {
            let cap = find_cap(output);
            let chan = ChannelStorage::bound(cap);
            for node in &mut nodes {
                node.set_port(output.as_str(), None, &chan);
            }
            outputs.insert(output.clone(), chan.receiver());
        }