Function handle_control_socket

packages/server/src/api/routes.rs:2835–2888

(state: AppState, udid: String, socket: WebSocket)

}

async fn handle_control_socket(state: AppState, udid: String, socket: WebSocket) {
    let session = match state.registry.get_or_create_async(&udid).await {
        Ok(session) => session,
        Err(error) => {
            tracing::debug!("Failed to create control session for {udid}: {error}");
            return;
        }
    };
    if let Err(error) = session.ensure_started_async().await {
        tracing::debug!("Failed to start control session for {udid}: {error}");
        return;
    }

    let (mut sender, mut receiver) = socket.split();
    let _ = sender
        .send(Message::Text(
            json_value!({ "type": "ready", "udid": udid })
                .to_string()
                .into(),
        ))
        .await;
    let (control_tx, control_rx) = mpsc::unbounded_channel::<ControlMessage>();
    let bridge = state.registry.bridge().clone();
    let control_task = task::spawn(run_control_queue(session, bridge, udid.clone(), control_rx));

    while let Some(message) = receiver.next().await {
        let text = match message {
            Ok(Message::Text(text)) => text,
            Ok(Message::Binary(bytes)) => match String::from_utf8(bytes.to_vec()) {
                Ok(text) => text.into(),
                Err(_) => continue,
            },
            Ok(Message::Close(_)) => break,
            Ok(Message::Ping(_)) | Ok(Message::Pong(_)) => continue,
            Err(error) => {
                tracing::debug!("Control WebSocket closed for {udid}: {error}");
                break;
            }
        };

        let control_message = match serde_json::from_str::<ControlMessage>(&text) {
            Ok(message) => message,
            Err(error) => {
                tracing::debug!("Invalid control message for {udid}: {error}");
                continue;
            }
        };
        if control_tx.send(control_message).is_err() {
            break;
        }
    }
    drop(control_tx);
    let _ = control_task.await;
}

async fn run_control_queue(
    session: SimulatorSession,