Method exec

flow-rs/src/node/demux.rs:39–81

(&mut self, _: &Context)

    }

    async fn exec(&mut self, _: &Context) -> Result<()> {
        if let Ok(msg) = self.inp.recv_any().await {
            let id = msg
                .info()
                .to_addr
                .expect("the envelope has no destination address");
            let tag = msg.info().tag.as_ref();
            if msg.is_none() {
                self.out.fetch_with_cache().await.remove(&id);
                if let Some(task) = self.tasks.remove(&id) {
                    task.await.ok();
                }
            } else {
                if !self.out.cache().contains_key(&id) {
                    if let Some(tag) = tag {
                        self.tasks.insert(
                            id,
                            self.out
                                .create_spec(
                                    id,
                                    tag,
                                    self.resources.clone().unwrap(),
                                    Default::default(),
                                )
                                .await
                                .expect("create subgraph fault"),
                        );
                    } else {
                        self.tasks.insert(
                            id,
                            self.out
                                .create(id, self.resources.clone().unwrap(), Default::default())
                                .await
                                .expect("create subgraph fault"),
                        );
                    }
                }
                let out = self.out.fetch_with_cache().await.get(&id).unwrap();
                out.send_any(msg).await.ok();
            }
        }
        Ok(())
    }
}

node_register!("DynDemux", DynDemux);