nodejs/stdlib/
stream_promises.rs

1//! Node `stream/promises` module: the Promise-based `finished` and `pipeline`.
2//!
3//! `require('stream/promises')` exposes promise-returning versions of
4//! `stream.finished` and `stream.pipeline`. Rather than duplicate the listener
5//! bookkeeping already in `stream.rs`, each wraps the callback-based
6//! `require('stream')` free function in a `Promise` — the same compile-a-JS-factory
7//! technique `util.promisify` uses (`crate::compile_completion` + `load_merged` +
8//! `host::run_chunk_on`, then invoke the factory).
9//!
10//! The returned promise is a REAL pending promise: `stream.finished(stream, cb)`
11//! registers listeners and drains its callback on the first terminal event
12//! (`end`/`finish`/`close`) or on `error`, at which point the callback settles the
13//! promise. A stream that has ALREADY reached a terminal state settles immediately
14//! (the callback-based `finished` fires synchronously in that case). No faked
15//! resolution — a stream that never terminates leaves the promise pending, exactly
16//! as Node does.
17
18use fusevm::Value;
19
20/// `stream/promises` module free-functions routed through `stdlib::call`.
21pub const METHODS: &[&str] = &["finished", "pipeline"];
22
23/// True if `name` is a `stream/promises` free function (for the parent's
24/// `is_method` wiring).
25pub fn is_method(name: &str) -> bool {
26    METHODS.contains(&name)
27}
28
29/// `stdlib::call` entry for `stream/promises.<method>`.
30pub fn call(method: &str, args: &[Value]) -> Option<Result<Value, String>> {
31    Some(match method {
32        "finished" => finished(args),
33        "pipeline" => pipeline(args),
34        _ => return None,
35    })
36}
37
38/// Compile a single JS expression and run it on the LIVE host, returning its
39/// completion value. Delegates to the frontend's ONE runtime-source evaluator
40/// (`crate::eval_in_global_scope`), which runs the factory in the program's
41/// module scope rather than in the calling function's frame.
42fn run_completion(src: &str) -> Result<Value, String> {
43    crate::eval_in_global_scope(src)
44}
45
46// `finished(stream[, options])` → a Promise. The incoming args (stream, and an
47// optional options object) are forwarded to the callback-based `stream.finished`
48// with an appended settling callback; the existing impl picks the last callable as
49// its callback and `args[0]` as the stream.
50const FINISHED_SRC: &str = "(function(){\n\
51  var stream = require('stream');\n\
52  return function(){\n\
53    var args = Array.prototype.slice.call(arguments);\n\
54    return new Promise(function(resolve, reject){\n\
55      args.push(function(err){ if (err) reject(err); else resolve(); });\n\
56      stream.finished.apply(stream, args);\n\
57    });\n\
58  };\n\
59})";
60
61// `pipeline(source, ...transforms, destination)` → a Promise that resolves when the
62// chain completes (rejects on error). Forwards to the callback-based
63// `stream.pipeline` with an appended settling callback.
64const PIPELINE_SRC: &str = "(function(){\n\
65  var stream = require('stream');\n\
66  return function(){\n\
67    var args = Array.prototype.slice.call(arguments);\n\
68    return new Promise(function(resolve, reject){\n\
69      args.push(function(err, val){ if (err) reject(err); else resolve(val); });\n\
70      stream.pipeline.apply(stream, args);\n\
71    });\n\
72  };\n\
73})";
74
75/// `stream.promises.finished(stream[, options])` → a Promise settled on the
76/// stream's first terminal event (or rejected on `error`).
77fn finished(args: &[Value]) -> Result<Value, String> {
78    let factory = run_completion(FINISHED_SRC)?;
79    crate::host::invoke(&factory, args.to_vec(), None)
80}
81
82/// `stream.promises.pipeline(...streams)` → a Promise resolved when the piped chain
83/// completes (or rejected on `error`).
84fn pipeline(args: &[Value]) -> Result<Value, String> {
85    let factory = run_completion(PIPELINE_SRC)?;
86    crate::host::invoke(&factory, args.to_vec(), None)
87}
nodejs/stdlib/stream_promises.rs

nodejs/stdlib/
stream_promises.rs
