Function analyze

flow-plugins/src/bytes_server.rs:45–75

(#[data] state: State, #[body] body: Bytes)

#[post("/analyze/bytes")]
#[openapi(summary = "analyze user define bytes")]
async fn analyze(#[data] state: State, #[body] body: Bytes) -> Result<impl Reply, Rejection> {
    let id = state.id();

    let pyobject: PyObject = Python::with_gil(|py| -> PyResult<_> {
        let ndarray = body.to_pyarray(py);

        Ok([("data", ndarray.to_object(py))].into_py_dict(py).into())
    })
    .map_err(reject_cause)?;

    let r = {
        let (s, r) = oneshot::channel();
        state.mapping.lock().await.insert(id, s);
        let envelope = Envelope::with_info(
            pyobject,
            EnvelopeInfo {
                partial_id: Some(id),
                ..Default::default()
            },
        );
        state.out.send(envelope).await.ok();
        r
    };
    let message = r.await.map_err(reject_cause)?;
    Python::with_gil(|py| -> PyResult<_> {
        let dict: &PyDict = message.extract(py)?;
        flow_rs::helper::uget_slice(py, dict.get_item("data").expect("error key<data>"))
    })
    .map_err(reject_cause)
    .map(|data| RwebBytes::new(&(data.to_vec())))
}

impl BytesServer {
    fn new(_: String, args: &Table) -> BytesServer {