Skip to main content

script/dom/stream/
transformstream.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
4
5use std::cell::Cell;
6use std::ptr::{self};
7
8use dom_struct::dom_struct;
9use js::context::JSContext;
10use js::jsapi::{Heap, JSObject};
11use js::jsval::{JSVal, ObjectValue, UndefinedValue};
12use js::realm::CurrentRealm;
13use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
14use rustc_hash::FxHashMap;
15use script_bindings::callback::ExceptionHandling;
16use script_bindings::cell::DomRefCell;
17use script_bindings::reflector::{Reflector, reflect_dom_object_with_proto};
18use script_bindings::structuredclone::StructuredData;
19use servo_base::id::{MessagePortId, MessagePortIndex};
20use servo_constellation_traits::TransformStreamData;
21
22use super::readablestream::CrossRealmTransformReadable;
23use super::writablestream::CrossRealmTransformWritable;
24use crate::dom::bindings::callback::RootedCallback;
25use crate::dom::bindings::codegen::Bindings::QueuingStrategyBinding::{
26    QueuingStrategy, QueuingStrategySize,
27};
28use crate::dom::bindings::codegen::Bindings::TransformStreamBinding::TransformStreamMethods;
29use crate::dom::bindings::codegen::Bindings::TransformerBinding::Transformer;
30use crate::dom::bindings::conversions::ConversionResult;
31use crate::dom::bindings::error::{Error, Fallible};
32use crate::dom::bindings::reflector::DomGlobal;
33use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
34use crate::dom::bindings::transferable::Transferable;
35use crate::dom::globalscope::GlobalScope;
36use crate::dom::messageport::MessagePort;
37use crate::dom::promise::{Promise, RootedPromise, TracedPromise};
38use crate::dom::promisenativehandler::Callback;
39use crate::dom::readablestream::{ReadableStream, create_readable_stream};
40use crate::dom::stream::countqueuingstrategy::{extract_high_water_mark, extract_size_algorithm};
41use crate::dom::stream::transformstreamdefaultcontroller::TransformerType;
42use crate::dom::stream::underlyingsourcecontainer::UnderlyingSourceType;
43use crate::dom::stream::writablestream::create_writable_stream;
44use crate::dom::stream::writablestreamdefaultcontroller::UnderlyingSinkTypeRef;
45use crate::dom::types::{PromiseNativeHandler, TransformStreamDefaultController, WritableStream};
46use crate::realms::enter_auto_realm;
47
48impl js::gc::Rootable for TransformBackPressureChangePromiseFulfillment {}
49
50/// Reacting to backpressureChangePromise as part of
51/// <https://streams.spec.whatwg.org/#transform-stream-default-sink-write-algorithm>
52#[derive(JSTraceable, MallocSizeOf)]
53#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
54struct TransformBackPressureChangePromiseFulfillment {
55    /// The result of reacting to backpressureChangePromise.
56    result_promise: TracedPromise,
57
58    #[ignore_malloc_size_of = "mozjs"]
59    chunk: Box<Heap<JSVal>>,
60
61    /// The writable used in the fulfillment steps
62    writable: Dom<WritableStream>,
63
64    controller: Dom<TransformStreamDefaultController>,
65}
66
67impl Callback for TransformBackPressureChangePromiseFulfillment {
68    /// Reacting to backpressureChangePromise with the following fulfillment steps:
69    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
70        // Let writable be stream.[[writable]].
71        // Let state be writable.[[state]].
72        // If state is "erroring", throw writable.[[storedError]].
73        if self.writable.is_erroring() {
74            rooted!(&in(cx) let mut error = UndefinedValue());
75            self.writable.get_stored_error(error.handle_mut());
76            self.result_promise.reject(cx, error.handle());
77            return;
78        }
79
80        // Assert: state is "writable".
81        assert!(self.writable.is_writable());
82
83        // Return ! TransformStreamDefaultControllerPerformTransform(controller, chunk).
84        rooted!(&in(cx) let mut chunk = UndefinedValue());
85        chunk.set(self.chunk.get());
86        let transform_result = self
87            .controller
88            .transform_stream_default_controller_perform_transform(
89                cx,
90                &self.writable.global(),
91                chunk.handle(),
92            )
93            .expect("perform transform failed");
94
95        // PerformTransformFulfillment and PerformTransformRejection do not need
96        // to be rooted because they only contain an Rc.
97        let handler = PromiseNativeHandler::new(
98            cx,
99            &self.writable.global(),
100            Some(Box::new(PerformTransformFulfillment {
101                result_promise: self.result_promise.clone(),
102            })),
103            Some(Box::new(PerformTransformRejection {
104                result_promise: self.result_promise.clone(),
105            })),
106        );
107
108        let mut realm = enter_auto_realm(cx, &*self.writable.global());
109        let realm = &mut realm.current_realm();
110        transform_result.append_native_handler(realm, &handler);
111    }
112}
113
114#[derive(JSTraceable, MallocSizeOf)]
115#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
116/// Reacting to fulfillment of performTransform as part of
117/// <https://streams.spec.whatwg.org/#transform-stream-default-sink-write-algorithm>
118struct PerformTransformFulfillment {
119    result_promise: TracedPromise,
120}
121
122impl Callback for PerformTransformFulfillment {
123    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
124        // Fulfilled: resolve the outer promise
125        self.result_promise.resolve_native(cx, &());
126    }
127}
128
129#[derive(JSTraceable, MallocSizeOf)]
130#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
131/// Reacting to rejection of performTransform as part of
132/// <https://streams.spec.whatwg.org/#transform-stream-default-sink-write-algorithm>
133struct PerformTransformRejection {
134    result_promise: TracedPromise,
135}
136
137impl Callback for PerformTransformRejection {
138    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
139        // Stream already errored in perform_transform, just reject result_promise
140        self.result_promise.reject(cx, v);
141    }
142}
143
144#[derive(JSTraceable, MallocSizeOf)]
145#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
146/// Reacting to rejection of backpressureChangePromise as part of
147/// <https://streams.spec.whatwg.org/#transform-stream-default-sink-write-algorithm>
148struct BackpressureChangeRejection {
149    result_promise: TracedPromise,
150}
151
152impl Callback for BackpressureChangeRejection {
153    fn callback(&self, cx: &mut CurrentRealm, reason: SafeHandleValue) {
154        self.result_promise.reject(cx, reason);
155    }
156}
157
158impl js::gc::Rootable for CancelPromiseFulfillment {}
159
160/// Reacting to fulfillment of the cancelpromise as part of
161/// <https://streams.spec.whatwg.org/#transform-stream-default-sink-abort-algorithm>
162#[derive(JSTraceable, MallocSizeOf)]
163#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
164struct CancelPromiseFulfillment {
165    readable: Dom<ReadableStream>,
166    controller: Dom<TransformStreamDefaultController>,
167    #[ignore_malloc_size_of = "mozjs"]
168    reason: Box<Heap<JSVal>>,
169}
170
171impl Callback for CancelPromiseFulfillment {
172    /// Reacting to backpressureChangePromise with the following fulfillment steps:
173    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
174        // If readable.[[state]] is "errored", reject controller.[[finishPromise]] with readable.[[storedError]].
175        if self.readable.is_errored() {
176            rooted!(&in(cx) let mut error = UndefinedValue());
177            self.readable.get_stored_error(error.handle_mut());
178            self.controller
179                .get_finish_promise(cx)
180                .expect("finish promise is not set")
181                .reject_native(cx, &error.handle());
182        } else {
183            // Otherwise:
184            // Perform ! ReadableStreamDefaultControllerError(readable.[[controller]], reason).
185            rooted!(&in(cx) let mut reason = UndefinedValue());
186            reason.set(self.reason.get());
187            self.readable
188                .get_default_controller()
189                .error(cx, reason.handle());
190
191            // Resolve controller.[[finishPromise]] with undefined.
192            self.controller
193                .get_finish_promise(cx)
194                .expect("finish promise is not set")
195                .resolve_native(cx, &());
196        }
197    }
198}
199
200impl js::gc::Rootable for CancelPromiseRejection {}
201
202/// Reacting to rejection of cancelpromise as part of
203/// <https://streams.spec.whatwg.org/#transform-stream-default-sink-abort-algorithm>
204#[derive(JSTraceable, MallocSizeOf)]
205#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
206struct CancelPromiseRejection {
207    readable: Dom<ReadableStream>,
208    controller: Dom<TransformStreamDefaultController>,
209}
210
211impl Callback for CancelPromiseRejection {
212    /// Reacting to backpressureChangePromise with the following fulfillment steps:
213    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
214        // Perform ! ReadableStreamDefaultControllerError(readable.[[controller]], r).
215        self.readable.get_default_controller().error(cx, v);
216
217        // Reject controller.[[finishPromise]] with r.
218        self.controller
219            .get_finish_promise(cx)
220            .expect("finish promise is not set")
221            .reject(cx, v);
222    }
223}
224
225impl js::gc::Rootable for SourceCancelPromiseFulfillment {}
226
227/// Reacting to fulfillment of the cancelpromise as part of
228/// <https://streams.spec.whatwg.org/#transform-stream-default-source-cancel>
229#[derive(JSTraceable, MallocSizeOf)]
230#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
231struct SourceCancelPromiseFulfillment {
232    writeable: Dom<WritableStream>,
233    controller: Dom<TransformStreamDefaultController>,
234    stream: Dom<TransformStream>,
235    #[ignore_malloc_size_of = "mozjs"]
236    reason: Box<Heap<JSVal>>,
237}
238
239impl Callback for SourceCancelPromiseFulfillment {
240    /// Reacting to backpressureChangePromise with the following fulfillment steps:
241    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
242        // If cancelPromise was fulfilled, then:
243        let finish_promise = self
244            .controller
245            .get_finish_promise(cx)
246            .expect("finish promise is not set");
247
248        let global = &self.writeable.global();
249        // If writable.[[state]] is "errored", reject controller.[[finishPromise]] with writable.[[storedError]].
250        if self.writeable.is_errored() {
251            rooted!(&in(cx) let mut error = UndefinedValue());
252            self.writeable.get_stored_error(error.handle_mut());
253            finish_promise.reject(cx, error.handle());
254        } else {
255            // Otherwise:
256            // Perform ! WritableStreamDefaultControllerErrorIfNeeded(writable.[[controller]], reason).
257            rooted!(&in(cx) let mut reason = UndefinedValue());
258            reason.set(self.reason.get());
259            self.writeable
260                .get_default_controller()
261                .error_if_needed(cx, reason.handle(), global);
262
263            // Perform ! TransformStreamUnblockWrite(stream).
264            self.stream.unblock_write(cx, global);
265
266            // Resolve controller.[[finishPromise]] with undefined.
267            finish_promise.resolve_native(cx, &());
268        }
269    }
270}
271
272impl js::gc::Rootable for SourceCancelPromiseRejection {}
273
274/// Reacting to rejection of cancelpromise as part of
275/// <https://streams.spec.whatwg.org/#transform-stream-default-source-cancel>
276#[derive(JSTraceable, MallocSizeOf)]
277#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
278struct SourceCancelPromiseRejection {
279    writeable: Dom<WritableStream>,
280    controller: Dom<TransformStreamDefaultController>,
281    stream: Dom<TransformStream>,
282}
283
284impl Callback for SourceCancelPromiseRejection {
285    /// Reacting to backpressureChangePromise with the following fulfillment steps:
286    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
287        // Perform ! WritableStreamDefaultControllerErrorIfNeeded(writable.[[controller]], r).
288        let global = &self.writeable.global();
289
290        self.writeable
291            .get_default_controller()
292            .error_if_needed(cx, v, global);
293
294        // Perform ! TransformStreamUnblockWrite(stream).
295        self.stream.unblock_write(cx, global);
296
297        // Reject controller.[[finishPromise]] with r.
298        self.controller
299            .get_finish_promise(cx)
300            .expect("finish promise is not set")
301            .reject(cx, v);
302    }
303}
304
305impl js::gc::Rootable for FlushPromiseFulfillment {}
306
307/// Reacting to fulfillment of the flushpromise as part of
308/// <https://streams.spec.whatwg.org/#transform-stream-default-sink-close-algorithm>
309#[derive(JSTraceable, MallocSizeOf)]
310#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
311struct FlushPromiseFulfillment {
312    readable: Dom<ReadableStream>,
313    controller: Dom<TransformStreamDefaultController>,
314}
315
316impl Callback for FlushPromiseFulfillment {
317    /// Reacting to flushpromise with the following fulfillment steps:
318    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
319        // If flushPromise was fulfilled, then:
320        let finish_promise = self
321            .controller
322            .get_finish_promise(cx)
323            .expect("finish promise is not set");
324
325        // If readable.[[state]] is "errored", reject controller.[[finishPromise]] with readable.[[storedError]].
326        if self.readable.is_errored() {
327            rooted!(&in(cx) let mut error = UndefinedValue());
328            self.readable.get_stored_error(error.handle_mut());
329            finish_promise.reject(cx, error.handle());
330        } else {
331            // Otherwise:
332            // Perform ! ReadableStreamDefaultControllerClose(readable.[[controller]]).
333            self.readable.get_default_controller().close(cx);
334
335            // Resolve controller.[[finishPromise]] with undefined.
336            finish_promise.resolve_native(cx, &());
337        }
338    }
339}
340
341impl js::gc::Rootable for FlushPromiseRejection {}
342/// Reacting to rejection of flushpromise as part of
343/// <https://streams.spec.whatwg.org/#transform-stream-default-sink-close-algorithm>
344
345#[derive(JSTraceable, MallocSizeOf)]
346#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
347struct FlushPromiseRejection {
348    readable: Dom<ReadableStream>,
349    controller: Dom<TransformStreamDefaultController>,
350}
351
352impl Callback for FlushPromiseRejection {
353    /// Reacting to flushpromise with the following fulfillment steps:
354    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
355        // If flushPromise was rejected with reason r, then:
356        // Perform ! ReadableStreamDefaultControllerError(readable.[[controller]], r).
357        self.readable.get_default_controller().error(cx, v);
358
359        // Reject controller.[[finishPromise]] with r.
360        self.controller
361            .get_finish_promise(cx)
362            .expect("finish promise is not set")
363            .reject(cx, v);
364    }
365}
366
367impl js::gc::Rootable for CrossRealmTransform {}
368
369/// A wrapper to handle `message` and `messageerror` events
370/// for the message port used by the transfered stream.
371#[derive(Clone, JSTraceable, MallocSizeOf)]
372#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
373pub(crate) enum CrossRealmTransform {
374    /// <https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformreadable>
375    Readable(CrossRealmTransformReadable),
376    /// <https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformwritable>
377    Writable(CrossRealmTransformWritable),
378}
379
380/// <https://streams.spec.whatwg.org/#ts-class>
381#[dom_struct]
382pub struct TransformStream {
383    reflector_: Reflector,
384
385    /// <https://streams.spec.whatwg.org/#transformstream-backpressure>
386    backpressure: Cell<bool>,
387
388    /// <https://streams.spec.whatwg.org/#transformstream-backpressurechangepromise>
389    backpressure_change_promise: DomRefCell<Option<TracedPromise>>,
390
391    /// <https://streams.spec.whatwg.org/#transformstream-controller>
392    controller: MutNullableDom<TransformStreamDefaultController>,
393
394    /// <https://streams.spec.whatwg.org/#transformstream-detached>
395    detached: Cell<bool>,
396
397    /// <https://streams.spec.whatwg.org/#transformstream-readable>
398    readable: MutNullableDom<ReadableStream>,
399
400    /// <https://streams.spec.whatwg.org/#transformstream-writable>
401    writable: MutNullableDom<WritableStream>,
402}
403
404impl TransformStream {
405    /// <https://streams.spec.whatwg.org/#initialize-transform-stream>
406    fn new_inherited() -> TransformStream {
407        TransformStream {
408            reflector_: Reflector::new(),
409            backpressure: Default::default(),
410            backpressure_change_promise: DomRefCell::new(None),
411            controller: MutNullableDom::new(None),
412            detached: Cell::new(false),
413            readable: MutNullableDom::new(None),
414            writable: MutNullableDom::new(None),
415        }
416    }
417
418    pub(crate) fn new_with_proto(
419        cx: &mut JSContext,
420        global: &GlobalScope,
421        proto: Option<SafeHandleObject>,
422    ) -> DomRoot<TransformStream> {
423        reflect_dom_object_with_proto(
424            cx,
425            Box::new(TransformStream::new_inherited()),
426            global,
427            proto,
428        )
429    }
430
431    /// Creates and set up the newly created transform stream following
432    /// <https://streams.spec.whatwg.org/#transformstream-set-up>
433    #[cfg_attr(crown, expect(crown::unrooted_must_root))]
434    pub(crate) fn set_up(
435        &self,
436        cx: &mut JSContext,
437        global: &GlobalScope,
438        transformer_type: TransformerType,
439    ) -> Fallible<()> {
440        // Step1. Let writableHighWaterMark be 1.
441        let writable_high_water_mark = 1.0;
442
443        // Step 2. Let writableSizeAlgorithm be an algorithm that returns 1.
444        let writable_size_algorithm = extract_size_algorithm(cx, &Default::default());
445
446        // Step 3. Let readableHighWaterMark be 0.
447        let readable_high_water_mark = 0.0;
448
449        // Step 4. Let readableSizeAlgorithm be an algorithm that returns 1.
450        let readable_size_algorithm = extract_size_algorithm(cx, &Default::default());
451
452        // Step 5. Let transformAlgorithmWrapper be an algorithm that runs these steps given a value chunk:
453        // Step 6. Let flushAlgorithmWrapper be an algorithm that runs these steps:
454        // Step 7. Let cancelAlgorithmWrapper be an algorithm that runs these steps given a value reason:
455        // NOTE: These steps are implemented in `TransformStreamDefaultController::new`
456
457        // Step 8. Let startPromise be a promise resolved with undefined.
458        let start_promise = Promise::new_resolved(cx, global, ());
459
460        // Step 9. Perform ! InitializeTransformStream(stream, startPromise,
461        // writableHighWaterMark, writableSizeAlgorithm, readableHighWaterMark,
462        // readableSizeAlgorithm).
463        self.initialize(
464            cx,
465            global,
466            &start_promise,
467            writable_high_water_mark,
468            writable_size_algorithm,
469            readable_high_water_mark,
470            readable_size_algorithm,
471        )?;
472
473        // Step 10. Let controller be a new TransformStreamDefaultController.
474        let controller = TransformStreamDefaultController::new(cx, global, transformer_type);
475
476        // Step 11. Perform ! SetUpTransformStreamDefaultController(stream,
477        // controller, transformAlgorithmWrapper, flushAlgorithmWrapper,
478        // cancelAlgorithmWrapper).
479        self.set_up_transform_stream_default_controller(&controller);
480
481        Ok(())
482    }
483
484    pub(crate) fn get_controller(&self) -> DomRoot<TransformStreamDefaultController> {
485        self.controller.get().expect("controller is not set")
486    }
487
488    pub(crate) fn get_writable(&self) -> DomRoot<WritableStream> {
489        self.writable.get().expect("writable stream is not set")
490    }
491
492    pub(crate) fn get_readable(&self) -> DomRoot<ReadableStream> {
493        self.readable.get().expect("readable stream is not set")
494    }
495
496    pub(crate) fn get_backpressure(&self) -> bool {
497        self.backpressure.get()
498    }
499
500    /// <https://streams.spec.whatwg.org/#initialize-transform-stream>
501    #[expect(clippy::too_many_arguments)]
502    fn initialize(
503        &self,
504        cx: &mut JSContext,
505        global: &GlobalScope,
506        start_promise: &RootedPromise,
507        writable_high_water_mark: f64,
508        writable_size_algorithm: RootedCallback<QueuingStrategySize>,
509        readable_high_water_mark: f64,
510        readable_size_algorithm: RootedCallback<QueuingStrategySize>,
511    ) -> Fallible<()> {
512        // Let startAlgorithm be an algorithm that returns startPromise.
513        // Let writeAlgorithm be the following steps, taking a chunk argument:
514        //  Return ! TransformStreamDefaultSinkWriteAlgorithm(stream, chunk).
515        // Let abortAlgorithm be the following steps, taking a reason argument:
516        //  Return ! TransformStreamDefaultSinkAbortAlgorithm(stream, reason).
517        // Let closeAlgorithm be the following steps:
518        //  Return ! TransformStreamDefaultSinkCloseAlgorithm(stream).
519        // Set stream.[[writable]] to ! CreateWritableStream(startAlgorithm, writeAlgorithm,
520        // closeAlgorithm, abortAlgorithm, writableHighWaterMark, writableSizeAlgorithm).
521        // Note: Those steps are implemented using UnderlyingSinkType::Transform.
522
523        let writable = create_writable_stream(
524            cx,
525            global,
526            writable_high_water_mark,
527            writable_size_algorithm,
528            UnderlyingSinkTypeRef::Transform(self, start_promise),
529        )?;
530        self.writable.set(Some(&writable));
531
532        // Let pullAlgorithm be the following steps:
533
534        // Return ! TransformStreamDefaultSourcePullAlgorithm(stream).
535
536        // Let cancelAlgorithm be the following steps, taking a reason argument:
537
538        // Return ! TransformStreamDefaultSourceCancelAlgorithm(stream, reason).
539
540        // Set stream.[[readable]] to ! CreateReadableStream(startAlgorithm, pullAlgorithm,
541        // cancelAlgorithm, readableHighWaterMark, readableSizeAlgorithm).
542
543        let readable = create_readable_stream(
544            cx,
545            global,
546            UnderlyingSourceType::Transform(self, start_promise),
547            Some(readable_size_algorithm),
548            Some(readable_high_water_mark),
549        );
550        self.readable.set(Some(&readable));
551
552        // Set stream.[[backpressure]] and stream.[[backpressureChangePromise]] to undefined.
553        // Note: This is done in the constructor.
554
555        // Perform ! TransformStreamSetBackpressure(stream, true).
556        self.set_backpressure(cx, global, true);
557
558        // Set stream.[[controller]] to undefined.
559        self.controller.set(None);
560
561        Ok(())
562    }
563
564    /// <https://streams.spec.whatwg.org/#transform-stream-set-backpressure>
565    pub(crate) fn set_backpressure(
566        &self,
567        cx: &mut JSContext,
568        global: &GlobalScope,
569        backpressure: bool,
570    ) {
571        // Assert: stream.[[backpressure]] is not backpressure.
572        assert!(self.backpressure.get() != backpressure);
573
574        // If stream.[[backpressureChangePromise]] is not undefined, resolve
575        // stream.[[backpressureChangePromise]] with undefined.
576        rooted!(&in(cx) let promise = self.backpressure_change_promise.borrow_mut().take());
577        if let Some(ref promise) = *promise {
578            promise.resolve_native(cx, &());
579        }
580
581        // Set stream.[[backpressureChangePromise]] to a new promise.;
582        *self.backpressure_change_promise.borrow_mut() = Some(Promise::new(cx, global).to_traced());
583
584        // Set stream.[[backpressure]] to backpressure.
585        self.backpressure.set(backpressure);
586    }
587
588    /// <https://streams.spec.whatwg.org/#set-up-transform-stream-default-controller>
589    fn set_up_transform_stream_default_controller(
590        &self,
591        controller: &TransformStreamDefaultController,
592    ) {
593        // Assert: stream implements TransformStream.
594        // Note: this is checked with type.
595
596        // Assert: stream.[[controller]] is undefined.
597        assert!(self.controller.get().is_none());
598
599        // Set controller.[[stream]] to stream.
600        controller.set_stream(self);
601
602        // Set stream.[[controller]] to controller.
603        self.controller.set(Some(controller));
604
605        // Set controller.[[transformAlgorithm]] to transformAlgorithm.
606        // Set controller.[[flushAlgorithm]] to flushAlgorithm.
607        // Set controller.[[cancelAlgorithm]] to cancelAlgorithm.
608        // Note: These are set in the constructor.
609    }
610
611    /// <https://streams.spec.whatwg.org/#set-up-transform-stream-default-controller-from-transformer>
612    fn set_up_transform_stream_default_controller_from_transformer(
613        &self,
614        cx: &mut JSContext,
615        global: &GlobalScope,
616        transformer_obj: SafeHandleObject,
617        transformer: &Transformer,
618    ) {
619        // Let controller be a new TransformStreamDefaultController.
620        let controller = TransformStreamDefaultController::new(
621            cx,
622            global,
623            TransformerType::new_from_js_transformer(transformer),
624        );
625
626        // Let transformAlgorithm be the following steps, taking a chunk argument:
627        // Let result be TransformStreamDefaultControllerEnqueue(controller, chunk).
628        // If result is an abrupt completion, return a promise rejected with result.[[Value]].
629        // Otherwise, return a promise resolved with undefined.
630
631        // Let flushAlgorithm be an algorithm which returns a promise resolved with undefined.
632        // Let cancelAlgorithm be an algorithm which returns a promise resolved with undefined.
633
634        // If transformerDict["transform"] exists, set transformAlgorithm to an algorithm which
635        // takes an argument
636        // chunk and returns the result of invoking transformerDict["transform"] with argument
637        // list « chunk, controller »
638        // and callback this value transformer.
639
640        // If transformerDict["flush"] exists, set flushAlgorithm to an algorithm which returns
641        // the result
642        // of invoking transformerDict["flush"] with argument list « controller » and callback
643        // this value transformer.
644
645        // If transformerDict["cancel"] exists, set cancelAlgorithm to an algorithm which takes an argument
646        // reason and returns the result of invoking transformerDict["cancel"] with argument list « reason »
647        // and callback this value transformer.
648        controller.set_transform_obj(transformer_obj);
649
650        // Perform ! SetUpTransformStreamDefaultController(stream, controller,
651        // transformAlgorithm, flushAlgorithm, cancelAlgorithm).
652        self.set_up_transform_stream_default_controller(&controller);
653    }
654
655    /// <https://streams.spec.whatwg.org/#transform-stream-default-sink-write-algorithm>
656    pub(crate) fn transform_stream_default_sink_write_algorithm(
657        &self,
658        cx: &mut JSContext,
659        global: &GlobalScope,
660        chunk: SafeHandleValue,
661    ) -> Fallible<RootedPromise> {
662        // Assert: stream.[[writable]].[[state]] is "writable".
663        assert!(self.writable.get().is_some());
664
665        // Let controller be stream.[[controller]].
666        let controller = self.controller.get().expect("controller is not set");
667
668        // If stream.[[backpressure]] is true,
669        if self.backpressure.get() {
670            // Let backpressureChangePromise be stream.[[backpressureChangePromise]].
671            let backpressure_change_promise = self.backpressure_change_promise.borrow();
672
673            // Assert: backpressureChangePromise is not undefined.
674            assert!(backpressure_change_promise.is_some());
675
676            // Return the result of reacting to backpressureChangePromise with the following fulfillment steps:
677            let result_promise = Promise::new(cx, global);
678            rooted!(&in(cx) let mut fulfillment_handler = Some(TransformBackPressureChangePromiseFulfillment {
679                controller: Dom::from_ref(&controller),
680                writable: Dom::from_ref(&self.writable.get().expect("writable stream")),
681                chunk: Heap::boxed(chunk.get()),
682                result_promise: result_promise.to_traced(),
683            }));
684
685            let handler = PromiseNativeHandler::new(
686                cx,
687                global,
688                fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
689                Some(Box::new(BackpressureChangeRejection {
690                    result_promise: result_promise.to_traced(),
691                })),
692            );
693            let mut realm = enter_auto_realm(cx, global);
694            let realm = &mut realm.current_realm();
695            backpressure_change_promise
696                .as_ref()
697                .expect("Promise must be some by now.")
698                .append_native_handler(realm, &handler);
699
700            return Ok(result_promise);
701        }
702
703        // Return ! TransformStreamDefaultControllerPerformTransform(controller, chunk).
704        controller.transform_stream_default_controller_perform_transform(cx, global, chunk)
705    }
706
707    /// <https://streams.spec.whatwg.org/#transform-stream-default-sink-abort-algorithm>
708    pub(crate) fn transform_stream_default_sink_abort_algorithm(
709        &self,
710        cx: &mut JSContext,
711        global: &GlobalScope,
712        reason: SafeHandleValue,
713    ) -> Fallible<RootedPromise> {
714        // Let controller be stream.[[controller]].
715        let controller = self.controller.get().expect("controller is not set");
716
717        // If controller.[[finishPromise]] is not undefined, return controller.[[finishPromise]].
718        if let Some(finish_promise) = controller.get_finish_promise(cx) {
719            return Ok(finish_promise);
720        }
721
722        // Let readable be stream.[[readable]].
723        let readable = self.readable.get().expect("readable stream is not set");
724
725        // Let controller.[[finishPromise]] be a new promise.
726        controller.set_finish_promise(&Promise::new(cx, global));
727
728        // Let cancelPromise be the result of performing controller.[[cancelAlgorithm]], passing reason.
729        let cancel_promise = controller.perform_cancel(cx, global, reason)?;
730
731        // Perform ! TransformStreamDefaultControllerClearAlgorithms(controller).
732        controller.clear_algorithms();
733
734        // React to cancelPromise:
735        let handler = PromiseNativeHandler::new(
736            cx,
737            global,
738            Some(Box::new(CancelPromiseFulfillment {
739                readable: Dom::from_ref(&readable),
740                controller: Dom::from_ref(&controller),
741                reason: Heap::boxed(reason.get()),
742            })),
743            Some(Box::new(CancelPromiseRejection {
744                readable: Dom::from_ref(&readable),
745                controller: Dom::from_ref(&controller),
746            })),
747        );
748        let mut realm = enter_auto_realm(cx, global);
749        let cx = &mut realm.current_realm();
750        cancel_promise.append_native_handler(cx, &handler);
751
752        // Return controller.[[finishPromise]].
753        let finish_promise = controller
754            .get_finish_promise(cx)
755            .expect("finish promise is not set");
756        Ok(finish_promise)
757    }
758
759    /// <https://streams.spec.whatwg.org/#transform-stream-default-sink-close-algorithm>
760    pub(crate) fn transform_stream_default_sink_close_algorithm(
761        &self,
762        cx: &mut JSContext,
763        global: &GlobalScope,
764    ) -> Fallible<RootedPromise> {
765        // Let controller be stream.[[controller]].
766        let controller = self
767            .controller
768            .get()
769            .ok_or(Error::Type(c"controller is not set".to_owned()))?;
770
771        // If controller.[[finishPromise]] is not undefined, return controller.[[finishPromise]].
772        if let Some(finish_promise) = controller.get_finish_promise(cx) {
773            return Ok(finish_promise);
774        }
775
776        // Let readable be stream.[[readable]].
777        let readable = self
778            .readable
779            .get()
780            .ok_or(Error::Type(c"readable stream is not set".to_owned()))?;
781
782        // Let controller.[[finishPromise]] be a new promise.
783        controller.set_finish_promise(&Promise::new(cx, global));
784
785        // Let flushPromise be the result of performing controller.[[flushAlgorithm]].
786        let flush_promise = controller.perform_flush(cx, global)?;
787
788        // Perform ! TransformStreamDefaultControllerClearAlgorithms(controller).
789        controller.clear_algorithms();
790
791        // React to flushPromise:
792        let handler = PromiseNativeHandler::new(
793            cx,
794            global,
795            Some(Box::new(FlushPromiseFulfillment {
796                readable: Dom::from_ref(&readable),
797                controller: Dom::from_ref(&controller),
798            })),
799            Some(Box::new(FlushPromiseRejection {
800                readable: Dom::from_ref(&readable),
801                controller: Dom::from_ref(&controller),
802            })),
803        );
804
805        {
806            let mut realm = enter_auto_realm(cx, global);
807            let realm = &mut realm.current_realm();
808            flush_promise.append_native_handler(realm, &handler);
809        }
810
811        // Return controller.[[finishPromise]].
812        let finish_promise = controller
813            .get_finish_promise(cx)
814            .expect("finish promise is not set");
815        Ok(finish_promise)
816    }
817
818    /// <https://streams.spec.whatwg.org/#transform-stream-default-source-cancel>
819    pub(crate) fn transform_stream_default_source_cancel(
820        &self,
821        cx: &mut JSContext,
822        global: &GlobalScope,
823        reason: SafeHandleValue,
824    ) -> Fallible<RootedPromise> {
825        // Let controller be stream.[[controller]].
826        let controller = self
827            .controller
828            .get()
829            .ok_or(Error::Type(c"controller is not set".to_owned()))?;
830
831        // If controller.[[finishPromise]] is not undefined, return controller.[[finishPromise]].
832        if let Some(finish_promise) = controller.get_finish_promise(cx) {
833            return Ok(finish_promise);
834        }
835
836        // Let writable be stream.[[writable]].
837        let writable = self
838            .writable
839            .get()
840            .ok_or(Error::Type(c"writable stream is not set".to_owned()))?;
841
842        // Let controller.[[finishPromise]] be a new promise.
843        controller.set_finish_promise(&Promise::new(cx, global));
844
845        // Let cancelPromise be the result of performing controller.[[cancelAlgorithm]], passing reason.
846        let cancel_promise = controller.perform_cancel(cx, global, reason)?;
847
848        // Perform ! TransformStreamDefaultControllerClearAlgorithms(controller).
849        controller.clear_algorithms();
850
851        // React to cancelPromise:
852        let handler = PromiseNativeHandler::new(
853            cx,
854            global,
855            Some(Box::new(SourceCancelPromiseFulfillment {
856                writeable: Dom::from_ref(&writable),
857                controller: Dom::from_ref(&controller),
858                stream: Dom::from_ref(self),
859                reason: Heap::boxed(reason.get()),
860            })),
861            Some(Box::new(SourceCancelPromiseRejection {
862                writeable: Dom::from_ref(&writable),
863                controller: Dom::from_ref(&controller),
864                stream: Dom::from_ref(self),
865            })),
866        );
867
868        let mut realm = enter_auto_realm(cx, global);
869        let cx = &mut realm.current_realm();
870        cancel_promise.append_native_handler(cx, &handler);
871
872        // Return controller.[[finishPromise]].
873        let finish_promise = controller
874            .get_finish_promise(cx)
875            .expect("finish promise is not set");
876        Ok(finish_promise)
877    }
878
879    /// <https://streams.spec.whatwg.org/#transform-stream-default-source-pull>
880    pub(crate) fn transform_stream_default_source_pull(
881        &self,
882        cx: &mut JSContext,
883        global: &GlobalScope,
884    ) -> Fallible<RootedPromise> {
885        // Assert: stream.[[backpressure]] is true.
886        assert!(self.backpressure.get());
887
888        // Assert: stream.[[backpressureChangePromise]] is not undefined.
889        assert!(self.backpressure_change_promise.borrow().is_some());
890
891        // Perform ! TransformStreamSetBackpressure(stream, false).
892        self.set_backpressure(cx, global, false);
893
894        // Return stream.[[backpressureChangePromise]].
895        Ok(self
896            .backpressure_change_promise
897            .borrow()
898            .as_ref()
899            .expect("Promise must be some by now.")
900            .root(cx))
901    }
902
903    /// <https://streams.spec.whatwg.org/#transform-stream-error-writable-and-unblock-write>
904    pub(crate) fn error_writable_and_unblock_write(
905        &self,
906        cx: &mut JSContext,
907        global: &GlobalScope,
908        error: SafeHandleValue,
909    ) {
910        // Perform ! TransformStreamDefaultControllerClearAlgorithms(stream.[[controller]]).
911        self.get_controller().clear_algorithms();
912
913        // Perform ! WritableStreamDefaultControllerErrorIfNeeded(stream.[[writable]].[[controller]], e).
914        self.get_writable()
915            .get_default_controller()
916            .error_if_needed(cx, error, global);
917
918        // Perform ! TransformStreamUnblockWrite(stream).
919        self.unblock_write(cx, global)
920    }
921
922    /// <https://streams.spec.whatwg.org/#transform-stream-unblock-write>
923    pub(crate) fn unblock_write(&self, cx: &mut JSContext, global: &GlobalScope) {
924        // If stream.[[backpressure]] is true, perform ! TransformStreamSetBackpressure(stream, false).
925        if self.backpressure.get() {
926            self.set_backpressure(cx, global, false);
927        }
928    }
929
930    /// <https://streams.spec.whatwg.org/#transform-stream-error>
931    pub(crate) fn error(&self, cx: &mut JSContext, global: &GlobalScope, error: SafeHandleValue) {
932        // Perform ! ReadableStreamDefaultControllerError(stream.[[readable]].[[controller]], e).
933        self.get_readable()
934            .get_default_controller()
935            .error(cx, error);
936
937        // Perform ! TransformStreamErrorWritableAndUnblockWrite(stream, e).
938        self.error_writable_and_unblock_write(cx, global, error);
939    }
940}
941
942impl TransformStreamMethods<crate::DomTypeHolder> for TransformStream {
943    /// <https://streams.spec.whatwg.org/#ts-constructor>
944    fn Constructor(
945        cx: &mut JSContext,
946        global: &GlobalScope,
947        proto: Option<SafeHandleObject>,
948        transformer: Option<*mut JSObject>,
949        writable_strategy: &QueuingStrategy,
950        readable_strategy: &QueuingStrategy,
951    ) -> Fallible<DomRoot<TransformStream>> {
952        // If transformer is missing, set it to null.
953        rooted!(&in(cx) let transformer_obj = transformer.unwrap_or(ptr::null_mut()));
954
955        // Let underlyingSinkDict be underlyingSink,
956        // converted to an IDL value of type UnderlyingSink.
957        let transformer_dict = if !transformer_obj.is_null() {
958            rooted!(&in(cx) let obj_val = ObjectValue(transformer_obj.get()));
959            match Transformer::new(cx, obj_val.handle()) {
960                Ok(ConversionResult::Success(val)) => val,
961                Ok(ConversionResult::Failure(error)) => {
962                    return Err(Error::Type(error.into_owned()));
963                },
964                _ => {
965                    return Err(Error::JSFailed);
966                },
967            }
968        } else {
969            Transformer::empty()
970        };
971
972        // If transformerDict["readableType"] exists, throw a RangeError exception.
973        if !transformer_dict.readableType.get().is_undefined() {
974            return Err(Error::Range(c"readableType is set".to_owned()));
975        }
976
977        // If transformerDict["writableType"] exists, throw a RangeError exception.
978        if !transformer_dict.writableType.get().is_undefined() {
979            return Err(Error::Range(c"writableType is set".to_owned()));
980        }
981
982        // Let readableHighWaterMark be ? ExtractHighWaterMark(readableStrategy, 0).
983        let readable_high_water_mark = extract_high_water_mark(readable_strategy, 0.0)?;
984
985        // Let readableSizeAlgorithm be ! ExtractSizeAlgorithm(readableStrategy).
986        let readable_size_algorithm = extract_size_algorithm(cx, readable_strategy);
987
988        // Let writableHighWaterMark be ? ExtractHighWaterMark(writableStrategy, 1).
989        let writable_high_water_mark = extract_high_water_mark(writable_strategy, 1.0)?;
990
991        // Let writableSizeAlgorithm be ! ExtractSizeAlgorithm(writableStrategy).
992        let writable_size_algorithm = extract_size_algorithm(cx, writable_strategy);
993
994        // Let startPromise be a new promise.
995        let start_promise = Promise::new(cx, global);
996
997        // Perform ! InitializeTransformStream(this, startPromise, writableHighWaterMark,
998        // writableSizeAlgorithm, readableHighWaterMark, readableSizeAlgorithm).
999        let stream = TransformStream::new_with_proto(cx, global, proto);
1000        stream.initialize(
1001            cx,
1002            global,
1003            &start_promise,
1004            writable_high_water_mark,
1005            writable_size_algorithm,
1006            readable_high_water_mark,
1007            readable_size_algorithm,
1008        )?;
1009
1010        // Perform ? SetUpTransformStreamDefaultControllerFromTransformer(this, transformer, transformerDict).
1011        stream.set_up_transform_stream_default_controller_from_transformer(
1012            cx,
1013            global,
1014            transformer_obj.handle(),
1015            &transformer_dict,
1016        );
1017
1018        // If transformerDict["start"] exists, then resolve startPromise with the
1019        // result of invoking transformerDict["start"]
1020        // with argument list « this.[[controller]] » and callback this value transformer.
1021        if let Some(start) = &transformer_dict.start {
1022            rooted!(&in(cx) let mut result: JSVal);
1023            rooted!(&in(cx) let this_object = transformer_obj.get());
1024            start.Call_(
1025                cx,
1026                &this_object.handle(),
1027                &stream.get_controller(),
1028                result.handle_mut(),
1029                ExceptionHandling::Rethrow,
1030            )?;
1031            let promise = Promise::resolve_or_wrap_promise(cx, result.handle(), global);
1032            start_promise.resolve_native(cx, &promise);
1033        } else {
1034            // Otherwise, resolve startPromise with undefined.
1035            start_promise.resolve_native(cx, &());
1036        };
1037
1038        Ok(stream)
1039    }
1040
1041    /// <https://streams.spec.whatwg.org/#ts-readable>
1042    fn Readable(&self) -> DomRoot<ReadableStream> {
1043        // Return this.[[readable]].
1044        self.readable.get().expect("readable stream is not set")
1045    }
1046
1047    /// <https://streams.spec.whatwg.org/#ts-writable>
1048    fn Writable(&self) -> DomRoot<WritableStream> {
1049        // Return this.[[writable]].
1050        self.writable.get().expect("writable stream is not set")
1051    }
1052}
1053
1054/// <https://streams.spec.whatwg.org/#ts-transfer>
1055impl Transferable for TransformStream {
1056    type Index = MessagePortIndex;
1057    type Data = TransformStreamData;
1058
1059    /// <https://streams.spec.whatwg.org/#ref-for-transfer-steps②>
1060    fn transfer(&self, cx: &mut JSContext) -> Fallible<(MessagePortId, TransformStreamData)> {
1061        let global = self.global();
1062        let mut realm = enter_auto_realm(cx, &*global);
1063        let mut realm = realm.current_realm();
1064        let cx = &mut realm;
1065
1066        // Step 1. Let readable be value.[[readable]].
1067        let readable = self.get_readable();
1068
1069        // Step 2. Let writable be value.[[writable]].
1070        let writable = self.get_writable();
1071
1072        // Step 3. If ! IsReadableStreamLocked(readable) is true, throw a
1073        // "DataCloneError" DOMException.
1074        // Step 4. If ! IsWritableStreamLocked(writable) is true, throw a
1075        // "DataCloneError" DOMException.
1076        if readable.is_locked() || writable.is_locked() {
1077            return Err(Error::DataClone(None));
1078        }
1079
1080        // First port pair (readable → proxy writable)
1081        let port1 = MessagePort::new(cx, &global);
1082        global.track_message_port(&port1, None);
1083        let port1_peer = MessagePort::new(cx, &global);
1084        global.track_message_port(&port1_peer, None);
1085        global.entangle_ports(*port1.message_port_id(), *port1_peer.message_port_id());
1086
1087        let proxy_readable = ReadableStream::new_with_proto(cx, &global, None);
1088        proxy_readable.setup_cross_realm_transform_readable(cx, &port1);
1089        proxy_readable
1090            .pipe_to(cx, &global, &writable, false, false, false, None)
1091            .set_promise_is_handled(cx);
1092
1093        // Second port pair (proxy readable → writable)
1094        let port2 = MessagePort::new(cx, &global);
1095        global.track_message_port(&port2, None);
1096        let port2_peer = MessagePort::new(cx, &global);
1097        global.track_message_port(&port2_peer, None);
1098        global.entangle_ports(*port2.message_port_id(), *port2_peer.message_port_id());
1099
1100        let proxy_writable = WritableStream::new_with_proto(cx, &global, None);
1101        proxy_writable.setup_cross_realm_transform_writable(cx, &port2);
1102
1103        // Pipe readable into the proxy writable (→ port_1)
1104        readable
1105            .pipe_to(cx, &global, &proxy_writable, false, false, false, None)
1106            .set_promise_is_handled(cx);
1107
1108        // Step 5. Set dataHolder.[[readable]] to !
1109        // StructuredSerializeWithTransfer(readable, « readable »).
1110        // Step 6. Set dataHolder.[[writable]] to !
1111        // StructuredSerializeWithTransfer(writable, « writable »).
1112        Ok((
1113            *port1_peer.message_port_id(),
1114            TransformStreamData {
1115                readable: port1_peer.transfer(cx)?,
1116                writable: port2_peer.transfer(cx)?,
1117            },
1118        ))
1119    }
1120
1121    /// <https://streams.spec.whatwg.org/#ref-for-transfer-receiving-steps②>
1122    fn transfer_receive(
1123        cx: &mut JSContext,
1124        owner: &GlobalScope,
1125        _id: MessagePortId,
1126        data: TransformStreamData,
1127    ) -> Result<DomRoot<Self>, ()> {
1128        let port1 = MessagePort::transfer_receive(cx, owner, data.readable.0, data.readable.1)?;
1129        let port2 = MessagePort::transfer_receive(cx, owner, data.writable.0, data.writable.1)?;
1130
1131        // Step 1. Let readableRecord be !
1132        // StructuredDeserializeWithTransfer(dataHolder.[[readable]], the
1133        // current Realm).
1134        let proxy_readable = ReadableStream::new_with_proto(cx, owner, None);
1135        proxy_readable.setup_cross_realm_transform_readable(cx, &port2);
1136
1137        // Step 2. Let writableRecord be !
1138        // StructuredDeserializeWithTransfer(dataHolder.[[writable]], the
1139        // current Realm).
1140        let proxy_writable = WritableStream::new_with_proto(cx, owner, None);
1141        proxy_writable.setup_cross_realm_transform_writable(cx, &port1);
1142
1143        // Step 3. Set value.[[readable]] to readableRecord.[[Deserialized]].
1144        // Step 4. Set value.[[writable]] to writableRecord.[[Deserialized]].
1145        // Step 5. Set value.[[backpressure]],
1146        // value.[[backpressureChangePromise]], and value.[[controller]] to
1147        // undefined.
1148        let stream = TransformStream::new_with_proto(cx, owner, None);
1149        stream.readable.set(Some(&proxy_readable));
1150        stream.writable.set(Some(&proxy_writable));
1151
1152        Ok(stream)
1153    }
1154
1155    fn serialized_storage<'a>(
1156        data: StructuredData<'a, '_>,
1157    ) -> &'a mut Option<FxHashMap<MessagePortId, Self::Data>> {
1158        match data {
1159            StructuredData::Reader(r) => &mut r.transform_streams_port_impls,
1160            StructuredData::Writer(w) => &mut w.transform_streams_port,
1161        }
1162    }
1163}