fix: accept typed arrays in writable.write()

crates/runtime/src/streams.rs

@@ -17,6 +17,34 @@ use rquickjs::{Ctx, Function, Object, Persistent, Value};
 
 use std::cell::Cell;
 
+macro_rules! copy_typed_array_as {
+    ($obj:expr, $ty:expr, $t:ty) => {{
+        let Some(ta) = $obj.as_typed_array::<$t>() else {
+            return Ok(None);
+        };
+
+        let slice: &[$t] = ta.as_ref();
+        let count = slice.len();
+
+        assert_eq!($ty.abi_payload_size(), std::mem::size_of::<$t>());
+        assert!($ty.abi_payload_align() >= std::mem::align_of::<$t>());
+
+        let byte_len = count
+            .checked_mul(std::mem::size_of::<$t>())
+            .ok_or_else(|| rquickjs::Error::new_from_js("number", "buffer size overflow"))?;
+
+        let buf = BufferGuard::new_zeroed(byte_len, $ty.abi_payload_align());
+        if byte_len > 0 {
+            unsafe {
+                let src = slice.as_ptr() as *const u8;
+                let dst = buf.ptr();
+                std::ptr::copy_nonoverlapping(src, dst, byte_len);
+            }
+        }
+        Some((buf, count))
+    }};
+}
+
 /// Rust side state for the readable end of a component-model stream.
 #[derive(Trace, JsLifetime)]
 pub(crate) struct StreamReadable {
@@ -137,6 +165,36 @@ pub(crate) fn make_stream<'js>(
     Ok(result.into_value())
 }
 
+/// Fast path for `writable.write(typedArray)`
+fn try_typed_array_to_buffer<'js>(
+    data: &Value<'js>,
+    ty: &wit_dylib_ffi::Stream,
+) -> rquickjs::Result<Option<(BufferGuard, usize)>> {
+    let Some(elem_ty) = ty.ty() else {
+        return Ok(None);
+    };
+
+    let Some(obj) = data.as_object() else {
+        return Ok(None);
+    };
+
+    let pair = match elem_ty {
+        wit_dylib_ffi::Type::U8 => copy_typed_array_as!(obj, ty, u8),
+        wit_dylib_ffi::Type::S8 => copy_typed_array_as!(obj, ty, i8),
+        wit_dylib_ffi::Type::U16 => copy_typed_array_as!(obj, ty, u16),
+        wit_dylib_ffi::Type::S16 => copy_typed_array_as!(obj, ty, i16),
+        wit_dylib_ffi::Type::U32 => copy_typed_array_as!(obj, ty, u32),
+        wit_dylib_ffi::Type::S32 => copy_typed_array_as!(obj, ty, i32),
+        wit_dylib_ffi::Type::U64 => copy_typed_array_as!(obj, ty, u64),
+        wit_dylib_ffi::Type::S64 => copy_typed_array_as!(obj, ty, i64),
+        wit_dylib_ffi::Type::F32 => copy_typed_array_as!(obj, ty, f32),
+        wit_dylib_ffi::Type::F64 => copy_typed_array_as!(obj, ty, f64),
+        _ => return Ok(None),
+    };
+
+    Ok(pair)
+}
+
 fn stream_read<'js>(
     this: This<Class<'js, StreamReadable>>,
     ctx: Ctx<'js>,
@@ -245,7 +303,9 @@ fn stream_write<'js>(
     let ty = ctx.wit().stream(type_index as usize);
 
     let mut call = QjsCallContext::default();
