atmos/os_lib/js/builtins/
streams.rs

1//! WHATWG Streams API の JavaScript バインディング
2//!
3//! `ReadableStream`、`WritableStream`、`TransformStream` のコンストラクタおよび
4//! リーダー/ライター、コントローラのネイティブ関数を提供する。
5
6use super::*;
7
8/// `ReadableStream` コンストラクタ (`new ReadableStream(underlyingSource, strategy)`)
9pub(crate) fn readable_stream_ctor(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
10    let source = arg(a, 0);
11    let obj = match this {
12        Value::Object(o) => o,
13        _ => Obj::plain(),
14    };
15
16    {
17        let mut o = obj.borrow_mut();
18        o.props.insert("locked".into(), Value::Bool(false));
19        o.props.insert("_state".into(), Value::str("readable"));
20        o.props.insert("_chunks".into(), Value::Object(Obj::array(alloc::vec![])));
21        o.props.insert("getReader".into(), nv("ReadableStream.getReader", readable_stream_get_reader));
22        o.props.insert("cancel".into(), nv("ReadableStream.cancel", readable_stream_cancel));
23        o.props.insert("pipeTo".into(), nv("ReadableStream.pipeTo", readable_stream_pipe_to));
24        o.props.insert("pipeThrough".into(), nv("ReadableStream.pipeThrough", readable_stream_pipe_through));
25    }
26
27    // underlyingSource.start(controller) があれば実行
28    if let Value::Object(src_obj) = source {
29        if let Some(start_fn) = src_obj.borrow().props.get("start").cloned() {
30            let controller = make_readable_stream_controller(Value::Object(obj.clone()));
31            let _ = it.call_listener(&start_fn, Value::Object(src_obj.clone()), &[controller], "start");
32        }
33    }
34
35    Ok(Value::Object(obj))
36}
37
38/// 文字列データから直接 ReadableStream を生成するヘルパー（Fetch Response.body 等で使用）
39pub(crate) fn make_readable_stream_from_string(text: &str) -> Value {
40    let stream_obj = Obj::plain();
41    {
42        let mut o = stream_obj.borrow_mut();
43        o.props.insert("locked".into(), Value::Bool(false));
44        o.props.insert("_state".into(), Value::str("closed"));
45        let chunks = if !text.is_empty() {
46            alloc::vec![Value::str(text)]
47        } else {
48            alloc::vec![]
49        };
50        o.props.insert("_chunks".into(), Value::Object(Obj::array(chunks)));
51        o.props.insert("getReader".into(), nv("ReadableStream.getReader", readable_stream_get_reader));
52        o.props.insert("cancel".into(), nv("ReadableStream.cancel", readable_stream_cancel));
53        o.props.insert("pipeTo".into(), nv("ReadableStream.pipeTo", readable_stream_pipe_to));
54        o.props.insert("pipeThrough".into(), nv("ReadableStream.pipeThrough", readable_stream_pipe_through));
55    }
56    Value::Object(stream_obj)
57}
58
59/// 生のバイト列を 1 つの Uint8Array チャンクとして流す ReadableStream（fetch の `response.body`）。
60/// 【2026-10-18】従来は本文を文字列にしたチャンクを流しており、`response.body.getReader()` で
61/// バイナリを読む YouTube の player（SABR の UMP 応答）が中身を扱えなかった。
62pub(crate) fn make_readable_stream_from_bytes(it: &mut Interp, bytes: &[u8]) -> Value {
63    let stream = make_readable_stream_from_string("");
64    if let Value::Object(o) = &stream {
65        let chunks = if bytes.is_empty() {
66            alloc::vec![]
67        } else {
68            alloc::vec![super::typed_array::make_uint8array(it, bytes.to_vec())]
69        };
70        o.borrow_mut().props.insert("_chunks".into(), Value::Object(Obj::array(chunks)));
71    }
72    stream
73}
74
75/// `stream.getReader()`
76pub(crate) fn readable_stream_get_reader(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
77    if let Value::Object(stream_obj) = &this {
78        let is_locked = stream_obj.borrow().props.get("locked").map(|v| v.truthy()).unwrap_or(false);
79        if is_locked {
80            return Err(it.type_error("ReadableStream is already locked by another reader"));
81        }
82        stream_obj.borrow_mut().props.insert("locked".into(), Value::Bool(true));
83
84        let reader = Obj::plain();
85        {
86            let mut r = reader.borrow_mut();
87            r.props.insert("_stream".into(), this.clone());
88            r.props.insert("read".into(), nv("ReadableStreamDefaultReader.read", readable_reader_read));
89            r.props.insert("releaseLock".into(), nv("ReadableStreamDefaultReader.releaseLock", readable_reader_release_lock));
90            r.props.insert("cancel".into(), nv("ReadableStreamDefaultReader.cancel", readable_reader_cancel));
91            r.props.insert("closed".into(), Value::Object(make_resolved_promise(Value::Undefined)));
92        }
93        return Ok(Value::Object(reader));
94    }
95    Err(it.type_error("getReader called on non-ReadableStream"))
96}
97
98/// `stream.cancel(reason)`
99pub(crate) fn readable_stream_cancel(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
100    if let Value::Object(stream_obj) = &this {
101        let mut s = stream_obj.borrow_mut();
102        s.props.insert("_state".into(), Value::str("closed"));
103        if let Some(Value::Object(chunks_obj)) = s.props.get("_chunks") {
104            if let ObjKind::Array(ref mut vec) = chunks_obj.borrow_mut().kind {
105                vec.clear();
106            }
107        }
108    }
109    Ok(resolved_promise(it, Value::Undefined))
110}
111
112/// `reader.read()`
113pub(crate) fn readable_reader_read(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
114    if let Value::Object(reader_obj) = &this {
115        let stream_val = reader_obj.borrow().props.get("_stream").cloned();
116        if let Some(Value::Object(stream_obj)) = stream_val {
117            let s = stream_obj.borrow_mut();
118            let state = s.props.get("_state").map(|v| v.to_js_string()).unwrap_or_else(|| "readable".into());
119
120            // チャンクがあれば先頭を取り出す
121            let chunk_opt = if let Some(Value::Object(chunks_obj)) = s.props.get("_chunks") {
122                if let ObjKind::Array(ref mut vec) = chunks_obj.borrow_mut().kind {
123                    if !vec.is_empty() {
124                        Some(vec.remove(0))
125                    } else {
126                        None
127                    }
128                } else {
129                    None
130                }
131            } else {
132                None
133            };
134
135            drop(s);
136
137            let res_obj = Obj::plain();
138            if let Some(chunk) = chunk_opt {
139                res_obj.borrow_mut().props.insert("value".into(), chunk);
140                res_obj.borrow_mut().props.insert("done".into(), Value::Bool(false));
141            } else {
142                let is_closed = state == "closed";
143                res_obj.borrow_mut().props.insert("value".into(), Value::Undefined);
144                res_obj.borrow_mut().props.insert("done".into(), Value::Bool(is_closed));
145            }
146            return Ok(resolved_promise(it, Value::Object(res_obj)));
147        }
148    }
149    let res_obj = Obj::plain();
150    res_obj.borrow_mut().props.insert("value".into(), Value::Undefined);
151    res_obj.borrow_mut().props.insert("done".into(), Value::Bool(true));
152    Ok(resolved_promise(it, Value::Object(res_obj)))
153}
154
155/// `reader.releaseLock()`
156pub(crate) fn readable_reader_release_lock(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
157    if let Value::Object(reader_obj) = &this {
158        let stream_val = reader_obj.borrow().props.get("_stream").cloned();
159        if let Some(Value::Object(stream_obj)) = stream_val {
160            let is_locked = stream_obj.borrow().props.get("locked").map(|v| v.truthy()).unwrap_or(false);
161            if !is_locked {
162                return Err(it.type_error("Cannot release lock: stream is not locked"));
163            }
164            stream_obj.borrow_mut().props.insert("locked".into(), Value::Bool(false));
165            return Ok(Value::Undefined);
166        }
167    }
168    Err(it.type_error("Cannot release lock"))
169}
170
171/// `reader.cancel(reason)`
172pub(crate) fn readable_reader_cancel(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
173    if let Value::Object(reader_obj) = &this {
174        let stream_val = reader_obj.borrow().props.get("_stream").cloned();
175        if let Some(stream) = stream_val {
176            return readable_stream_cancel(it, stream, a);
177        }
178    }
179    Ok(resolved_promise(it, Value::Undefined))
180}
181
182/// `ReadableStreamDefaultController` オブジェクトを構築
183fn make_readable_stream_controller(stream: Value) -> Value {
184    let ctrl = Obj::plain();
185    {
186        let mut c = ctrl.borrow_mut();
187        c.props.insert("_stream".into(), stream);
188        c.props.insert("enqueue".into(), nv("ReadableStreamDefaultController.enqueue", controller_enqueue));
189        c.props.insert("close".into(), nv("ReadableStreamDefaultController.close", controller_close));
190        c.props.insert("error".into(), nv("ReadableStreamDefaultController.error", controller_error));
191        c.props.insert("desiredSize".into(), Value::Number(1.0));
192    }
193    Value::Object(ctrl)
194}
195
196pub(crate) fn controller_enqueue(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
197    if let Value::Object(ctrl_obj) = &this {
198        let stream_val = ctrl_obj.borrow().props.get("_stream").cloned();
199        if let Some(Value::Object(stream_obj)) = stream_val {
200            let s = stream_obj.borrow_mut();
201            let state = s.props.get("_state").map(|v| v.to_js_string()).unwrap_or_else(|| "readable".into());
202            if state != "readable" {
203                return Err(it.type_error("Cannot enqueue a chunk into a closed stream"));
204            }
205            let chunk = arg(a, 0);
206            if let Some(Value::Object(chunks_obj)) = s.props.get("_chunks") {
207                if let ObjKind::Array(ref mut vec) = chunks_obj.borrow_mut().kind {
208                    vec.push(chunk);
209                }
210            }
211            return Ok(Value::Undefined);
212        }
213    }
214    Ok(Value::Undefined)
215}
216
217pub(crate) fn controller_close(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
218    if let Value::Object(ctrl_obj) = &this {
219        let stream_val = ctrl_obj.borrow().props.get("_stream").cloned();
220        if let Some(Value::Object(stream_obj)) = stream_val {
221            let mut s = stream_obj.borrow_mut();
222            let state = s.props.get("_state").map(|v| v.to_js_string()).unwrap_or_else(|| "readable".into());
223            if state == "closed" {
224                return Err(it.type_error("Cannot close an already closed stream"));
225            }
226            s.props.insert("_state".into(), Value::str("closed"));
227            return Ok(Value::Undefined);
228        }
229    }
230    Ok(Value::Undefined)
231}
232
233pub(crate) fn controller_error(_it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
234    if let Value::Object(ctrl_obj) = &this {
235        let stream_val = ctrl_obj.borrow().props.get("_stream").cloned();
236        if let Some(Value::Object(stream_obj)) = stream_val {
237            let mut s = stream_obj.borrow_mut();
238            s.props.insert("_state".into(), Value::str("errored"));
239            let reason = arg(a, 0);
240            s.props.insert("_error".into(), reason);
241            if let Some(Value::Object(chunks_obj)) = s.props.get("_chunks") {
242                if let ObjKind::Array(ref mut vec) = chunks_obj.borrow_mut().kind {
243                    vec.clear();
244                }
245            }
246        }
247    }
248    Ok(Value::Undefined)
249}
250
251/// `WritableStream` コンストラクタ (`new WritableStream(underlyingSink, strategy)`)
252pub(crate) fn writable_stream_ctor(_it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
253    let obj = match this {
254        Value::Object(o) => o,
255        _ => Obj::plain(),
256    };
257
258    {
259        let mut o = obj.borrow_mut();
260        o.props.insert("locked".into(), Value::Bool(false));
261        o.props.insert("_state".into(), Value::str("writable"));
262        o.props.insert("_buffer".into(), Value::Object(Obj::array(alloc::vec![])));
263        o.props.insert("getWriter".into(), nv("WritableStream.getWriter", writable_stream_get_writer));
264        o.props.insert("abort".into(), nv("WritableStream.abort", writable_stream_abort));
265        o.props.insert("close".into(), nv("WritableStream.close", writable_stream_close));
266    }
267
268    Ok(Value::Object(obj))
269}
270
271pub(crate) fn writable_stream_get_writer(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
272    if let Value::Object(stream_obj) = &this {
273        let is_locked = stream_obj.borrow().props.get("locked").map(|v| v.truthy()).unwrap_or(false);
274        if is_locked {
275            return Err(it.type_error("WritableStream is already locked by another writer"));
276        }
277        stream_obj.borrow_mut().props.insert("locked".into(), Value::Bool(true));
278
279        let writer = Obj::plain();
280        {
281            let mut w = writer.borrow_mut();
282            w.props.insert("_stream".into(), this.clone());
283            w.props.insert("write".into(), nv("WritableStreamDefaultWriter.write", writable_writer_write));
284            w.props.insert("close".into(), nv("WritableStreamDefaultWriter.close", writable_writer_close));
285            w.props.insert("abort".into(), nv("WritableStreamDefaultWriter.abort", writable_writer_abort));
286            w.props.insert("releaseLock".into(), nv("WritableStreamDefaultWriter.releaseLock", writable_writer_release_lock));
287        }
288        return Ok(Value::Object(writer));
289    }
290    Err(it.type_error("getWriter called on non-WritableStream"))
291}
292
293pub(crate) fn writable_stream_abort(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
294    if let Value::Object(stream_obj) = &this {
295        let mut s = stream_obj.borrow_mut();
296        s.props.insert("_state".into(), Value::str("errored"));
297        s.props.insert("_error".into(), arg(a, 0));
298    }
299    Ok(resolved_promise(it, Value::Undefined))
300}
301
302pub(crate) fn writable_stream_close(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
303    if let Value::Object(stream_obj) = &this {
304        stream_obj.borrow_mut().props.insert("_state".into(), Value::str("closed"));
305    }
306    Ok(resolved_promise(it, Value::Undefined))
307}
308
309pub(crate) fn writable_writer_write(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
310    if let Value::Object(writer_obj) = &this {
311        let stream_val = writer_obj.borrow().props.get("_stream").cloned();
312        if let Some(Value::Object(stream_obj)) = stream_val {
313            let s = stream_obj.borrow_mut();
314            let state = s.props.get("_state").map(|v| v.to_js_string()).unwrap_or_else(|| "writable".into());
315            if state != "writable" {
316                let err = it.type_error("Cannot write to a closed or errored stream");
317                return Ok(rejected_promise(it, err));
318            }
319            let chunk = arg(a, 0);
320            if let Some(Value::Object(buf_obj)) = s.props.get("_buffer") {
321                if let ObjKind::Array(ref mut vec) = buf_obj.borrow_mut().kind {
322                    vec.push(chunk);
323                }
324            }
325            return Ok(resolved_promise(it, Value::Undefined));
326        }
327    }
328    Ok(resolved_promise(it, Value::Undefined))
329}
330
331pub(crate) fn writable_writer_close(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
332    if let Value::Object(writer_obj) = &this {
333        let stream_val = writer_obj.borrow().props.get("_stream").cloned();
334        if let Some(stream) = stream_val {
335            return writable_stream_close(it, stream, a);
336        }
337    }
338    Ok(resolved_promise(it, Value::Undefined))
339}
340
341pub(crate) fn writable_writer_abort(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
342    if let Value::Object(writer_obj) = &this {
343        let stream_val = writer_obj.borrow().props.get("_stream").cloned();
344        if let Some(stream) = stream_val {
345            return writable_stream_abort(it, stream, a);
346        }
347    }
348    Ok(resolved_promise(it, Value::Undefined))
349}
350
351pub(crate) fn writable_writer_release_lock(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
352    if let Value::Object(writer_obj) = &this {
353        let stream_val = writer_obj.borrow().props.get("_stream").cloned();
354        if let Some(Value::Object(stream_obj)) = stream_val {
355            let is_locked = stream_obj.borrow().props.get("locked").map(|v| v.truthy()).unwrap_or(false);
356            if !is_locked {
357                return Err(it.type_error("Cannot release lock: stream is not locked"));
358            }
359            stream_obj.borrow_mut().props.insert("locked".into(), Value::Bool(false));
360            return Ok(Value::Undefined);
361        }
362    }
363    Err(it.type_error("Cannot release lock"))
364}
365
366/// `TransformStream` コンストラクタ (`new TransformStream()`)
367pub(crate) fn transform_stream_ctor(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
368    let obj = match this {
369        Value::Object(o) => o,
370        _ => Obj::plain(),
371    };
372
373    let readable = make_readable_stream_from_string("");
374    let writable = writable_stream_ctor(it, Value::Undefined, &[])?;
375
376    {
377        let mut o = obj.borrow_mut();
378        o.props.insert("readable".into(), readable);
379        o.props.insert("writable".into(), writable);
380    }
381
382    Ok(Value::Object(obj))
383}
384
385/// `stream.pipeTo(destinationWritable, options?)`
386pub(crate) fn readable_stream_pipe_to(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
387    let dest = arg(a, 0);
388    if let Value::Object(src_obj) = &this {
389        if let Value::Object(dest_obj) = &dest {
390            let src_chunks: Vec<Value> = match src_obj.borrow().props.get("_chunks") {
391                Some(Value::Object(arr)) => match &arr.borrow().kind {
392                    ObjKind::Array(vec) => vec.clone(),
393                    _ => Vec::new(),
394                },
395                _ => Vec::new(),
396            };
397
398            if let Some(Value::Object(dest_buf)) = dest_obj.borrow().props.get("_buffer") {
399                if let ObjKind::Array(ref mut d_vec) = dest_buf.borrow_mut().kind {
400                    for c in src_chunks {
401                        d_vec.push(c);
402                    }
403                }
404            }
405
406            return Ok(resolved_promise(it, Value::Undefined));
407        }
408    }
409    Err(it.type_error("Failed to execute 'pipeTo': Parameter 1 is not of type 'WritableStream'."))
410}
411
412/// `stream.pipeThrough(transformStream, options?)`
413pub(crate) fn readable_stream_pipe_through(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
414    let transform = arg(a, 0);
415    if let Value::Object(t_obj) = &transform {
416        let writable = t_obj.borrow().props.get("writable").cloned();
417        let readable = t_obj.borrow().props.get("readable").cloned();
418        if let (Some(w), Some(r)) = (writable, readable) {
419            let _ = readable_stream_pipe_to(it, this, &[w])?;
420            return Ok(r);
421        }
422    }
423    Err(it.type_error("Failed to execute 'pipeThrough': Invalid transform stream."))
424}
425
426/// `TextEncoderStream` コンストラクタ (`new TextEncoderStream()`)
427pub(crate) fn text_encoder_stream_ctor(it: &mut Interp, this: Value, _a: &[Value]) -> Result<Value, Value> {
428    let obj = match this {
429        Value::Object(o) => o,
430        _ => Obj::plain(),
431    };
432    let readable = make_readable_stream_from_string("");
433    let writable = writable_stream_ctor(it, Value::Undefined, &[])?;
434    {
435        let mut o = obj.borrow_mut();
436        o.props.insert("encoding".into(), Value::str("utf-8"));
437        o.props.insert("readable".into(), readable);
438        o.props.insert("writable".into(), writable);
439    }
440    Ok(Value::Object(obj))
441}
442
443/// `TextDecoderStream` コンストラクタ (`new TextDecoderStream(label, options)`)
444pub(crate) fn text_decoder_stream_ctor(it: &mut Interp, this: Value, a: &[Value]) -> Result<Value, Value> {
445    let label = arg(a, 0).to_js_string();
446    let enc = if label.is_empty() { "utf-8" } else { &label };
447    let obj = match this {
448        Value::Object(o) => o,
449        _ => Obj::plain(),
450    };
451    let readable = make_readable_stream_from_string("");
452    let writable = writable_stream_ctor(it, Value::Undefined, &[])?;
453    {
454        let mut o = obj.borrow_mut();
455        o.props.insert("encoding".into(), Value::str(enc));
456        o.props.insert("fatal".into(), Value::Bool(false));
457        o.props.insert("ignoreBOM".into(), Value::Bool(false));
458        o.props.insert("readable".into(), readable);
459        o.props.insert("writable".into(), writable);
460    }
461    Ok(Value::Object(obj))
462}
atmos/os_lib/js/builtins/streams.rs

atmos/os_lib/js/builtins/
streams.rs
