1use std::cell::{Cell, RefCell};
6use std::ptr;
7use std::rc::Rc;
8
9use dom_struct::dom_struct;
10use js::context::JSContext;
11use js::jsapi::{Heap, JSObject};
12use js::jsval::{JSVal, UndefinedValue};
13use js::realm::CurrentRealm;
14use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
15use script_bindings::reflector::{Reflector, reflect_dom_object_with_cx};
16
17use crate::dom::bindings::callback::ExceptionHandling;
18use crate::dom::bindings::codegen::Bindings::QueuingStrategyBinding::QueuingStrategySize;
19use crate::dom::bindings::codegen::Bindings::UnderlyingSinkBinding::{
20 UnderlyingSinkAbortCallback, UnderlyingSinkCloseCallback, UnderlyingSinkStartCallback,
21 UnderlyingSinkWriteCallback,
22};
23use crate::dom::bindings::codegen::Bindings::WritableStreamDefaultControllerBinding::WritableStreamDefaultControllerMethods;
24use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible};
25use crate::dom::bindings::reflector::DomGlobal;
26use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
27use crate::dom::globalscope::GlobalScope;
28use crate::dom::messageport::MessagePort;
29use crate::dom::promise::{Promise, RootedPromise, TracedPromise};
30use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
31use crate::dom::readablestreamdefaultcontroller::{EnqueuedValue, QueueWithSizes, ValueWithSize};
32use crate::dom::stream::writablestream::WritableStream;
33use crate::dom::types::{AbortController, AbortSignal, TransformStream};
34use crate::realms::enter_auto_realm;
35
36impl js::gc::Rootable for CloseAlgorithmFulfillmentHandler {}
37
38#[derive(Clone, JSTraceable, MallocSizeOf)]
41#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
42struct CloseAlgorithmFulfillmentHandler {
43 stream: Dom<WritableStream>,
44}
45
46impl Callback for CloseAlgorithmFulfillmentHandler {
47 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
48 let stream = self.stream.as_rooted();
49
50 stream.finish_in_flight_close(cx);
52 }
53}
54
55impl js::gc::Rootable for CloseAlgorithmRejectionHandler {}
56
57#[derive(Clone, JSTraceable, MallocSizeOf)]
60#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
61struct CloseAlgorithmRejectionHandler {
62 stream: Dom<WritableStream>,
63}
64
65impl Callback for CloseAlgorithmRejectionHandler {
66 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
67 let stream = self.stream.as_rooted();
68
69 let global = GlobalScope::from_current_realm(cx);
70
71 stream.finish_in_flight_close_with_error(cx, &global, v);
73 }
74}
75
76impl js::gc::Rootable for StartAlgorithmFulfillmentHandler {}
77
78#[derive(Clone, JSTraceable, MallocSizeOf)]
81#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
82struct StartAlgorithmFulfillmentHandler {
83 controller: Dom<WritableStreamDefaultController>,
84}
85
86impl Callback for StartAlgorithmFulfillmentHandler {
87 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
90 let controller = self.controller.as_rooted();
91 let stream = controller
92 .stream
93 .get()
94 .expect("Controller should have a stream.");
95
96 assert!(stream.is_erroring() || stream.is_writable());
98
99 controller.started.set(true);
101
102 let global = GlobalScope::from_current_realm(cx);
103
104 controller.advance_queue_if_needed(cx, &global)
106 }
107}
108
109impl js::gc::Rootable for StartAlgorithmRejectionHandler {}
110
111#[derive(Clone, JSTraceable, MallocSizeOf)]
114#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
115struct StartAlgorithmRejectionHandler {
116 controller: Dom<WritableStreamDefaultController>,
117}
118
119impl Callback for StartAlgorithmRejectionHandler {
120 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
123 let controller = self.controller.as_rooted();
124 let stream = controller
125 .stream
126 .get()
127 .expect("Controller should have a stream.");
128
129 assert!(stream.is_erroring() || stream.is_writable());
131
132 controller.started.set(true);
134
135 let global = GlobalScope::from_current_realm(cx);
136
137 stream.deal_with_rejection(cx, &global, v);
139 }
140}
141
142impl js::gc::Rootable for TransferBackPressurePromiseReaction {}
143
144#[derive(JSTraceable, MallocSizeOf)]
147#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
148struct TransferBackPressurePromiseReaction {
149 result_promise: TracedPromise,
151
152 #[ignore_malloc_size_of = "nested Rc"]
154 backpressure_promise: Rc<RefCell<Option<TracedPromise>>>,
155
156 #[ignore_malloc_size_of = "mozjs"]
158 chunk: Box<Heap<JSVal>>,
159
160 port: Dom<MessagePort>,
162}
163
164impl Callback for TransferBackPressurePromiseReaction {
165 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
167 let global = self.result_promise.global();
168 let promise = Promise::new_rooted(cx, &global);
170 *self.backpressure_promise.borrow_mut() = Some(promise.to_traced());
171
172 rooted!(&in(cx) let mut chunk = UndefinedValue());
174 chunk.set(self.chunk.get());
175 let result = self
176 .port
177 .pack_and_post_message_handling_error(cx, "chunk", chunk.handle());
178
179 if let Err(error) = result {
181 global.disentangle_port(cx, &self.port);
183
184 self.result_promise.reject_error(cx, error);
186 } else {
187 self.result_promise.resolve_native(cx, &());
189 }
190 }
191}
192
193impl js::gc::Rootable for WriteAlgorithmFulfillmentHandler {}
194
195#[derive(Clone, JSTraceable, MallocSizeOf)]
198#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
199struct WriteAlgorithmFulfillmentHandler {
200 controller: Dom<WritableStreamDefaultController>,
201}
202
203impl Callback for WriteAlgorithmFulfillmentHandler {
204 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
205 let controller = self.controller.as_rooted();
206 let stream = controller
207 .stream
208 .get()
209 .expect("Controller should have a stream.");
210
211 stream.finish_in_flight_write(cx);
213
214 assert!(stream.is_erroring() || stream.is_writable());
217
218 rooted!(&in(cx) let mut rval = UndefinedValue());
220 controller.queue.dequeue_value(cx, Some(rval.handle_mut()));
221
222 let global = GlobalScope::from_current_realm(cx);
223
224 if !stream.close_queued_or_in_flight() && stream.is_writable() {
226 let backpressure = controller.get_backpressure();
228
229 stream.update_backpressure(cx, backpressure, &global);
231 }
232
233 controller.advance_queue_if_needed(cx, &global)
235 }
236}
237
238impl js::gc::Rootable for WriteAlgorithmRejectionHandler {}
239
240#[derive(Clone, JSTraceable, MallocSizeOf)]
243#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
244struct WriteAlgorithmRejectionHandler {
245 controller: Dom<WritableStreamDefaultController>,
246}
247
248impl Callback for WriteAlgorithmRejectionHandler {
249 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
250 let controller = self.controller.as_rooted();
251 let stream = controller
252 .stream
253 .get()
254 .expect("Controller should have a stream.");
255
256 if stream.is_writable() {
258 controller.clear_algorithms();
260 }
261
262 let global = GlobalScope::from_current_realm(cx);
263
264 stream.finish_in_flight_write_with_error(cx, &global, v);
266 }
267}
268
269#[derive(JSTraceable, PartialEq)]
271#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
272pub enum UnderlyingSinkType {
273 Js {
275 abort: RefCell<Option<Rc<UnderlyingSinkAbortCallback>>>,
277
278 start: RefCell<Option<Rc<UnderlyingSinkStartCallback>>>,
279
280 close: RefCell<Option<Rc<UnderlyingSinkCloseCallback>>>,
282
283 write: RefCell<Option<Rc<UnderlyingSinkWriteCallback>>>,
285 },
286 Transfer {
289 backpressure_promise: Rc<RefCell<Option<TracedPromise>>>,
290 port: Dom<MessagePort>,
291 },
292 Transform(Dom<TransformStream>, TracedPromise),
294}
295
296impl UnderlyingSinkType {
297 pub(crate) fn new_js(
298 abort: Option<Rc<UnderlyingSinkAbortCallback>>,
299 start: Option<Rc<UnderlyingSinkStartCallback>>,
300 close: Option<Rc<UnderlyingSinkCloseCallback>>,
301 write: Option<Rc<UnderlyingSinkWriteCallback>>,
302 ) -> Self {
303 UnderlyingSinkType::Js {
304 abort: RefCell::new(abort),
305 start: RefCell::new(start),
306 close: RefCell::new(close),
307 write: RefCell::new(write),
308 }
309 }
310}
311
312#[dom_struct]
314pub struct WritableStreamDefaultController {
315 reflector_: Reflector,
316
317 #[ignore_malloc_size_of = "underlying_sink_type"]
320 underlying_sink_type: UnderlyingSinkType,
321
322 #[ignore_malloc_size_of = "mozjs"]
324 underlying_sink_obj: Heap<*mut JSObject>,
325
326 queue: QueueWithSizes,
328
329 started: Cell<bool>,
331
332 strategy_hwm: f64,
334
335 #[ignore_malloc_size_of = "QueuingStrategySize"]
337 strategy_size: RefCell<Option<Rc<QueuingStrategySize>>>,
338
339 stream: MutNullableDom<WritableStream>,
341
342 abort_controller: Dom<AbortController>,
344}
345
346impl WritableStreamDefaultController {
347 #[cfg_attr(crown, expect(crown::unrooted_must_root))]
349 fn new_inherited(
350 cx: &mut JSContext,
351 global: &GlobalScope,
352 underlying_sink_type: UnderlyingSinkType,
353 strategy_hwm: f64,
354 strategy_size: Rc<QueuingStrategySize>,
355 ) -> WritableStreamDefaultController {
356 WritableStreamDefaultController {
357 reflector_: Reflector::new(),
358 underlying_sink_type,
359 queue: Default::default(),
360 stream: Default::default(),
361 underlying_sink_obj: Default::default(),
362 strategy_hwm,
363 strategy_size: RefCell::new(Some(strategy_size)),
364 started: Default::default(),
365 abort_controller: Dom::from_ref(&AbortController::new_with_proto(cx, global, None)),
366 }
367 }
368
369 #[cfg_attr(crown, expect(crown::unrooted_must_root))]
370 pub(crate) fn new(
371 cx: &mut JSContext,
372 global: &GlobalScope,
373 underlying_sink_type: UnderlyingSinkType,
374 strategy_hwm: f64,
375 strategy_size: Rc<QueuingStrategySize>,
376 ) -> DomRoot<WritableStreamDefaultController> {
377 reflect_dom_object_with_cx(
378 Box::new(WritableStreamDefaultController::new_inherited(
379 cx,
380 global,
381 underlying_sink_type,
382 strategy_hwm,
383 strategy_size,
384 )),
385 global,
386 cx,
387 )
388 }
389
390 pub(crate) fn started(&self) -> bool {
391 self.started.get()
392 }
393
394 pub(crate) fn set_underlying_sink_this_object(&self, this_object: SafeHandleObject) {
396 self.underlying_sink_obj.set(*this_object);
397 }
398
399 pub(crate) fn signal_abort(&self, cx: &mut CurrentRealm, reason: SafeHandleValue) {
401 self.abort_controller.signal_abort(cx, reason);
402 }
403
404 fn clear_algorithms(&self) {
406 match &self.underlying_sink_type {
407 UnderlyingSinkType::Js {
408 abort,
409 start: _,
410 close,
411 write,
412 } => {
413 write.borrow_mut().take();
415
416 close.borrow_mut().take();
418
419 abort.borrow_mut().take();
421 },
422 UnderlyingSinkType::Transfer {
423 backpressure_promise,
424 ..
425 } => {
426 backpressure_promise.borrow_mut().take();
427 },
428 UnderlyingSinkType::Transform(_, _) => {
429 return;
430 },
431 }
432
433 self.strategy_size.borrow_mut().take();
435 }
436
437 pub(crate) fn setup(
439 &self,
440 cx: &mut JSContext,
441 global: &GlobalScope,
442 stream: &WritableStream,
443 ) -> Result<(), Error> {
444 stream.assert_no_controller();
449
450 self.stream.set(Some(stream));
452
453 stream.set_default_controller(self);
455
456 let backpressure = self.get_backpressure();
476
477 stream.update_backpressure(cx, backpressure, global);
479
480 let start_promise = self.start_algorithm(cx, global)?;
483
484 let rooted_default_controller = DomRoot::from_ref(self);
485
486 rooted!(&in(cx) let mut fulfillment_handler = Some(StartAlgorithmFulfillmentHandler {
488 controller: Dom::from_ref(&rooted_default_controller),
489 }));
490
491 rooted!(&in(cx) let mut rejection_handler = Some(StartAlgorithmRejectionHandler {
493 controller: Dom::from_ref(&rooted_default_controller),
494 }));
495
496 let handler = PromiseNativeHandler::new(
497 cx,
498 global,
499 fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
500 rejection_handler.take().map(|h| Box::new(h) as Box<_>),
501 );
502 let mut realm = enter_auto_realm(cx, global);
503 let cx = &mut realm.current_realm();
504 start_promise.append_native_handler(cx, &handler);
505
506 Ok(())
507 }
508
509 pub(crate) fn close(&self, cx: &mut JSContext, global: &GlobalScope) {
511 self.queue
513 .enqueue_value_with_size(EnqueuedValue::CloseSentinel)
514 .expect("Enqueuing the close sentinel should not fail.");
515 self.advance_queue_if_needed(cx, global);
517 }
518
519 fn start_algorithm(&self, cx: &mut JSContext, global: &GlobalScope) -> Fallible<RootedPromise> {
520 match &self.underlying_sink_type {
521 UnderlyingSinkType::Js {
522 start,
523 abort: _,
524 close: _,
525 write: _,
526 } => {
527 let algo = start.borrow().clone();
528 let start_promise = if let Some(start) = algo {
529 rooted!(&in(cx) let mut result: JSVal);
530 rooted!(&in(cx) let this_object = self.underlying_sink_obj.get());
531 start.Call_(
532 cx,
533 &this_object.handle(),
534 self,
535 result.handle_mut(),
536 ExceptionHandling::Rethrow,
537 )?;
538 Promise::resolve_or_wrap_promise(cx, result.handle(), global)
539 } else {
540 Promise::new_resolved_rooted(cx, global, ())
542 };
543
544 Ok(start_promise)
545 },
546 UnderlyingSinkType::Transfer { .. } => {
547 Ok(Promise::new_resolved_rooted(cx, global, ()))
549 },
550 UnderlyingSinkType::Transform(_, start_promise) => {
551 Ok(start_promise.root(cx))
553 },
554 }
555 }
556
557 pub(crate) fn abort_steps(
559 &self,
560 cx: &mut JSContext,
561 global: &GlobalScope,
562 reason: SafeHandleValue,
563 ) -> RootedPromise {
564 let result = match &self.underlying_sink_type {
565 UnderlyingSinkType::Js {
566 abort,
567 start: _,
568 close: _,
569 write: _,
570 } => {
571 rooted!(&in(cx) let this_object = self.underlying_sink_obj.get());
572 let algo = abort.borrow().clone();
573 let result = if let Some(algo) = algo {
575 algo.Call_(
576 cx,
577 &this_object.handle(),
578 Some(reason),
579 ExceptionHandling::Rethrow,
580 )
581 } else {
582 Ok(Promise::new_resolved_rooted(cx, global, ()))
583 };
584 result.unwrap_or_else(|e| {
585 let promise = Promise::new_rooted(cx, global);
586 promise.reject_error(cx, e);
587 promise
588 })
589 },
590 UnderlyingSinkType::Transfer { port, .. } => {
591 let result = port.pack_and_post_message_handling_error(cx, "error", reason);
596
597 global.disentangle_port(cx, port);
599
600 let promise = Promise::new_rooted(cx, global);
601
602 if let Err(error) = result {
604 promise.reject_error(cx, error);
605 } else {
606 promise.resolve_native(cx, &());
608 }
609 promise
610 },
611 UnderlyingSinkType::Transform(stream, _) => {
612 stream
614 .transform_stream_default_sink_abort_algorithm(cx, global, reason)
615 .expect("Transform stream default sink abort algorithm should not fail.")
616 },
617 };
618
619 self.clear_algorithms();
621
622 result
623 }
624
625 fn call_write_algorithm(
627 &self,
628 cx: &mut JSContext,
629 chunk: SafeHandleValue,
630 global: &GlobalScope,
631 ) -> RootedPromise {
632 match &self.underlying_sink_type {
633 UnderlyingSinkType::Js {
634 abort: _,
635 start: _,
636 close: _,
637 write,
638 } => {
639 rooted!(&in(cx) let this_object = self.underlying_sink_obj.get());
640 let algo = write.borrow().clone();
641 let result = if let Some(algo) = algo {
642 algo.Call_(
643 cx,
644 &this_object.handle(),
645 chunk,
646 self,
647 ExceptionHandling::Rethrow,
648 )
649 } else {
650 Ok(Promise::new_resolved_rooted(cx, global, ()))
651 };
652 result.unwrap_or_else(|e| {
653 let promise = Promise::new_rooted(cx, global);
654 promise.reject_error(cx, e);
655 promise
656 })
657 },
658 UnderlyingSinkType::Transfer {
659 backpressure_promise,
660 port,
661 } => {
662 if backpressure_promise.borrow().is_none() {
668 let promise = Promise::new_resolved_rooted(cx, global, ());
669 *backpressure_promise.borrow_mut() = Some(promise.to_traced());
670 }
671
672 let result_promise = Promise::new_rooted(cx, global);
674 rooted!(&in(cx) let mut fulfillment_handler = Some(TransferBackPressurePromiseReaction {
675 port: port.clone(),
676 backpressure_promise: backpressure_promise.clone(),
677 chunk: Heap::boxed(chunk.get()),
678 result_promise: result_promise.to_traced(),
679 }));
680 let handler = PromiseNativeHandler::new(
681 cx,
682 global,
683 fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
684 None,
685 );
686 let mut realm = enter_auto_realm(cx, global);
687 let realm = &mut realm.current_realm();
688 backpressure_promise
689 .borrow()
690 .as_ref()
691 .expect("Promise must be some by now.")
692 .append_native_handler(realm, &handler);
693 result_promise
694 },
695 UnderlyingSinkType::Transform(stream, _) => {
696 stream
698 .transform_stream_default_sink_write_algorithm(cx, global, chunk)
699 .expect("Transform stream default sink write algorithm should not fail.")
700 },
701 }
702 }
703
704 fn call_close_algorithm(&self, cx: &mut JSContext, global: &GlobalScope) -> RootedPromise {
706 match &self.underlying_sink_type {
707 UnderlyingSinkType::Js {
708 abort: _,
709 start: _,
710 close,
711 write: _,
712 } => {
713 rooted!(&in(cx) let mut this_object = ptr::null_mut::<JSObject>());
714 this_object.set(self.underlying_sink_obj.get());
715 let algo = close.borrow().clone();
716 let result = if let Some(algo) = algo {
717 algo.Call_(cx, &this_object.handle(), ExceptionHandling::Rethrow)
718 } else {
719 Ok(Promise::new_resolved_rooted(cx, global, ()))
720 };
721 result.unwrap_or_else(|e| {
722 let promise = Promise::new_rooted(cx, global);
723 promise.reject_error(cx, e);
724 promise
725 })
726 },
727 UnderlyingSinkType::Transfer { port, .. } => {
728 rooted!(&in(cx) let mut value = UndefinedValue());
733 port.pack_and_post_message(cx, "close", value.handle())
734 .expect("Sending close should not fail.");
735
736 global.disentangle_port(cx, port);
738
739 Promise::new_resolved_rooted(cx, global, ())
741 },
742 UnderlyingSinkType::Transform(stream, _) => {
743 stream
745 .transform_stream_default_sink_close_algorithm(cx, global)
746 .expect("Transform stream default sink close algorithm should not fail.")
747 },
748 }
749 }
750
751 pub(crate) fn process_close(&self, cx: &mut JSContext, global: &GlobalScope) {
753 let Some(stream) = self.stream.get() else {
755 unreachable!("Controller should have a stream");
756 };
757
758 stream.mark_close_request_in_flight();
760
761 self.queue.dequeue_value(cx, None);
763
764 assert!(self.queue.is_empty());
766
767 let sink_close_promise = self.call_close_algorithm(cx, global);
769
770 self.clear_algorithms();
772
773 rooted!(&in(cx) let mut fulfillment_handler = Some(CloseAlgorithmFulfillmentHandler {
775 stream: Dom::from_ref(&stream),
776 }));
777
778 rooted!(&in(cx) let mut rejection_handler = Some(CloseAlgorithmRejectionHandler {
780 stream: Dom::from_ref(&stream),
781 }));
782
783 let handler = PromiseNativeHandler::new(
785 cx,
786 global,
787 fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
788 rejection_handler.take().map(|h| Box::new(h) as Box<_>),
789 );
790 let mut realm = enter_auto_realm(cx, global);
791 let realm = &mut realm.current_realm();
792 sink_close_promise.append_native_handler(realm, &handler);
793 }
794
795 fn advance_queue_if_needed(&self, cx: &mut JSContext, global: &GlobalScope) {
797 let Some(stream) = self.stream.get() else {
799 unreachable!("Controller should have a stream");
800 };
801
802 if !self.started.get() {
804 return;
805 }
806
807 if stream.has_in_flight_write_request() {
809 return;
810 }
811
812 assert!(!(stream.is_errored() || stream.is_closed()));
816
817 if stream.is_erroring() {
819 stream.finish_erroring(cx, global);
821
822 return;
824 }
825
826 rooted!(&in(cx) let mut value = UndefinedValue());
828 let is_closed = {
829 if self.queue.is_empty() {
831 return;
832 }
833 self.queue.peek_queue_value(cx, value.handle_mut())
834 };
835
836 if is_closed {
837 self.process_close(cx, global);
839 } else {
840 self.process_write(cx, value.handle(), global);
842 };
843 }
844
845 pub(crate) fn perform_error_steps(&self) {
847 self.queue.reset();
849 }
850
851 fn process_write(&self, cx: &mut JSContext, chunk: SafeHandleValue, global: &GlobalScope) {
853 let Some(stream) = self.stream.get() else {
855 unreachable!("Controller should have a stream");
856 };
857
858 stream.mark_first_write_request_in_flight();
860
861 let sink_write_promise = self.call_write_algorithm(cx, chunk, global);
863
864 rooted!(&in(cx) let mut fulfillment_handler = Some(WriteAlgorithmFulfillmentHandler {
866 controller: Dom::from_ref(self),
867 }));
868
869 rooted!(&in(cx) let mut rejection_handler = Some(WriteAlgorithmRejectionHandler {
871 controller: Dom::from_ref(self),
872 }));
873
874 let handler = PromiseNativeHandler::new(
876 cx,
877 global,
878 fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
879 rejection_handler.take().map(|h| Box::new(h) as Box<_>),
880 );
881 let mut realm = enter_auto_realm(cx, global);
882 let realm = &mut realm.current_realm();
883 sink_write_promise.append_native_handler(realm, &handler);
884 }
885
886 pub(crate) fn get_desired_size(&self) -> f64 {
888 let desired_size = self.strategy_hwm - self.queue.total_size.get().clamp(0.0, f64::MAX);
890 desired_size.clamp(desired_size, self.strategy_hwm)
891 }
892
893 fn get_backpressure(&self) -> bool {
895 let desired_size = self.get_desired_size();
897
898 desired_size == 0.0 || desired_size.is_sign_negative()
900 }
901
902 pub(crate) fn get_chunk_size(
904 &self,
905 cx: &mut JSContext,
906 global: &GlobalScope,
907 chunk: SafeHandleValue,
908 ) -> f64 {
909 let Some(strategy_size) = self.strategy_size.borrow().clone() else {
911 let Some(stream) = self.stream.get() else {
913 unreachable!("Controller should have a stream");
914 };
915 assert!(!stream.is_writable());
916
917 return 1.0;
919 };
920
921 let result = strategy_size.Call__(cx, chunk, ExceptionHandling::Rethrow);
924
925 match result {
926 Ok(size) => size,
928 Err(error) => {
929 rooted!(&in(cx) let mut rooted_error = UndefinedValue());
934 error.to_jsval(cx, global, rooted_error.handle_mut());
935 self.error_if_needed(cx, rooted_error.handle(), global);
936
937 1.0
939 },
940 }
941 }
942
943 pub(crate) fn write(
945 &self,
946 cx: &mut JSContext,
947 global: &GlobalScope,
948 chunk: SafeHandleValue,
949 chunk_size: f64,
950 ) {
951 let enqueue_result = self
953 .queue
954 .enqueue_value_with_size(EnqueuedValue::Js(ValueWithSize {
955 value: Heap::boxed(chunk.get()),
956 size: chunk_size,
957 }));
958
959 if let Err(error) = enqueue_result {
961 rooted!(&in(cx) let mut rooted_error = UndefinedValue());
964 error.to_jsval(cx, global, rooted_error.handle_mut());
965 self.error_if_needed(cx, rooted_error.handle(), global);
966
967 return;
969 }
970
971 let Some(stream) = self.stream.get() else {
973 unreachable!("Controller should have a stream");
974 };
975
976 if !stream.close_queued_or_in_flight() && stream.is_writable() {
978 let backpressure = self.get_backpressure();
980
981 stream.update_backpressure(cx, backpressure, global);
983 }
984
985 self.advance_queue_if_needed(cx, global);
987 }
988
989 pub(crate) fn error_if_needed(
991 &self,
992 cx: &mut JSContext,
993 error: SafeHandleValue,
994 global: &GlobalScope,
995 ) {
996 let Some(stream) = self.stream.get() else {
998 unreachable!("Controller should have a stream");
999 };
1000
1001 if stream.is_writable() {
1003 self.error(cx, &stream, error, global);
1005 }
1006 }
1007
1008 fn error(
1010 &self,
1011 cx: &mut JSContext,
1012 stream: &WritableStream,
1013 e: SafeHandleValue,
1014 global: &GlobalScope,
1015 ) {
1016 assert!(stream.is_writable());
1021
1022 self.clear_algorithms();
1024
1025 stream.start_erroring(cx, global, e);
1027 }
1028}
1029
1030impl WritableStreamDefaultControllerMethods<crate::DomTypeHolder>
1031 for WritableStreamDefaultController
1032{
1033 fn Error(&self, cx: &mut CurrentRealm, e: SafeHandleValue) {
1035 let Some(stream) = self.stream.get() else {
1037 unreachable!("Controller should have a stream");
1038 };
1039
1040 if !stream.is_writable() {
1042 return;
1043 }
1044
1045 let global = GlobalScope::from_current_realm(cx);
1046
1047 self.error(cx, &stream, e, &global);
1049 }
1050
1051 fn Signal(&self) -> DomRoot<AbortSignal> {
1053 self.abort_controller.signal()
1055 }
1056}