diff --git a/crates/gpui_web/Cargo.toml b/crates/gpui_web/Cargo.toml index a908243796..aed7c9fd27 100644 --- a/crates/gpui_web/Cargo.toml +++ b/crates/gpui_web/Cargo.toml @@ -61,6 +61,9 @@ web-sys = { version = "0.3", features = [ "MouseEvent", "Navigator", "PointerEvent", + "ReadableStream", + "ReadableStreamDefaultReader", + "ReadableStreamReadResult", "ResizeObserver", "ResizeObserverBoxOptions", "ResizeObserverEntry", diff --git a/crates/gpui_web/src/http_client.rs b/crates/gpui_web/src/http_client.rs index 6e585682a0..4a00324dc3 100644 --- a/crates/gpui_web/src/http_client.rs +++ b/crates/gpui_web/src/http_client.rs @@ -1,8 +1,16 @@ use crate::WebDispatcher; use anyhow::{Context as _, anyhow}; -use futures::{AsyncReadExt as _, channel::oneshot}; +use futures::{ + AsyncRead, AsyncReadExt as _, FutureExt as _, SinkExt as _, TryStreamExt as _, + channel::{mpsc, oneshot}, +}; use http_client::{AsyncBody, HttpClient, RedirectPolicy}; -use std::sync::Arc; +use std::{ + io, + pin::Pin, + sync::Arc, + task::{Context, Poll}, +}; use wasm_bindgen::JsCast as _; use wasm_bindgen::prelude::*; @@ -172,26 +180,24 @@ async fn fetch( } } - // The entire response body is eagerly buffered into memory via - // `arrayBuffer()`. The Fetch API does not expose a synchronous - // streaming interface; streaming would require `ReadableStream` - // interop which is significantly more complex. - let body_promise = web_response - .array_buffer() - .map_err(|error| anyhow!("failed to initiate response body read: {error:?}"))?; - let body_value = wasm_bindgen_futures::JsFuture::from(body_promise) - .await - .map_err(|error| anyhow!("failed to read response body: {error:?}"))?; - let array_buffer: js_sys::ArrayBuffer = body_value - .dyn_into() - .map_err(|error| anyhow!("response body is not an ArrayBuffer: {error:?}"))?; - let response_bytes = js_sys::Uint8Array::new(&array_buffer).to_vec(); + let body = match web_response.body() { + Some(stream) => { + let reader = stream + .get_reader() + .dyn_into::() + .map_err(|error| { + anyhow!("response body reader has an unexpected type: {error:?}") + })?; + AsyncBody::from_reader(ReadableStreamBody::new(reader)) + } + None => AsyncBody::empty(), + }; - builder - .body(AsyncBody::from(response_bytes)) - .map_err(|error| anyhow!(error)) + builder.body(body).map_err(|error| anyhow!(error)) } +// Request bodies are buffered into memory because streaming uploads require +// half-duplex Fetch support that browsers largely don't ship yet. async fn read_body_to_bytes(mut body: AsyncBody) -> anyhow::Result>> { let mut buffer = Vec::new(); body.read_to_end(&mut buffer).await?; @@ -201,3 +207,103 @@ async fn read_body_to_bytes(mut body: AsyncBody) -> anyhow::Result>>>, + // Dropping this sender resolves the pump's cancellation future, which + // cancels the browser-side `ReadableStream`. + _cancellation: oneshot::Sender<()>, +} + +impl ReadableStreamBody { + fn new(reader: web_sys::ReadableStreamDefaultReader) -> Self { + let (chunks_sender, chunks_receiver) = mpsc::channel(RESPONSE_BODY_CHANNEL_CAPACITY); + let (cancellation, cancellation_receiver) = oneshot::channel(); + wasm_bindgen_futures::spawn_local(pump_response_body( + reader, + chunks_sender, + cancellation_receiver, + )); + Self { + chunks: chunks_receiver.into_async_read(), + _cancellation: cancellation, + } + } +} + +impl AsyncRead for ReadableStreamBody { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut [u8], + ) -> Poll> { + Pin::new(&mut self.chunks).poll_read(cx, buffer) + } +} + +async fn pump_response_body( + reader: web_sys::ReadableStreamDefaultReader, + mut chunks: mpsc::Sender>>, + cancellation: oneshot::Receiver<()>, +) { + let cancellation = cancellation.fuse(); + futures::pin_mut!(cancellation); + + loop { + let read = wasm_bindgen_futures::JsFuture::from(reader.read()).fuse(); + futures::pin_mut!(read); + let result = futures::select_biased! { + _ = cancellation => { + cancel_reader(&reader).await; + return; + } + result = read => result, + }; + + let chunk = result + .map_err(|error| io::Error::other(format!("response stream failed: {error:?}"))) + .and_then(response_chunk); + match chunk { + Ok(Some(chunk)) => { + if chunks.send(Ok(chunk)).await.is_err() { + cancel_reader(&reader).await; + return; + } + } + Ok(None) => return, + Err(error) => { + if chunks.send(Err(error)).await.is_err() { + log::debug!("response body receiver was dropped after a stream error"); + } + return; + } + } + } +} + +fn response_chunk(result: JsValue) -> io::Result>> { + // `ReadableStreamReadResult` is a dictionary type, so there is no runtime + // class to check against; `unchecked_into` is the only available cast. + let result: web_sys::ReadableStreamReadResult = result.unchecked_into(); + if result.get_done().unwrap_or(false) { + return Ok(None); + } + + result + .get_value() + .dyn_into::() + .map(|bytes| Some(bytes.to_vec())) + .map_err(|value| { + io::Error::other(format!( + "response stream yielded a non-byte chunk: {value:?}" + )) + }) +} + +async fn cancel_reader(reader: &web_sys::ReadableStreamDefaultReader) { + if let Err(error) = wasm_bindgen_futures::JsFuture::from(reader.cancel()).await { + log::debug!("failed to cancel response body reader: {error:?}"); + } +}