use std::time::Duration; use anyhow::{Context as _, Result, anyhow};use cloud_api_types::websocket_protocol::{PROTOCOL_VERSION, PROTOCOL_VERSION_HEADER_NAME};use futures::channel::mpsc::unbounded;use futures::stream::{SplitSink, SplitStream};use futures::{FutureExt as _, SinkExt as _, StreamExt as _, TryStreamExt as _};use gpui::{App, Task};use http_client::http::request;use yawc::frame::Frame;use yawc::{TcpWebSocket, WebSocket}; use super::{MessageStream, forward_frame};use crate::CloudApiClient; const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(1); pub struct Connection {    tx: SplitSink<TcpWebSocket, Frame>,    rx: SplitStream<TcpWebSocket>,} impl Connection {    fn new(websocket: TcpWebSocket) -> Self {        let (tx, rx) = websocket.split();        Self { tx, rx }    }     pub fn spawn(self, cx: &App) -> (MessageStream, Task<()>) {        let mut tx = self.tx;        let rx = self.rx.fuse();        let (message_tx, message_rx) = unbounded();        let executor = cx.background_executor().clone();        let task = cx.spawn(async move |_cx| {            let keepalive_timer = executor.timer(KEEPALIVE_INTERVAL).fuse();            futures::pin_mut!(keepalive_timer, rx);             loop {                futures::select_biased! {                    _ = keepalive_timer => {                        if tx.send(Frame::ping(Vec::new())).await.is_err() {                            break;                        }                        keepalive_timer.set(executor.timer(KEEPALIVE_INTERVAL).fuse());                    }                    frame = rx.next() => {                        let Some(frame) = frame else {                            break;                        };                        if !forward_frame(frame, &message_tx) {                            break;                        }                    }                }            }        });         (message_rx.into_stream().boxed(), task)    }} impl CloudApiClient {    pub fn connect(self: &std::sync::Arc<Self>, cx: &App) -> Result<Task<Result<Connection>>> {        let mut connect_url = self            .http_client            .build_zed_cloud_url("/client/users/connect")?;        connect_url            .set_scheme(match connect_url.scheme() {                "https" => "wss",                "http" => "ws",                scheme => Err(anyhow!("invalid URL scheme: {scheme}"))?,            })            .map_err(|_| anyhow!("failed to set URL scheme"))?;         let credentials = self.credentials.read();        let credentials = credentials.as_ref().context("no credentials provided")?;        let authorization_header = format!("{} {}", credentials.user_id, credentials.access_token);         Ok(gpui_tokio::Tokio::spawn_result(cx, async move {            let websocket = WebSocket::connect(connect_url)                .with_request(                    request::Builder::new()                        .header("Authorization", authorization_header)                        .header(PROTOCOL_VERSION_HEADER_NAME, PROTOCOL_VERSION.to_string()),                )                .await?;             Ok(Connection::new(websocket))        }))    }}