Method exec

flow-plugins/src/video_server.rs:114–201

(&mut self, _: &flow_rs::graph::Context)

    }
    async fn finalize(&mut self) {}
    async fn exec(&mut self, _: &flow_rs::graph::Context) -> Result<()> {
        let (s, r) = flow_rs::rt::channel::unbounded();
        let state = State::new(s);
        let mapping = state.mapping.clone();
        let messages = Messages::new().with_lock();
        let messages_cloned = messages.clone();

        let (spec, filter) = openapi::spec().build(move || {
            start(state.clone())
                .or(stop(state.clone()))
                .or(list(state))
                .or(get_msgs(messages_cloned))
        });

        let mut recv_msgs = FuturesUnordered::new();
        let mut recv_conns = FuturesUnordered::new();
        let mut spawn_decode = FuturesUnordered::new();
        let listen = serve(filter.or(openapi_docs(spec)))
            .run(([0, 0, 0, 0], self.port))
            .fuse();
        recv_conns.push(self.inp.fetch());
        spawn_decode.push(r.recv());

        pin_mut!(listen);

        loop {
            select! {
                // server listen
                _ = listen => break,
                // spawn task to wait message from subgraph
                conns = recv_conns.select_next_some() => {
                    if let Ok((id, port)) = conns {
                        let messages = messages.clone();
                        recv_msgs.push(async move {
                            while let Ok(mut msg) = port.recv().await {
                                let messages_map = &mut messages.lock().await.mapping;
                                let messages = messages_map.entry(id).or_default();
                                let msg = Python::with_gil(|py| -> PyResult<_> {
                                    let msg: PyObject = msg.unpack();
                                    let msg = msg.as_ref(py).extract()?;
                                    Ok(msg)
                                }).expect("plugin[VideoServier] only support python string as input");
                                messages.push(msg);
                            }
                        });
                        recv_conns.push(self.inp.fetch());
                    }
                },
                // wait message from subgraph
                _ = recv_msgs.select_next_some() => {},
                // spawn subgraph
                ret = spawn_decode.select_next_some() => {
                    if let Ok((id, url, waker)) = ret {
                        self.out.create(id, self.resources.clone().unwrap(), Default::default()).await.expect("broker has closed");
                        let (_, port) = self.out.fetch().await.expect("broker has closed");
                        let url = urlencoding::decode(&url).unwrap().into_owned();
                        let url_cloned = url.clone();
                        let port_cloned = port.clone();