chore: add outgoing Readable stream from WASM DataStore.get

Add an implementation for the outgoing StreamReadable from DataStore.get,
which uses wrapped EventEmitters to build the output stream.

Author: enmand

crates/dwn-rs-wasm/src/streams/stream.rs

@@ -3,10 +3,13 @@ use std::{
     task::{Context, Poll},
 };
 
+use futures_util::{pin_mut, StreamExt};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
 use tokio_stream::Stream;
 use wasm_bindgen::prelude::*;
 
+use crate::streams::sys::EventEmitter;
+
 use super::sys::Readable;
 
 #[derive(Clone, Debug)]
@@ -40,6 +43,42 @@ impl StreamReadable {
         &self.readable
     }
 
+    /// from_stream creates a new StreamReadable from a Rust Stream. This function will return a
+    /// new StreamReadable, and the Readable (accessible as as_raw) will stream data to the
+    /// JavaScript stream, as JsValues.
+    pub async fn from_stream<St>(stream: St) -> Self
+    where
+        St: Stream<Item = Result<JsValue, JsValue>>,
+    {
+        // TODO: this is an extremely "hacky" implementation, that uses a legacy trait of
+        // streams in Node, and wraps an EventEmitter, emitting data from the Rust Stream,
+        // using the Readable.wrap, turning it into a proper Node ReadableStream. Once (if)
+        // dwn-sdk-js is on Web Streams, we can remove this (or someone can find a better way).
+        let ee = EventEmitter::new();
+        let readable = Readable::new().wrap(JsCast::unchecked_into::<Readable>(ee.clone()));
+        readable.resume();
+
+        pin_mut!(stream);
+
+        while let Some(item) = stream.next().await {
+            let item = match item {
+                Ok(i) => i,
+                Err(e) => {
+                    if e.is_null() {
+                        ee.emit("end", JsValue::NULL);
+                    } else {
+                        ee.emit("error", e);
+                    }
+                    return Self::new(readable);
+                }
+            };
+
+            ee.emit("data", item.clone());
+        }
+
+        Self::new(readable)
+    }
+
     /// into_stream creates a new Stream from the StreamReadable stream. This function locks the StreamReadable in
     /// JavaScript, and attaches the handlers for data and end events. It then returns a new Stream
     /// from the locked data, and passes the values through unbounded channels.

crates/dwn-rs-wasm/src/streams/sys.rs

@@ -9,6 +9,9 @@ extern "C" {
     /// This is a very simple wrapper on the EventEmitter class in node.js (and provided by the `events module).
     pub type EventEmitter;
 
+    #[wasm_bindgen(constructor)]
+    pub fn new() -> EventEmitter;
+
     #[wasm_bindgen(method, js_name = on)]
     pub fn on(this: &EventEmitter, event: &str, callback: &Function);
 
@@ -19,7 +22,7 @@ extern "C" {
     pub fn off(this: &EventEmitter, event: &str, callback: &Function);
 
     #[wasm_bindgen(method, js_name = emit)]
-    pub fn emit(this: &EventEmitter, event: &str, args: Box<[JsValue]>);
+    pub fn emit(this: &EventEmitter, event: &str, args: JsValue);
 }
 
 // Basic bindings to the node.js Writable stream module.

crates/dwn-rs-wasm/src/surrealdb/data_store.rs

@@ -1,6 +1,5 @@
 use async_stream::stream;
-use core::iter::IntoIterator;
-use futures_util::{stream, StreamExt, TryStreamExt};
+use futures_util::StreamExt;
 use js_sys::{Object, Reflect};
 use thiserror::Error;
 use wasm_bindgen::prelude::*;
@@ -115,18 +114,18 @@ impl SurrealDataStore {
         let size = v.size;
         let reader = stream! {
             while let Some(chunk) = v.data.next().await {
-                yield Ok(wasm_bindgen::JsCast::unchecked_into(js_sys::Uint8Array::from(chunk.as_slice())));
+                yield Ok(js_sys::Uint8Array::from(chunk.as_slice()).into());
             }
+
+            yield Err(JsValue::NULL);
         };
 
         let obj: DataStoreGetResult = JsCast::unchecked_into(Object::new());
         Reflect::set(&obj, &"dataSize".into(), &size.into())?;
         Reflect::set(
             &obj,
             &"dataStream".into(),
-            &Readable::from_web(JsCast::unchecked_into(
-                wasm_streams::ReadableStream::from_stream(reader).into_raw(),
-            )),
+            StreamReadable::from_stream(reader).await.as_raw(),
         )?;
 
         Ok(Some(obj))