1use std::cell::Cell;
6use std::collections::VecDeque;
7use std::mem;
8use std::rc::Rc;
9
10use dom_struct::dom_struct;
11use js::context::JSContext;
12use js::conversions::{ConversionResult, FromJSValConvertible};
13use js::jsapi::Heap;
14use js::jsval::{JSVal, UndefinedValue};
15use js::realm::CurrentRealm;
16use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
17use js::typedarray::Uint8Array;
18use script_bindings::cell::DomRefCell;
19use script_bindings::reflector::{Reflector, reflect_dom_object, reflect_dom_object_with_proto};
20
21use super::byteteereadrequest::ByteTeeReadRequest;
22use super::readablebytestreamcontroller::ReadableByteStreamController;
23use crate::dom::bindings::codegen::Bindings::ReadableStreamDefaultReaderBinding::{
24 ReadableStreamDefaultReaderMethods, ReadableStreamReadResult,
25};
26use crate::dom::bindings::conversions::ConversionBehavior;
27use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible};
28use crate::dom::bindings::reflector::DomGlobal;
29use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
30use crate::dom::bindings::trace::RootedTraceableBox;
31use crate::dom::globalscope::GlobalScope;
32use crate::dom::promise::{Promise, RootedPromise, TracedPromise};
33use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
34use crate::dom::readablestream::ReadableStream;
35use crate::dom::stream::defaultteereadrequest::DefaultTeeReadRequest;
36use crate::dom::stream::readablestreamgenericreader::ReadableStreamGenericReader;
37use crate::dom::types::ReadableStreamDefaultController;
38use crate::realms::enter_auto_realm;
39
40type ReadAllBytesSuccessSteps = dyn Fn(&mut js::context::JSContext, Vec<u8>);
41type ReadAllBytesFailureSteps = dyn Fn(&mut js::context::JSContext, SafeHandleValue);
42
43impl js::gc::Rootable for ContinueReadMicrotask {}
44
45#[derive(Clone, JSTraceable, MallocSizeOf)]
51#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
52struct ContinueReadMicrotask {
53 reader: Dom<ReadableStreamDefaultReader>,
54 request: ReadRequest,
55}
56
57impl Callback for ContinueReadMicrotask {
58 fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
59 self.reader.read(cx, &self.request);
62 }
63}
64
65fn read_loop(
67 cx: &mut js::context::JSContext,
68 reader: &ReadableStreamDefaultReader,
69 success_steps: Rc<ReadAllBytesSuccessSteps>,
70 failure_steps: Rc<ReadAllBytesFailureSteps>,
71) {
72 rooted!(&in(cx) let req = ReadRequest::ReadLoop {
77 success_steps,
78 failure_steps,
79 reader: Dom::from_ref(reader),
80 bytes: Rc::new(DomRefCell::new(Vec::new())),
81 });
82 reader.read(cx, &req);
84}
85
86#[derive(Clone, JSTraceable, MallocSizeOf)]
88#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
89pub(crate) enum ReadRequest {
90 Read(TracedPromise),
92 DefaultTee {
94 tee_read_request: Dom<DefaultTeeReadRequest>,
95 },
96 ReadLoop {
99 #[ignore_malloc_size_of = "dyn Fn"]
100 #[no_trace]
101 success_steps: Rc<ReadAllBytesSuccessSteps>,
102 #[ignore_malloc_size_of = "dyn Fn"]
103 #[no_trace]
104 failure_steps: Rc<ReadAllBytesFailureSteps>,
105 reader: Dom<ReadableStreamDefaultReader>,
106 #[conditional_malloc_size_of]
107 bytes: Rc<DomRefCell<Vec<u8>>>,
108 },
109 ByteTee {
110 byte_tee_read_request: Dom<ByteTeeReadRequest>,
111 },
112}
113
114impl js::rust::Rootable for ReadRequest {}
115
116impl ReadRequest {
117 pub(crate) fn chunk_steps(
119 &self,
120 cx: &mut js::context::JSContext,
121 chunk: RootedTraceableBox<Heap<JSVal>>,
122 global: &GlobalScope,
123 ) {
124 match self {
125 ReadRequest::Read(promise) => {
126 promise.resolve_native(
129 cx,
130 &ReadableStreamReadResult {
131 done: Some(false),
132 value: chunk.into_box(),
133 },
134 );
135 },
136 ReadRequest::DefaultTee { tee_read_request } => {
137 tee_read_request.enqueue_chunk_steps(cx, chunk);
138 },
139 ReadRequest::ByteTee {
140 byte_tee_read_request,
141 } => {
142 byte_tee_read_request.enqueue_chunk_steps(cx, global, chunk);
143 },
144 ReadRequest::ReadLoop {
145 success_steps: _,
146 failure_steps,
147 reader,
148 bytes,
149 } => {
150 let global = reader.global();
152
153 auto_root!(&in(cx) let array = match Uint8Array::from_jsval(cx, chunk.handle(), ()) {
155 Ok(ConversionResult::Success(array)) => array,
156 _ => {
157 rooted!(&in(cx) let mut v = UndefinedValue());
158 Error::Type(c"Chunk is not a Uint8Array.".to_owned())
159 .to_jsval(cx, &global, v.handle_mut());
160 (failure_steps)(cx, v.handle());
161 return;
162 },
163 });
164
165 if array.is_shared() {
167 let chunk_bytes = match Vec::<u8>::from_jsval(
168 cx,
169 chunk.handle(),
170 ConversionBehavior::EnforceRange,
171 ) {
172 Ok(ConversionResult::Success(chunk_bytes)) => chunk_bytes,
173 _ => {
174 rooted!(&in(cx) let mut v = UndefinedValue());
175 Error::Type(c"Unknown format for bytes read.".to_owned()).to_jsval(
176 cx,
177 &global,
178 v.handle_mut(),
179 );
180 (failure_steps)(cx, v.handle());
181 return;
182 },
183 };
184 bytes.borrow_mut().extend_from_slice(&chunk_bytes);
185 } else {
186 bytes
187 .borrow_mut()
188 .extend_from_slice(array.as_slice_safe(cx.no_gc()).unwrap_or(&[]));
189 }
190
191 let tick = Promise::new(cx, &global);
195 tick.resolve_native(cx, &());
196
197 let handler = PromiseNativeHandler::new(
198 cx,
199 &global,
200 Some(Box::new(ContinueReadMicrotask {
201 reader: Dom::from_ref(reader),
202 request: self.clone(),
203 })),
204 None,
205 );
206
207 let mut realm = enter_auto_realm(cx, &*global);
208 let cx = &mut realm.current_realm();
209 tick.append_native_handler(cx, &handler);
210 },
211 }
212 }
213
214 pub(crate) fn close_steps(&self, cx: &mut js::context::JSContext) {
216 match self {
217 ReadRequest::Read(promise) => {
218 promise.resolve_native(
221 cx,
222 &ReadableStreamReadResult {
223 done: Some(true),
224 value: Heap::boxed(UndefinedValue()),
225 },
226 );
227 },
228 ReadRequest::DefaultTee { tee_read_request } => {
229 tee_read_request.close_steps(cx);
230 },
231 ReadRequest::ByteTee {
232 byte_tee_read_request,
233 } => {
234 byte_tee_read_request
235 .close_steps(cx)
236 .expect("ByteTeeReadRequest close steps should not fail");
237 },
238 ReadRequest::ReadLoop {
239 success_steps,
240 reader,
241 bytes,
242 ..
243 } => {
244 let bytes = mem::take(&mut *bytes.borrow_mut());
246 (success_steps)(cx, bytes);
247
248 reader
249 .release(cx)
250 .expect("Releasing the read-all-bytes reader should succeed");
251 },
252 }
253 }
254
255 pub(crate) fn error_steps(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) {
257 match self {
258 ReadRequest::Read(promise) => {
259 promise.reject_native(cx, &e)
262 },
263 ReadRequest::DefaultTee { tee_read_request } => {
264 tee_read_request.error_steps();
265 },
266 ReadRequest::ByteTee {
267 byte_tee_read_request,
268 } => {
269 byte_tee_read_request.error_steps();
270 },
271 ReadRequest::ReadLoop {
272 failure_steps,
273 reader,
274 ..
275 } => {
276 (failure_steps)(cx, e);
278
279 reader
280 .release(cx)
281 .expect("Releasing the read-all-bytes reader should succeed");
282 },
283 }
284 }
285}
286
287#[derive(Clone, JSTraceable, MallocSizeOf)]
290#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
291struct ByteTeeClosedPromiseRejectionHandler {
292 branch_1_controller: Dom<ReadableByteStreamController>,
293 branch_2_controller: Dom<ReadableByteStreamController>,
294 #[conditional_malloc_size_of]
295 canceled_1: Rc<Cell<bool>>,
296 #[conditional_malloc_size_of]
297 canceled_2: Rc<Cell<bool>>,
298 cancel_promise: TracedPromise,
299 #[conditional_malloc_size_of]
300 reader_version: Rc<Cell<u64>>,
301 expected_version: u64,
302}
303
304impl Callback for ByteTeeClosedPromiseRejectionHandler {
305 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
308 if self.reader_version.get() != self.expected_version {
310 return;
311 }
312
313 self.branch_1_controller.error(cx, v);
315
316 self.branch_2_controller.error(cx, v);
318
319 if !self.canceled_1.get() || !self.canceled_2.get() {
321 self.cancel_promise.resolve_native(cx, &());
322 }
323 }
324}
325
326#[derive(Clone, JSTraceable, MallocSizeOf)]
329#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
330struct DefaultTeeClosedPromiseRejectionHandler {
331 branch_1_controller: Dom<ReadableStreamDefaultController>,
332 branch_2_controller: Dom<ReadableStreamDefaultController>,
333 #[conditional_malloc_size_of]
334 canceled_1: Rc<Cell<bool>>,
335 #[conditional_malloc_size_of]
336 canceled_2: Rc<Cell<bool>>,
337 cancel_promise: TracedPromise,
338}
339
340impl Callback for DefaultTeeClosedPromiseRejectionHandler {
341 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
344 self.branch_1_controller.error(cx, v);
346 self.branch_2_controller.error(cx, v);
348
349 if !self.canceled_1.get() || !self.canceled_2.get() {
351 self.cancel_promise.resolve_native(cx, &());
352 }
353 }
354}
355
356#[dom_struct]
358pub(crate) struct ReadableStreamDefaultReader {
359 reflector_: Reflector,
360
361 stream: MutNullableDom<ReadableStream>,
363
364 read_requests: DomRefCell<VecDeque<ReadRequest>>,
365
366 closed_promise: DomRefCell<TracedPromise>,
368}
369
370impl ReadableStreamDefaultReader {
371 fn new_with_proto(
372 cx: &mut JSContext,
373 global: &GlobalScope,
374 proto: Option<SafeHandleObject>,
375 ) -> DomRoot<ReadableStreamDefaultReader> {
376 let closed_promise = Promise::new(cx, global);
377 reflect_dom_object_with_proto(
378 cx,
379 Box::new(ReadableStreamDefaultReader::new_inherited(&closed_promise)),
380 global,
381 proto,
382 )
383 }
384
385 fn new_inherited(promise: &RootedPromise) -> ReadableStreamDefaultReader {
386 ReadableStreamDefaultReader {
387 reflector_: Reflector::new(),
388 stream: MutNullableDom::new(None),
389 read_requests: DomRefCell::new(Default::default()),
390 closed_promise: DomRefCell::new(promise.to_traced()),
391 }
392 }
393
394 pub(crate) fn new(
395 cx: &mut JSContext,
396 global: &GlobalScope,
397 ) -> DomRoot<ReadableStreamDefaultReader> {
398 let closed_promise = Promise::new(cx, global);
399 reflect_dom_object(cx, Box::new(Self::new_inherited(&closed_promise)), global)
400 }
401
402 pub(crate) fn set_up(
404 &self,
405 cx: &mut JSContext,
406 stream: &ReadableStream,
407 global: &GlobalScope,
408 ) -> Fallible<()> {
409 if stream.is_locked() {
411 return Err(Error::Type(c"stream is locked".to_owned()));
412 }
413 self.generic_initialize(cx, global, stream);
416
417 self.read_requests.borrow_mut().clear();
419
420 Ok(())
421 }
422
423 pub(crate) fn close(&self, cx: &mut js::context::JSContext) {
425 self.closed_promise.borrow().resolve_native(cx, &());
427 rooted!(&in(cx) let mut read_requests = self.take_read_requests());
430 for request in read_requests.iter() {
433 request.close_steps(cx);
435 }
436 }
437
438 pub(crate) fn add_read_request(&self, read_request: &ReadRequest) {
440 self.read_requests
441 .borrow_mut()
442 .push_back(read_request.clone());
443 }
444
445 pub(crate) fn get_num_read_requests(&self) -> usize {
447 self.read_requests.borrow().len()
448 }
449
450 pub(crate) fn error(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) {
452 self.closed_promise.borrow().reject_native(cx, &e);
454
455 self.closed_promise.borrow().set_promise_is_handled(cx);
457
458 self.error_read_requests(cx, e);
460 }
461
462 pub(crate) fn remove_read_request(&self) -> ReadRequest {
464 self.read_requests
465 .borrow_mut()
466 .pop_front()
467 .expect("Reader must have read request when remove is called into.")
468 }
469
470 pub(crate) fn release(&self, cx: &mut js::context::JSContext) -> Fallible<()> {
472 self.generic_release(cx).expect("Generic release failed");
474 rooted!(&in(cx) let mut error = UndefinedValue());
476 Error::Type(c"Reader is released".to_owned()).to_jsval(
477 cx,
478 &self.global(),
479 error.handle_mut(),
480 );
481
482 self.error_read_requests(cx, error.handle());
484 Ok(())
485 }
486
487 fn take_read_requests(&self) -> VecDeque<ReadRequest> {
488 mem::take(&mut *self.read_requests.borrow_mut())
489 }
490
491 fn error_read_requests(&self, cx: &mut js::context::JSContext, rval: SafeHandleValue) {
493 rooted!(&in(cx) let mut read_requests = self.take_read_requests());
495
496 for request in read_requests.iter() {
498 request.error_steps(cx, rval);
499 }
500 }
501
502 pub(crate) fn read(&self, cx: &mut js::context::JSContext, read_request: &ReadRequest) {
504 assert!(self.stream.get().is_some());
508
509 let stream = self.stream.get().unwrap();
510
511 stream.set_is_disturbed(true);
513 if stream.is_closed() {
515 read_request.close_steps(cx);
516 } else if stream.is_errored() {
517 rooted!(&in(cx) let mut error = UndefinedValue());
520 stream.get_stored_error(error.handle_mut());
521 read_request.error_steps(cx, error.handle());
522 } else {
523 assert!(stream.is_readable());
526 stream.perform_pull_steps(cx, read_request);
528 }
529 }
530
531 #[allow(clippy::too_many_arguments)]
534 pub(crate) fn byte_tee_append_native_handler_to_closed_promise(
535 &self,
536 cx: &mut js::context::JSContext,
537 branch_1: &ReadableStream,
538 branch_2: &ReadableStream,
539 canceled_1: Rc<Cell<bool>>,
540 canceled_2: Rc<Cell<bool>>,
541 cancel_promise: &RootedPromise,
542 reader_version: Rc<Cell<u64>>,
543 expected_version: u64,
544 ) {
545 let branch_1_controller = branch_1.get_byte_controller();
547 let branch_2_controller = branch_2.get_byte_controller();
548
549 let global = self.global();
550 let handler = PromiseNativeHandler::new(
551 cx,
552 &global,
553 None,
554 Some(Box::new(ByteTeeClosedPromiseRejectionHandler {
555 branch_1_controller: Dom::from_ref(&branch_1_controller),
556 branch_2_controller: Dom::from_ref(&branch_2_controller),
557 canceled_1,
558 canceled_2,
559 cancel_promise: cancel_promise.to_traced(),
560 reader_version,
561 expected_version,
562 })),
563 );
564
565 let mut realm = enter_auto_realm(cx, &*global);
566 let cx = &mut realm.current_realm();
567
568 self.closed_promise
569 .borrow()
570 .append_native_handler(cx, &handler);
571 }
572
573 pub(crate) fn default_tee_append_native_handler_to_closed_promise(
575 &self,
576 cx: &mut js::context::JSContext,
577 branch_1: &ReadableStream,
578 branch_2: &ReadableStream,
579 canceled_1: Rc<Cell<bool>>,
580 canceled_2: Rc<Cell<bool>>,
581 cancel_promise: &RootedPromise,
582 ) {
583 let branch_1_controller = branch_1.get_default_controller();
584
585 let branch_2_controller = branch_2.get_default_controller();
586
587 let global = self.global();
588 let handler = PromiseNativeHandler::new(
589 cx,
590 &global,
591 None,
592 Some(Box::new(DefaultTeeClosedPromiseRejectionHandler {
593 branch_1_controller: Dom::from_ref(&branch_1_controller),
594 branch_2_controller: Dom::from_ref(&branch_2_controller),
595 canceled_1,
596 canceled_2,
597 cancel_promise: cancel_promise.to_traced(),
598 })),
599 );
600
601 let mut realm = enter_auto_realm(cx, &*global);
602 let cx = &mut realm.current_realm();
603
604 self.closed_promise
605 .borrow()
606 .append_native_handler(cx, &handler);
607 }
608
609 pub(crate) fn read_all_bytes(
611 &self,
612 cx: &mut js::context::JSContext,
613 success_steps: Rc<ReadAllBytesSuccessSteps>,
614 failure_steps: Rc<ReadAllBytesFailureSteps>,
615 ) {
616 read_loop(cx, self, success_steps, failure_steps);
621 }
622
623 pub(crate) fn process_read_requests(
625 &self,
626 cx: &mut js::context::JSContext,
627 controller: &ReadableByteStreamController,
628 ) -> Fallible<()> {
629 while !self.read_requests.borrow().is_empty() {
631 if controller.get_queue_total_size() == 0.0 {
633 return Ok(());
634 }
635
636 rooted!(&in(cx) let read_request = self.remove_read_request());
639
640 controller
642 .fill_read_request_from_queue(cx, &read_request)
643 .expect("Fill read request from queue failed");
644 }
645 Ok(())
646 }
647}
648
649impl ReadableStreamDefaultReaderMethods<crate::DomTypeHolder> for ReadableStreamDefaultReader {
650 fn Constructor(
652 cx: &mut JSContext,
653 global: &GlobalScope,
654 proto: Option<SafeHandleObject>,
655 stream: &ReadableStream,
656 ) -> Fallible<DomRoot<Self>> {
657 let reader = Self::new_with_proto(cx, global, proto);
658
659 reader.set_up(cx, stream, global)?;
661
662 Ok(reader)
663 }
664
665 fn Read(&self, cx: &mut js::context::JSContext) -> RootedPromise {
667 if self.stream.get().is_none() {
669 rooted!(&in(cx) let mut error = UndefinedValue());
670 Error::Type(c"stream is undefined".to_owned()).to_jsval(
671 cx,
672 &self.global(),
673 error.handle_mut(),
674 );
675 return Promise::new_rejected(cx, &self.global(), error.handle());
676 }
677 let promise = Promise::new(cx, &self.global());
679
680 rooted!(&in(cx) let read_request = ReadRequest::Read(promise.to_traced()));
691
692 self.read(cx, &read_request);
694
695 promise
697 }
698
699 fn ReleaseLock(&self, cx: &mut js::context::JSContext) -> Fallible<()> {
701 if self.stream.get().is_none() {
702 return Ok(());
704 }
705
706 self.release(cx)
708 }
709
710 fn Closed(&self, cx: &JSContext) -> RootedPromise {
712 self.closed(cx)
713 }
714
715 fn Cancel(&self, cx: &mut js::context::JSContext, reason: SafeHandleValue) -> RootedPromise {
717 self.generic_cancel(cx, &self.global(), reason)
718 }
719}
720
721impl ReadableStreamGenericReader for ReadableStreamDefaultReader {
722 fn get_closed_promise(&self, cx: &JSContext) -> RootedPromise {
723 self.closed_promise.borrow().root(cx)
724 }
725
726 fn set_closed_promise(&self, promise: &RootedPromise) {
727 *self.closed_promise.borrow_mut() = promise.to_traced();
728 }
729
730 fn set_stream(&self, stream: Option<&ReadableStream>) {
731 self.stream.set(stream);
732 }
733
734 fn get_stream(&self) -> Option<DomRoot<ReadableStream>> {
735 self.stream.get()
736 }
737
738 fn as_default_reader(&self) -> Option<&ReadableStreamDefaultReader> {
739 Some(self)
740 }
741}