yenru0/zed
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cloud_api_client: Implement Wasm support (#61328)

Browse Source 
Release Notes:

- N/A or Added/Fixed/Improved ...

---------

Co-authored-by: Gaauwe Rombouts <mail@grombouts.nl>
This commit is contained in:
Lukas WirthandGaauwe Rombouts authored and GitHub committed 2026-07-21 13:20:31 +00:00
1 parent abfced8e47
commit dc0c66cfeb
10 files changed
+283 -118
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.github/workflows/run_tests.yml
+1 -1
View File
@@ -660,7 +660,7 @@ jobs:
R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
SCCACHE_BUCKET: sccache-zed
- name: run_tests::check_wasm::cargo_check_wasm
run: cargo -Zbuild-std=std,panic_abort check --target wasm32-unknown-unknown -p gpui_platform
run: cargo -Zbuild-std=std,panic_abort check --target wasm32-unknown-unknown -p gpui_platform -p cloud_api_client
env:
CARGO_TARGET_WASM32_UNKNOWN_UNKNOWN_RUSTFLAGS: -C target-feature=+atomics,+bulk-memory,+mutable-globals
RUSTC_BOOTSTRAP: '1'
Cargo.lock
Generated
+16 -6
View File
@@ -17486,7 +17486,7 @@ dependencies = [
"tracing",
"url",
"uuid",
"webpki-roots",
"webpki-roots 0.26.8",
]
[[package]]
@@ -21160,6 +21160,15 @@ dependencies = [
"rustls-pki-types",
]
[[package]]
name = "webpki-roots"
version = "1.0.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf85cb06032201fa7c6f829d7db5a7e5aa45bcc0655327713065f6f0576731bf"
dependencies = [
"rustls-pki-types",
]
[[package]]
name = "webrtc-sys"
version = "0.3.23"
@@ -22978,23 +22987,24 @@ checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049"
[[package]]
name = "yawc"
version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "19a5d82922135b4ae73a079a4ffb5501e9aadb4d785b8c660eaa0a8b899028c5"
version = "0.3.3"
source = "git+https://github.com/zed-industries/yawc?rev=71a452f551cac178367eaac5d7418a09afa1f3a2#71a452f551cac178367eaac5d7418a09afa1f3a2"
dependencies = [
"base64 0.22.1",
"bytes 1.11.1",
"flate2",
"futures 0.3.32",
"getrandom 0.2.16",
"http-body-util",
"hyper 1.7.0",
"hyper-util",
"js-sys",
"log",
"nom 8.0.0",
"pin-project",
"rand 0.8.6",
"sha1",
"thiserror 1.0.69",
"thiserror 2.0.17",
"tokio",
"tokio-rustls 0.26.4",
"tokio-util",
@@ -23002,7 +23012,7 @@ dependencies = [
"wasm-bindgen",
"wasm-bindgen-futures",
"web-sys",
"webpki-roots",
"webpki-roots 1.0.8",
]
[[package]]
Cargo.toml
+1 -1
View File
@@ -888,7 +888,7 @@ webrtc-sys = "0.3.23"
wgpu = "29.0.4"
windows-core = "0.61"
yaml-rust2 = "0.8"
yawc = "0.2.5"
yawc = { git = "https://github.com/zed-industries/yawc", rev = "71a452f551cac178367eaac5d7418a09afa1f3a2", version = "0.3.3" }
zeroize = "1.8"
zstd = "0.11"
crates/cloud_api_client/Cargo.toml
+3 -1
View File
@@ -16,7 +16,6 @@ anyhow.workspace = true
cloud_api_types.workspace = true
futures.workspace = true
gpui.workspace = true
gpui_tokio.workspace = true
http_client.workspace = true
parking_lot.workspace = true
serde.workspace = true
@@ -24,3 +23,6 @@ serde_json.workspace = true
async-lock.workspace = true
thiserror.workspace = true
yawc.workspace = true
[target.'cfg(not(target_family = "wasm"))'.dependencies]
gpui_tokio.workspace = true
crates/cloud_api_client/src/cloud_api_client.rs
+76 -45
View File
@@ -3,12 +3,9 @@ mod websocket;
use std::sync::Arc;
use anyhow::{Context, Result, anyhow};
use cloud_api_types::websocket_protocol::{PROTOCOL_VERSION, PROTOCOL_VERSION_HEADER_NAME};
use anyhow::{Result, anyhow};
pub use cloud_api_types::*;
use futures::AsyncReadExt as _;
use gpui::{App, Task};
use gpui_tokio::Tokio;
use http_client::http::request;
use http_client::{
AsyncBody, HttpClientWithUrl, HttpRequestExt, Json, Method, Request, Response, StatusCode,
@@ -16,9 +13,6 @@ use http_client::{
use parking_lot::RwLock;
use serde::de::DeserializeOwned;
use thiserror::Error;
use yawc::WebSocket;
use crate::websocket::Connection;
pub use llm_token::LlmApiToken;
@@ -27,6 +21,12 @@ struct Credentials {
access_token: String,
}
#[derive(Clone, Copy)]
enum Authentication {
Credentials,
Session,
}
#[derive(Debug, Error)]
pub enum ClientApiError {
/// 401 — credentials are invalid or expired.
@@ -62,6 +62,7 @@ pub enum ClientApiError {
pub struct CloudApiClient {
credentials: RwLock<Option<Credentials>>,
http_client: Arc<HttpClientWithUrl>,
authentication: Authentication,
}
impl CloudApiClient {
@@ -69,6 +70,15 @@ impl CloudApiClient {
Self {
credentials: RwLock::new(None),
http_client,
authentication: Authentication::Credentials,
}
}
pub fn new_with_session_authentication(http_client: Arc<HttpClientWithUrl>) -> Self {
Self {
credentials: RwLock::new(None),
http_client,
authentication: Authentication::Session,
}
}
@@ -115,35 +125,6 @@ impl CloudApiClient {
.await
}
pub fn connect(&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(Tokio::spawn_result(cx, async move {
let ws = 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(ws))
}))
}
async fn create_llm_token(
&self,
system_id: Option<String>,
@@ -203,10 +184,13 @@ impl CloudApiClient {
request_builder: request::Builder,
body: impl Into<AsyncBody>,
) -> Result<Response<AsyncBody>, ClientApiError> {
let request = {
let request = if matches!(self.authentication, Authentication::Session) {
build_request(request_builder, body, None)
.map_err(ClientApiError::RequestBuildFailed)?
} else {
let credentials = self.credentials.read();
let credentials = credentials.as_ref().ok_or(ClientApiError::NotSignedIn)?;
build_request(request_builder, body, credentials)
build_request(request_builder, body, Some(credentials))
.map_err(ClientApiError::RequestBuildFailed)?
};
@@ -261,10 +245,10 @@ impl CloudApiClient {
.as_ref(),
),
AsyncBody::default(),
&Credentials {
Some(&Credentials {
user_id,
access_token: access_token.into(),
},
}),
)?;
let mut response = self.http_client.send(request).await?;
@@ -331,13 +315,60 @@ impl CloudApiClient {
fn build_request(
req: request::Builder,
body: impl Into<AsyncBody>,
credentials: &Credentials,
credentials: Option<&Credentials>,
) -> Result<Request<AsyncBody>> {
Ok(req
.header("Content-Type", "application/json")
.header(
"Authorization",
format!("{} {}", credentials.user_id, credentials.access_token),
)
.when_some(credentials, |request, credentials| {
request.header(
"Authorization",
format!("{} {}", credentials.user_id, credentials.access_token),
)
})
.body(body.into())?)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn build_session_authenticated_request_without_authorization_header() -> Result<()> {
let request = build_request(
Request::builder().uri("https://cloud.zed.dev/client/users/me"),
AsyncBody::default(),
None,
)?;
assert_eq!(
request
.headers()
.get("Content-Type")
.and_then(|value| value.to_str().ok()),
Some("application/json")
);
assert!(!request.headers().contains_key("Authorization"));
Ok(())
}
#[test]
fn build_credentials_authenticated_request_with_authorization_header() -> Result<()> {
let request = build_request(
Request::builder().uri("https://cloud.zed.dev/client/users/me"),
AsyncBody::default(),
Some(&Credentials {
user_id: 123,
access_token: "token".into(),
}),
)?;
assert_eq!(
request
.headers()
.get("Authorization")
.and_then(|value| value.to_str().ok()),
Some("123 token")
);
Ok(())
}
}
crates/cloud_api_client/src/websocket.rs
+14 -63
View File
@@ -1,73 +1,24 @@
use std::pin::Pin;
use std::time::Duration;
use anyhow::Result;
use cloud_api_types::websocket_protocol::MessageToClient;
use futures::channel::mpsc::unbounded;
use futures::stream::{SplitSink, SplitStream};
use futures::{FutureExt as _, SinkExt as _, Stream, StreamExt as _, TryStreamExt as _, pin_mut};
use gpui::{App, BackgroundExecutor, Task};
use yawc::WebSocket;
use yawc::frame::{FrameView, OpCode};
use futures::Stream;
use futures::channel::mpsc::UnboundedSender;
use yawc::frame::{Frame, OpCode};
const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(1);
#[cfg(not(target_family = "wasm"))]
mod native;
#[cfg(target_family = "wasm")]
mod web;
pub type MessageStream = Pin<Box<dyn Stream<Item = Result<MessageToClient>>>>;
pub struct Connection {
tx: SplitSink<WebSocket, FrameView>,
rx: SplitStream<WebSocket>,
}
impl Connection {
pub fn new(ws: WebSocket) -> Self {
let (tx, rx) = ws.split();
Self { tx, rx }
}
pub fn spawn(self, cx: &App) -> (MessageStream, Task<()>) {
let (mut tx, rx) = (self.tx, self.rx);
let (message_tx, message_rx) = unbounded();
let handle_io = |executor: BackgroundExecutor| async move {
// Send messages on this frequency so the connection isn't closed.
let keepalive_timer = executor.timer(KEEPALIVE_INTERVAL).fuse();
futures::pin_mut!(keepalive_timer);
let rx = rx.fuse();
pin_mut!(rx);
loop {
futures::select_biased! {
_ = keepalive_timer => {
let _ = tx.send(FrameView::ping(Vec::new())).await;
keepalive_timer.set(executor.timer(KEEPALIVE_INTERVAL).fuse());
}
frame = rx.next() => {
let Some(frame) = frame else {
break;
};
match frame.opcode {
OpCode::Binary => {
let message_result = MessageToClient::deserialize(&frame.payload);
message_tx.unbounded_send(message_result).ok();
}
OpCode::Close => {
break;
}
_ => {}
}
}
}
}
};
let task = cx.spawn(async move |cx| handle_io(cx.background_executor().clone()).await);
(message_rx.into_stream().boxed(), task)
fn forward_frame(frame: Frame, message_tx: &UnboundedSender<Result<MessageToClient>>) -> bool {
match frame.opcode() {
OpCode::Binary => message_tx
.unbounded_send(MessageToClient::deserialize(frame.payload()))
.is_ok(),
OpCode::Close => false,
OpCode::Continuation | OpCode::Text | OpCode::Ping | OpCode::Pong => true,
}
}
crates/cloud_api_client/src/websocket/native.rs
+91
View File
@@ -0,0 +1,91 @@
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))
}))
}
}
crates/cloud_api_client/src/websocket/web.rs
+79
View File
@@ -0,0 +1,79 @@
use 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))
}))
}
}
script/licenses/zed-licenses.toml
+1
View File
@@ -20,6 +20,7 @@ accepted = [
"Zlib",
"BSL-1.0",
"bzip2-1.0.6",
"CDLA-Permissive-2.0",
]
[procinfo.clarify]
tooling/xtask/src/tasks/workflows/run_tests.rs
+1 -1
View File
@@ -499,7 +499,7 @@ fn check_wasm() -> NamedJob {
fn cargo_check_wasm() -> Step<Run> {
named::bash(concat!(
"cargo -Zbuild-std=std,panic_abort ",
"check --target wasm32-unknown-unknown -p gpui_platform",
"check --target wasm32-unknown-unknown -p gpui_platform -p cloud_api_client",
))
.add_env((
"CARGO_TARGET_WASM32_UNKNOWN_UNKNOWN_RUSTFLAGS",