Function connected

flow-rs/src/debug/server.rs:27–82

(ws: WebSocket, graph: String)

static CONNECTED: AtomicBool = AtomicBool::new(false);

async fn connected(ws: WebSocket, graph: String) -> Result<()> {
    let (mut tx, mut rx) = ws.split();

    while PORT.1.try_recv().is_ok() {}
    let init_event = serde_json::to_string(&serde_json::json!({
        "ty": TYPE_EVENT,
        "event": EVENT_INITIALIZED,
        "features": FeatureCommand::registry_global().keys(),
        "graph": graph
    }))?;
    tx.send(Message::text(init_event)).await?;

    flow_rs::rt::task::spawn(async move {
        while let Ok(msg) = PORT.1.recv().await {
            if msg.ty == "__CLOSED__" {
                tx.send(Message::text(
                    serde_json::to_string(
                        &serde_json::json!({ "ty": TYPE_EVENT, "event": EVENT_TERMINATED }),
                    )
                    .unwrap(),
                ))
                .await
                .ok();
                break;
            }
            if tx
                .send(Message::text(serde_json::to_string(&msg).unwrap()))
                .await
                .is_err()
            {
                break;
            }
        }
    });

    while let Some(Ok(msg)) = rx.next().await {
        if let Ok(s) = msg.to_str() {
            let protocol: ProtocolMessage = serde_json::from_str(s)?;
            match protocol.ty.as_str() {
                TYPE_REQUEST => {
                    let req: RequestMessage =
                        serde_json::from_value(serde_json::Value::Object(protocol.others))?;
                    dowith_request(req).await?;
                }
                TYPE_EVENT => {
                    let event: EventMessage =
                        serde_json::from_value(serde_json::Value::Object(protocol.others))?;
                    dowith_event(event).await?;
                }
                _ => unreachable!(),
            }
        }
    }

    Ok(())
}

async fn reset() {