Method start

flow-rs/src/node/transform.rs:50–85

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

impl Actor for DynOutTransform {
    fn start(
        mut self: Box<Self>,
        _: Context,
        _: ResourceCollection,
    ) -> rt::task::JoinHandle<Result<()>> {
        let (s, r) = rt::channel::unbounded();
        rt::task::spawn(async move {
            while let Ok((_, out)) = self.out.fetch().await {
                let inp = self.inp.clone();
                s.send(rt::task::spawn(async move {
                    let mut empty_n = 0;
                    loop {
                        while let Ok(msg) = inp.recv_any().await {
                            out.send_any(msg).await.ok();
                        }
                        if inp.is_closed() {
                            break;
                        }
                        let n = inp.empty_n();
                        for _ in empty_n..n {
                            out.send_any(DummyEnvelope {}.seal()).await.ok();
                        }
                        empty_n = n;
                    }
                }))
                .await
                .ok();
            }
        });
        rt::task::spawn(async move {
            while let Ok(handle) = r.recv().await {
                handle.await;
            }
            Ok(())
        })
    }
}

node_register!("DynOutTransform", DynOutTransform);