1use 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#[derive(JSTraceable, MallocSizeOf)]
53#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
54struct TransformBackPressureChangePromiseFulfillment {
55 result_promise: TracedPromise,
57
58 #[ignore_malloc_size_of = "mozjs"]
59 chunk: Box<Heap<JSVal>>,
60
61 writable: Dom<WritableStream>,
63
64 controller: Dom<TransformStreamDefaultController>,
65}
66
67impl Callback for TransformBackPressureChangePromiseFulfillment {
68 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
70 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!(self.writable.is_writable());
82
83 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 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)]
116struct PerformTransformFulfillment {
119 result_promise: TracedPromise,
120}
121
122impl Callback for PerformTransformFulfillment {
123 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
124 self.result_promise.resolve_native(cx, &());
126 }
127}
128
129#[derive(JSTraceable, MallocSizeOf)]
130#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
131struct PerformTransformRejection {
134 result_promise: TracedPromise,
135}
136
137impl Callback for PerformTransformRejection {
138 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
139 self.result_promise.reject(cx, v);
141 }
142}
143
144#[derive(JSTraceable, MallocSizeOf)]
145#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
146struct 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#[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 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
174 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 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 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#[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 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
214 self.readable.get_default_controller().error(cx, v);
216
217 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#[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 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
242 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 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 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 self.stream.unblock_write(cx, global);
265
266 finish_promise.resolve_native(cx, &());
268 }
269 }
270}
271
272impl js::gc::Rootable for SourceCancelPromiseRejection {}
273
274#[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 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
287 let global = &self.writeable.global();
289
290 self.writeable
291 .get_default_controller()
292 .error_if_needed(cx, v, global);
293
294 self.stream.unblock_write(cx, global);
296
297 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#[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 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
319 let finish_promise = self
321 .controller
322 .get_finish_promise(cx)
323 .expect("finish promise is not set");
324
325 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 self.readable.get_default_controller().close(cx);
334
335 finish_promise.resolve_native(cx, &());
337 }
338 }
339}
340
341impl js::gc::Rootable for FlushPromiseRejection {}
342#[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 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
355 self.readable.get_default_controller().error(cx, v);
358
359 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#[derive(Clone, JSTraceable, MallocSizeOf)]
372#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
373pub(crate) enum CrossRealmTransform {
374 Readable(CrossRealmTransformReadable),
376 Writable(CrossRealmTransformWritable),
378}
379
380#[dom_struct]
382pub struct TransformStream {
383 reflector_: Reflector,
384
385 backpressure: Cell<bool>,
387
388 backpressure_change_promise: DomRefCell<Option<TracedPromise>>,
390
391 controller: MutNullableDom<TransformStreamDefaultController>,
393
394 detached: Cell<bool>,
396
397 readable: MutNullableDom<ReadableStream>,
399
400 writable: MutNullableDom<WritableStream>,
402}
403
404impl TransformStream {
405 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 #[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 let writable_high_water_mark = 1.0;
442
443 let writable_size_algorithm = extract_size_algorithm(cx, &Default::default());
445
446 let readable_high_water_mark = 0.0;
448
449 let readable_size_algorithm = extract_size_algorithm(cx, &Default::default());
451
452 let start_promise = Promise::new_resolved(cx, global, ());
459
460 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 let controller = TransformStreamDefaultController::new(cx, global, transformer_type);
475
476 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 #[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 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 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 self.set_backpressure(cx, global, true);
557
558 self.controller.set(None);
560
561 Ok(())
562 }
563
564 pub(crate) fn set_backpressure(
566 &self,
567 cx: &mut JSContext,
568 global: &GlobalScope,
569 backpressure: bool,
570 ) {
571 assert!(self.backpressure.get() != backpressure);
573
574 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 *self.backpressure_change_promise.borrow_mut() = Some(Promise::new(cx, global).to_traced());
583
584 self.backpressure.set(backpressure);
586 }
587
588 fn set_up_transform_stream_default_controller(
590 &self,
591 controller: &TransformStreamDefaultController,
592 ) {
593 assert!(self.controller.get().is_none());
598
599 controller.set_stream(self);
601
602 self.controller.set(Some(controller));
604
605 }
610
611 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 = TransformStreamDefaultController::new(
621 cx,
622 global,
623 TransformerType::new_from_js_transformer(transformer),
624 );
625
626 controller.set_transform_obj(transformer_obj);
649
650 self.set_up_transform_stream_default_controller(&controller);
653 }
654
655 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!(self.writable.get().is_some());
664
665 let controller = self.controller.get().expect("controller is not set");
667
668 if self.backpressure.get() {
670 let backpressure_change_promise = self.backpressure_change_promise.borrow();
672
673 assert!(backpressure_change_promise.is_some());
675
676 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 controller.transform_stream_default_controller_perform_transform(cx, global, chunk)
705 }
706
707 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 = self.controller.get().expect("controller is not set");
716
717 if let Some(finish_promise) = controller.get_finish_promise(cx) {
719 return Ok(finish_promise);
720 }
721
722 let readable = self.readable.get().expect("readable stream is not set");
724
725 controller.set_finish_promise(&Promise::new(cx, global));
727
728 let cancel_promise = controller.perform_cancel(cx, global, reason)?;
730
731 controller.clear_algorithms();
733
734 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 let finish_promise = controller
754 .get_finish_promise(cx)
755 .expect("finish promise is not set");
756 Ok(finish_promise)
757 }
758
759 pub(crate) fn transform_stream_default_sink_close_algorithm(
761 &self,
762 cx: &mut JSContext,
763 global: &GlobalScope,
764 ) -> Fallible<RootedPromise> {
765 let controller = self
767 .controller
768 .get()
769 .ok_or(Error::Type(c"controller is not set".to_owned()))?;
770
771 if let Some(finish_promise) = controller.get_finish_promise(cx) {
773 return Ok(finish_promise);
774 }
775
776 let readable = self
778 .readable
779 .get()
780 .ok_or(Error::Type(c"readable stream is not set".to_owned()))?;
781
782 controller.set_finish_promise(&Promise::new(cx, global));
784
785 let flush_promise = controller.perform_flush(cx, global)?;
787
788 controller.clear_algorithms();
790
791 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 let finish_promise = controller
813 .get_finish_promise(cx)
814 .expect("finish promise is not set");
815 Ok(finish_promise)
816 }
817
818 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 = self
827 .controller
828 .get()
829 .ok_or(Error::Type(c"controller is not set".to_owned()))?;
830
831 if let Some(finish_promise) = controller.get_finish_promise(cx) {
833 return Ok(finish_promise);
834 }
835
836 let writable = self
838 .writable
839 .get()
840 .ok_or(Error::Type(c"writable stream is not set".to_owned()))?;
841
842 controller.set_finish_promise(&Promise::new(cx, global));
844
845 let cancel_promise = controller.perform_cancel(cx, global, reason)?;
847
848 controller.clear_algorithms();
850
851 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 let finish_promise = controller
874 .get_finish_promise(cx)
875 .expect("finish promise is not set");
876 Ok(finish_promise)
877 }
878
879 pub(crate) fn transform_stream_default_source_pull(
881 &self,
882 cx: &mut JSContext,
883 global: &GlobalScope,
884 ) -> Fallible<RootedPromise> {
885 assert!(self.backpressure.get());
887
888 assert!(self.backpressure_change_promise.borrow().is_some());
890
891 self.set_backpressure(cx, global, false);
893
894 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 pub(crate) fn error_writable_and_unblock_write(
905 &self,
906 cx: &mut JSContext,
907 global: &GlobalScope,
908 error: SafeHandleValue,
909 ) {
910 self.get_controller().clear_algorithms();
912
913 self.get_writable()
915 .get_default_controller()
916 .error_if_needed(cx, error, global);
917
918 self.unblock_write(cx, global)
920 }
921
922 pub(crate) fn unblock_write(&self, cx: &mut JSContext, global: &GlobalScope) {
924 if self.backpressure.get() {
926 self.set_backpressure(cx, global, false);
927 }
928 }
929
930 pub(crate) fn error(&self, cx: &mut JSContext, global: &GlobalScope, error: SafeHandleValue) {
932 self.get_readable()
934 .get_default_controller()
935 .error(cx, error);
936
937 self.error_writable_and_unblock_write(cx, global, error);
939 }
940}
941
942impl TransformStreamMethods<crate::DomTypeHolder> for TransformStream {
943 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 rooted!(&in(cx) let transformer_obj = transformer.unwrap_or(ptr::null_mut()));
954
955 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 !transformer_dict.readableType.get().is_undefined() {
974 return Err(Error::Range(c"readableType is set".to_owned()));
975 }
976
977 if !transformer_dict.writableType.get().is_undefined() {
979 return Err(Error::Range(c"writableType is set".to_owned()));
980 }
981
982 let readable_high_water_mark = extract_high_water_mark(readable_strategy, 0.0)?;
984
985 let readable_size_algorithm = extract_size_algorithm(cx, readable_strategy);
987
988 let writable_high_water_mark = extract_high_water_mark(writable_strategy, 1.0)?;
990
991 let writable_size_algorithm = extract_size_algorithm(cx, writable_strategy);
993
994 let start_promise = Promise::new(cx, global);
996
997 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 stream.set_up_transform_stream_default_controller_from_transformer(
1012 cx,
1013 global,
1014 transformer_obj.handle(),
1015 &transformer_dict,
1016 );
1017
1018 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 start_promise.resolve_native(cx, &());
1036 };
1037
1038 Ok(stream)
1039 }
1040
1041 fn Readable(&self) -> DomRoot<ReadableStream> {
1043 self.readable.get().expect("readable stream is not set")
1045 }
1046
1047 fn Writable(&self) -> DomRoot<WritableStream> {
1049 self.writable.get().expect("writable stream is not set")
1051 }
1052}
1053
1054impl Transferable for TransformStream {
1056 type Index = MessagePortIndex;
1057 type Data = TransformStreamData;
1058
1059 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 let readable = self.get_readable();
1068
1069 let writable = self.get_writable();
1071
1072 if readable.is_locked() || writable.is_locked() {
1077 return Err(Error::DataClone(None));
1078 }
1079
1080 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 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 readable
1105 .pipe_to(cx, &global, &proxy_writable, false, false, false, None)
1106 .set_promise_is_handled(cx);
1107
1108 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 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 let proxy_readable = ReadableStream::new_with_proto(cx, owner, None);
1135 proxy_readable.setup_cross_realm_transform_readable(cx, &port2);
1136
1137 let proxy_writable = WritableStream::new_with_proto(cx, owner, None);
1141 proxy_writable.setup_cross_realm_transform_writable(cx, &port1);
1142
1143 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}