use crate::test::test_async_with;
use rquickjs::Promise;
fn eval_async<'js>(ctx: &rquickjs::Ctx<'js>, js: &str) -> rquickjs::Result<Promise<'js>> {
ctx.eval(format!("(async () => {{ {js} }})()"))
}
#[tokio::test]
async fn identity_passthrough() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream();
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
writer.write("one");
writer.write("two");
writer.close();
const chunks = [];
while (true) {
const { value, done } = await reader.read();
if (done) break;
chunks.push(value);
}
if (chunks.join(",") !== "one,two") throw new Error("got: " + chunks);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn transform_chunks() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream({
transform(chunk, controller) {
controller.enqueue(chunk.toUpperCase());
}
});
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
writer.write("hello");
writer.write("world");
writer.close();
const chunks = [];
while (true) {
const { value, done } = await reader.read();
if (done) break;
chunks.push(value);
}
if (chunks.join(" ") !== "HELLO WORLD") throw new Error("got: " + chunks);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn one_to_many_expansion() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream({
transform(chunk, controller) {
for (const byte of chunk) {
controller.enqueue(byte);
}
}
});
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
writer.write([1, 2, 3]);
writer.close();
const chunks = [];
while (true) {
const { value, done } = await reader.read();
if (done) break;
chunks.push(value);
}
if (chunks.join(",") !== "1,2,3") throw new Error("got: " + chunks);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn readable_high_water_mark_applies_backpressure() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
ctx.eval::<(), _>(
r#"
globalThis.transformed = [];
globalThis.ts = new TransformStream({
transform(chunk, controller) {
transformed.push(chunk);
controller.enqueue(chunk);
}
}, undefined, { highWaterMark: 3 });
globalThis.writer = ts.writable.getWriter();
[0, 1, 2, 3].forEach(chunk => writer.write(chunk));
"#,
)
.unwrap();
while ctx.execute_pending_job() {}
assert_eq!(
ctx.eval::<String, _>("transformed.join(',')").unwrap(),
"0,1,2"
);
ctx.eval::<(), _>("globalThis.reader = ts.readable.getReader(); reader.read();")
.unwrap();
while ctx.execute_pending_job() {}
assert_eq!(
ctx.eval::<String, _>("transformed.join(',')").unwrap(),
"0,1,2,3"
);
})
})
.await;
}
#[tokio::test]
async fn flush_on_close() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream({
transform(chunk, controller) {
controller.enqueue(chunk);
},
flush(controller) {
controller.enqueue("DONE");
}
});
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
writer.write("a");
writer.close();
const chunks = [];
while (true) {
const { value, done } = await reader.read();
if (done) break;
chunks.push(value);
}
if (chunks.join(",") !== "a,DONE") throw new Error("got: " + chunks);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn pipe_through_chain() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const source = new ReadableStream({
start(controller) {
controller.enqueue("hello");
controller.enqueue("world");
controller.close();
}
});
const upper = new TransformStream({
transform(chunk, c) { c.enqueue(chunk.toUpperCase()); }
});
const exclaim = new TransformStream({
transform(chunk, c) { c.enqueue(chunk + "!"); }
});
const reader = source
.pipeThrough(upper)
.pipeThrough(exclaim)
.getReader();
const chunks = [];
while (true) {
const { value, done } = await reader.read();
if (done) break;
chunks.push(value);
}
if (chunks.join(" ") !== "HELLO! WORLD!") throw new Error("got: " + chunks);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn async_transform() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream({
async transform(chunk, controller) {
await new Promise(r => r());
controller.enqueue(chunk * 2);
}
});
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
writer.write(5);
writer.close();
const { value } = await reader.read();
if (value !== 10) throw new Error("expected 10, got " + value);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn error_propagates_to_reader() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream({
transform(chunk, controller) {
controller.error(new Error("broken"));
}
});
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
writer.write("x").catch(() => {});
try {
await reader.read();
throw new Error("should have thrown");
} catch (e) {
if (e.message !== "broken") throw new Error("wrong error: " + e.message);
}
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn start_receives_controller() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
let controllerRef;
const ts = new TransformStream({
start(controller) {
controllerRef = controller;
controller.enqueue("from-start");
}
});
if (typeof controllerRef.desiredSize !== "number")
throw new Error("controller.desiredSize should be a number");
const reader = ts.readable.getReader();
const { value } = await reader.read();
if (value !== "from-start") throw new Error("got: " + value);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn terminate_closes_readable() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream({
transform(chunk, controller) {
if (chunk === "stop") {
controller.terminate();
return;
}
controller.enqueue(chunk);
}
});
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
writer.write("keep").catch(() => {});
writer.write("stop").catch(() => {});
const { value } = await reader.read();
if (value !== "keep") throw new Error("got: " + value);
const { done } = await reader.read();
if (!done) throw new Error("expected stream to be closed after terminate");
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn illegal_constructor() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
try {
new TransformStreamDefaultController();
throw new Error("should have thrown");
} catch (e) {
if (!(e instanceof TypeError)) throw new Error("expected TypeError, got " + e);
}
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn pipe_to_writable_stream() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const collected = [];
const source = new ReadableStream({
start(c) { c.enqueue(1); c.enqueue(2); c.enqueue(3); c.close(); }
});
const transform = new TransformStream({
transform(chunk, c) { c.enqueue(chunk * 10); }
});
const sink = new WritableStream({
write(chunk) { collected.push(chunk); }
});
await source.pipeThrough(transform).pipeTo(sink);
if (collected.join(",") !== "10,20,30") throw new Error("got: " + collected);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn transform_error_rejects_write_without_leaking() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream(
{ transform() { throw new Error("boom"); } },
undefined,
{ highWaterMark: 1 },
);
const writer = ts.writable.getWriter();
let message = null;
try { await writer.write("x"); } catch (e) { message = e.message; }
if (message !== "boom") throw new Error("got: " + message);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
})
.await;
}
#[tokio::test]
async fn thrown_transform_error_reaches_a_pending_read() {
test_async_with(|ctx| {
crate::stream_web::init(&ctx).unwrap();
Box::pin(async move {
eval_async(
&ctx,
r#"
const ts = new TransformStream({ transform() { throw new Error("boom"); } });
const w = ts.writable.getWriter(); const r = ts.readable.getReader();
w.write("x").catch(() => {});
let message = null;
try { await r.read(); } catch (e) { message = e.message; }
if (message !== "boom") throw new Error("got: " + message);
"#,
)
.unwrap()
.into_future::<()>()
.await
.unwrap();
})
}).await;
}