-    let (buffer, write_count) = if let Some(arr) = data.as_array() {
+    let (buffer, write_count) = if let Some(pair) = try_typed_array_to_buffer(&data, &ty)? {
+        pair
+    } else if let Some(arr) = data.as_array() {
         let count = arr.len();
         let buf_size = ty
             .abi_payload_size()

tests/async_types.rs

@@ -989,6 +989,161 @@ async fn test_stream_build_with_input_output() {
         .unwrap();
 }
 
+#[tokio::test]
+async fn test_stream_write_uint8_array() {
+    let mut instance = TestCase::new()
+        .wit(
+            r#"
+            package test:stream-typed;
+            world stream-typed {
+                export round-trip-u8: async func(input: stream<u8>) -> list<u8>;
+            }
+            "#,
+        )
+        .script(
+            r#"
+            export async function roundTripU8(input) {
+                input.drop();
+                const { readable, writable } = wit.Stream();
+                const readPromise = readable.read(1024);
+                await writable.write(new Uint8Array([97, 98, 99, 0, 255]));
+                writable.drop();
+                const data = await readPromise;
+                readable.drop();
+                return Array.from(data);
+            }
+            "#,
+        )
+        .build_async()
+        .await
+        .unwrap();
+
+    let (inst, store) = instance.parts();
+    let reader = StreamReader::new(&mut *store, ByteProducer::new(vec![])).unwrap();
+    let func = inst
+        .get_typed_func::<(StreamReader<u8>,), (Vec<u8>,)>(&mut *store, "round-trip-u8")
+        .unwrap();
+    let (bytes,) = func.call_async(&mut *store, (reader,)).await.unwrap();
+    assert_eq!(bytes, vec![97, 98, 99, 0, 255]);
+}
+
+#[tokio::test]
+async fn test_stream_write_uint32_array() {
+    // Verify the typed-array fast path handles wider primitive element types
+    // (Uint32Array → stream<u32>): element-count semantics, not byte count.
+    let mut instance = TestCase::new()
+        .wit(
+            r#"
+            package test:stream-typed-u32;
+            world stream-typed-u32 {
+                export round-trip-u32: async func(input: stream<u32>) -> list<u32>;
+            }
+            "#,
+        )
+        .script(
+            r#"
+            export async function roundTripU32(input) {
+                input.drop();
+                const { readable, writable } = wit.Stream();
+                const readPromise = readable.read(1024);
+                await writable.write(new Uint32Array([1, 2, 3, 4294967295]));
+                writable.drop();
+                const data = await readPromise;
+                readable.drop();
+                return Array.from(data);
+            }
+            "#,
+        )
+        .build_async()
+        .await
+        .unwrap();
+
+    let (inst, store) = instance.parts();
+    let reader = StreamReader::new(&mut *store, EmptyProducer::<u32>::new()).unwrap();
+    let func = inst
+        .get_typed_func::<(StreamReader<u32>,), (Vec<u32>,)>(&mut *store, "round-trip-u32")
+        .unwrap();
+    let (values,) = func.call_async(&mut *store, (reader,)).await.unwrap();
+    assert_eq!(values, vec![1, 2, 3, 4_294_967_295]);
+}
+
+#[tokio::test]
+async fn test_stream_write_int32_array() {
+    // Verify the signed flavor (Int32Array → stream<s32>).
+    let mut instance = TestCase::new()
+        .wit(
+            r#"
+            package test:stream-typed-s32;
+            world stream-typed-s32 {
+                export round-trip-s32: async func(input: stream<s32>) -> list<s32>;
+            }
+            "#,
+        )
+        .script(
+            r#"
+            export async function roundTripS32(input) {
+                input.drop();
+                const { readable, writable } = wit.Stream();
+                const readPromise = readable.read(1024);
+                await writable.write(new Int32Array([-2147483648, -1, 0, 1, 2147483647]));
+                writable.drop();
+                const data = await readPromise;
+                readable.drop();
+                return Array.from(data);
+            }
+            "#,
+        )
+        .build_async()
+        .await
+        .unwrap();
+
+    let (inst, store) = instance.parts();
+    let reader = StreamReader::new(&mut *store, EmptyProducer::<i32>::new()).unwrap();
+    let func = inst
+        .get_typed_func::<(StreamReader<i32>,), (Vec<i32>,)>(&mut *store, "round-trip-s32")
+        .unwrap();
+    let (values,) = func.call_async(&mut *store, (reader,)).await.unwrap();
+    assert_eq!(values, vec![-2_147_483_648, -1, 0, 1, 2_147_483_647]);
+}
+
+#[tokio::test]
+async fn test_stream_write_plain_array_still_works() {
+    let mut instance = TestCase::new()
+        .wit(
+            r#"
+            package test:stream-typed;
+            world stream-typed {
+                export round-trip-array: async func(input: stream<u8>) -> list<u8>;
+            }
+            "#,
+        )
+        .script(
+            r#"
+            export async function roundTripArray(input) {
+                input.drop();
+                const { readable, writable } = wit.Stream();
+                const readPromise = readable.read(1024);
+                await writable.write([10, 20, 30]);
+                writable.drop();
+                const data = await readPromise;
+                readable.drop();
+                return Array.from(data);
+            }
+            "#,
+        )
+        .build_async()
+        .await
+        .unwrap();
+
+    let (inst, store) = instance.parts();
+    let reader = StreamReader::new(&mut *store, ByteProducer::new(vec![])).unwrap();
+    let func = inst
+        .get_typed_func::<(StreamReader<u8>,), (Vec<u8>,)>(&mut *store, "round-trip-array")
+        .unwrap();
+    let (bytes,) = func.call_async(&mut *store, (reader,)).await.unwrap();
+    assert_eq!(bytes, vec![10, 20, 30]);
+}
+
 #[tokio::test]
 async fn test_future_create_and_return_u32() {
     let mut instance = TestCase::new()
@@ -1287,6 +1442,31 @@ async fn test_async_variant_mixed_payloads() {
     }
 }
 
+/// A StreamProducer that yields nothing and closes immediately.
+/// Useful for tests that need a closed input stream of any element type.
+struct EmptyProducer<T>(std::marker::PhantomData<T>);
+
+impl<T> EmptyProducer<T> {
+    fn new() -> Self {
+        Self(std::marker::PhantomData)
+    }
+}
+
+impl<T: Send + Sync + 'static> StreamProducer<WasiCtxState> for EmptyProducer<T> {
+    type Item = T;
+    type Buffer = VecBuffer<T>;
+
+    fn poll_produce<'a>(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        _store: StoreContextMut<'a, WasiCtxState>,
+        _destination: Destination<'a, Self::Item, Self::Buffer>,
+        _finish: bool,
+    ) -> Poll<wasmtime::Result<StreamResult>> {
+        Poll::Ready(Ok(StreamResult::Dropped))
+    }
+}
+
 /// A StreamProducer that yields a fixed set of bytes.
 struct ByteProducer {
     data: Vec<u8>,