Function send_stream

modules/llrt_fetch/src/fetch.rs:190–299

(
    ctx: &Ctx<'js>,
    client: &HyperClient,
    stream: Class<'js, ReadableStream<'js>>,
    method: Method,
    method_string: String,
    uri: Uri,
    headers: Option<&Headers>,
    start: Inst

}

#[allow(clippy::too_many_arguments)]
async fn send_stream<'js>(
    ctx: &Ctx<'js>,
    client: &HyperClient,
    stream: Class<'js, ReadableStream<'js>>,
    method: Method,
    method_string: String,
    uri: Uri,
    headers: Option<&Headers>,
    start: Instant,
    abort_receiver: Option<mc_oneshot::Receiver<Value<'js>>>,
) -> Result<Response<'js>> {
    let (tx, rx) = tokio::sync::mpsc::channel::<Bytes>(128);
    let (err_tx, err_rx) = tokio::sync::oneshot::channel::<String>();
    let ctx2 = ctx.clone();

    // Spawn stream reader in JS context
    ctx.spawn_exit_simple({
        async move {
            let get_reader: Function = stream.get("getReader")?;
            let reader: Object = get_reader.call((This(stream.clone()),))?;
            let read_fn: Function = reader.get("read")?;

            loop {
                let promise: rquickjs::Promise = read_fn.call((This(reader.clone()),))?;
                let read_result = match promise.into_future::<Object>().await.catch(&ctx2) {
                    Ok(r) => r,
                    Err(e) => {
                        let msg = match e {
                            CaughtError::Exception(ex) => ex.message().unwrap_or_default(),
                            CaughtError::Value(v) => v
                                .as_string()
                                .and_then(|s| s.to_string().ok())
                                .unwrap_or_else(|| "Stream error".into()),
                            CaughtError::Error(e) => e.to_string(),
                        };
                        let _ = err_tx.send(msg);
                        break;
                    },
                };
                let done: bool = read_result.get("done").unwrap_or(true);
                if done {
                    break;
                }
                if let Ok(value) = read_result.get::<_, Value>("value") {
                    // Per fetch spec, stream chunks must be Uint8Array. Anything
                    // else (ArrayBuffer, Blob, String, null, etc.) is an error.
                    let bytes = match rquickjs::TypedArray::<u8>::from_value(value) {
                        Ok(typed_array) => typed_array.as_bytes().map(Bytes::copy_from_slice),
                        Err(_) => {
                            let _ = err_tx
                                .send("Failed to read body: chunk is not a Uint8Array".into());
                            break;
                        },
                    };
                    if let Some(bytes) = bytes {
                        if tx.send(bytes).await.is_err() {
                            break;