nodejs/stdlib/
stream_consumers.rs

1//! Node `stream/consumers` module: read an entire stream to a single value.
2//!
3//! Each function (`text`/`json`/`arrayBuffer`/`buffer`/`bytes`/`blob`) returns a
4//! Promise that settles once the stream ends. The reader is a small compiled JS
5//! factory (same re-entrant `compile_completion` → `load_merged` → `run_chunk_on`
6//! path `util.promisify` uses): it attaches `data`/`end`/`error` listeners on the
7//! stream's real EventEmitter surface, accumulates chunks with `Buffer.concat`,
8//! and resolves the Promise on `end` (rejecting on `error`). Building it in JS
9//! means the genuine Promise + emitter machinery does the work — no bespoke
10//! native listener/accumulator.
11//!
12//! Runtime byte container: this engine represents raw bytes as a `Buffer`, so
13//! `arrayBuffer`/`bytes` resolve with a `Buffer` (matching `buffer::blob_call`,
14//! which likewise resolves its `arrayBuffer`/`bytes` accessors with a `Buffer`
15//! rather than a bare `ArrayBuffer`/`Uint8Array`). `blob` resolves a real `Blob`.
16
17use crate::host::with_host;
18use fusevm::Value;
19
20/// The free functions exported by `require('stream/consumers')`.
21pub const METHODS: &[&str] = &["text", "json", "arrayBuffer", "buffer", "bytes", "blob"];
22
23/// The compiled reader factory: `(stream, kind) => Promise<value>`. A `data`
24/// listener collects chunks (string chunks are wrapped to `Buffer`); the `end`
25/// listener concatenates and finalizes per `kind`; `error` rejects.
26const CONSUMER_SRC: &str = "(function (stream, kind) {\n\
27  var B = require('buffer');\n\
28  return new Promise(function (resolve, reject) {\n\
29    var chunks = [];\n\
30    stream.on('data', function (c) {\n\
31      chunks.push(typeof c === 'string' ? B.Buffer.from(c) : c);\n\
32    });\n\
33    stream.on('error', function (e) { reject(e); });\n\
34    stream.on('end', function () {\n\
35      try {\n\
36        var buf = B.Buffer.concat(chunks);\n\
37        if (kind === 'text') return resolve(buf.toString('utf8'));\n\
38        if (kind === 'json') return resolve(JSON.parse(buf.toString('utf8')));\n\
39        if (kind === 'blob') return resolve(new B.Blob([buf]));\n\
40        return resolve(buf);\n\
41      } catch (err) { reject(err); }\n\
42    });\n\
43  });\n\
44})";
45
46/// Compile a single JS expression and run it on the LIVE host, returning its
47/// completion value (mirrors `util`'s promisify factory path).
48fn run_completion(src: &str) -> Result<Value, String> {
49    let prog = crate::compile_completion(src)?;
50    let chunk = crate::load_merged(prog);
51    crate::host::run_chunk_on(chunk)
52}
53
54/// Module free-function dispatch (`consumers.text`, `consumers.json`, …).
55pub fn call(method: &str, args: &[Value]) -> Option<Result<Value, String>> {
56    let kind = match method {
57        "text" | "json" | "arrayBuffer" | "buffer" | "bytes" | "blob" => method,
58        _ => return None,
59    };
60    let stream = args.first().cloned().unwrap_or(Value::Undef);
61    Some(consume(stream, kind))
62}
63
64/// Invoke the reader factory with `stream` and the `kind` selector, returning the
65/// Promise it produces.
66fn consume(stream: Value, kind: &str) -> Result<Value, String> {
67    let factory = run_completion(CONSUMER_SRC)?;
68    let kv = with_host(|h| h.new_str(kind.to_string()));
69    crate::host::invoke(&factory, vec![stream, kv], None)
70}
nodejs/stdlib/stream_consumers.rs

nodejs/stdlib/
stream_consumers.rs
