Something went wrong. Try again.
fork of gpui
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980use std::time::Duration;
use anyhow::{Result, anyhow};
use futures::channel::mpsc::unbounded;use futures::stream::SplitStream;use futures::{FutureExt as _, StreamExt as _, TryStreamExt as _};use gpui::{App, Task};use yawc::WebSocket;
use super::{MessageStream, forward_frame};use crate::CloudApiClient;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
pub struct Connection { rx: SplitStream<WebSocket>,}
impl Connection { fn new(websocket: WebSocket) -> Self { let (_, rx) = websocket.split(); Self { rx } }
pub fn spawn(self, cx: &App) -> (MessageStream, Task<()>) { let mut rx = self.rx; let (message_tx, message_rx) = unbounded(); let task = cx.spawn(async move |_cx| { while let Some(frame) = rx.next().await { let frame = match frame { Ok(frame) => frame, Err(error) => { let error = anyhow!("Cloud WebSocket error: {error}"); if message_tx.unbounded_send(Err(error)).is_err() { break; } continue; } }; 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 client = self.clone(); let executor = cx.background_executor().clone(); Ok(cx.spawn(async move |_cx| { let mut connect_url = client .http_client .build_zed_cloud_url("/client/users/connect")?; connect_url .set_scheme(match connect_url.scheme() { "https" => "wss", "http" => "ws", scheme => return Err(anyhow!("invalid URL scheme: {scheme}")), }) .map_err(|_| anyhow!("failed to set URL scheme"))?;
let connect = WebSocket::connect(connect_url).fuse(); let timeout = executor.timer(CONNECT_TIMEOUT).fuse(); futures::pin_mut!(connect, timeout); let websocket = futures::select_biased! { result = connect => result.map_err(|error| anyhow!("failed to connect to Cloud WebSocket: {error}"))?, _ = timeout => return Err(anyhow!("timed out connecting to Cloud WebSocket")), };
Ok(Connection::new(websocket)) })) }}