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, &[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 (success_steps)(cx, &bytes.borrow());
246
247 reader
248 .release(cx)
249 .expect("Releasing the read-all-bytes reader should succeed");
250 },
251 }
252 }
253
254 pub(crate) fn error_steps(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) {
256 match self {
257 ReadRequest::Read(promise) => {
258 promise.reject_native(cx, &e)
261 },
262 ReadRequest::DefaultTee { tee_read_request } => {
263 tee_read_request.error_steps();
264 },
265 ReadRequest::ByteTee {
266 byte_tee_read_request,
267 } => {
268 byte_tee_read_request.error_steps();
269 },
270 ReadRequest::ReadLoop {
271 failure_steps,
272 reader,
273 ..
274 } => {
275 (failure_steps)(cx, e);
277
278 reader
279 .release(cx)
280 .expect("Releasing the read-all-bytes reader should succeed");
281 },
282 }
283 }
284}
285
286#[derive(Clone, JSTraceable, MallocSizeOf)]
289#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
290struct ByteTeeClosedPromiseRejectionHandler {
291 branch_1_controller: Dom<ReadableByteStreamController>,
292 branch_2_controller: Dom<ReadableByteStreamController>,
293 #[conditional_malloc_size_of]
294 canceled_1: Rc<Cell<bool>>,
295 #[conditional_malloc_size_of]
296 canceled_2: Rc<Cell<bool>>,
297 cancel_promise: TracedPromise,
298 #[conditional_malloc_size_of]
299 reader_version: Rc<Cell<u64>>,
300 expected_version: u64,
301}
302
303impl Callback for ByteTeeClosedPromiseRejectionHandler {
304 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
307 if self.reader_version.get() != self.expected_version {
309 return;
310 }
311
312 self.branch_1_controller.error(cx, v);
314
315 self.branch_2_controller.error(cx, v);
317
318 if !self.canceled_1.get() || !self.canceled_2.get() {
320 self.cancel_promise.resolve_native(cx, &());
321 }
322 }
323}
324
325#[derive(Clone, JSTraceable, MallocSizeOf)]
328#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
329struct DefaultTeeClosedPromiseRejectionHandler {
330 branch_1_controller: Dom<ReadableStreamDefaultController>,
331 branch_2_controller: Dom<ReadableStreamDefaultController>,
332 #[conditional_malloc_size_of]
333 canceled_1: Rc<Cell<bool>>,
334 #[conditional_malloc_size_of]
335 canceled_2: Rc<Cell<bool>>,
336 cancel_promise: TracedPromise,
337}
338
339impl Callback for DefaultTeeClosedPromiseRejectionHandler {
340 fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
343 self.branch_1_controller.error(cx, v);
345 self.branch_2_controller.error(cx, v);
347
348 if !self.canceled_1.get() || !self.canceled_2.get() {
350 self.cancel_promise.resolve_native(cx, &());
351 }
352 }
353}
354
355#[dom_struct]
357pub(crate) struct ReadableStreamDefaultReader {
358 reflector_: Reflector,
359
360 stream: MutNullableDom<ReadableStream>,
362
363 read_requests: DomRefCell<VecDeque<ReadRequest>>,
364
365 closed_promise: DomRefCell<TracedPromise>,
367}
368
369impl ReadableStreamDefaultReader {
370 fn new_with_proto(
371 cx: &mut JSContext,
372 global: &GlobalScope,
373 proto: Option<SafeHandleObject>,
374 ) -> DomRoot<ReadableStreamDefaultReader> {
375 let closed_promise = Promise::new(cx, global);
376 reflect_dom_object_with_proto(
377 cx,
378 Box::new(ReadableStreamDefaultReader::new_inherited(&closed_promise)),
379 global,
380 proto,
381 )
382 }
383
384 fn new_inherited(promise: &RootedPromise) -> ReadableStreamDefaultReader {
385 ReadableStreamDefaultReader {
386 reflector_: Reflector::new(),
387 stream: MutNullableDom::new(None),
388 read_requests: DomRefCell::new(Default::default()),
389 closed_promise: DomRefCell::new(promise.to_traced()),
390 }
391 }
392
393 pub(crate) fn new(
394 cx: &mut JSContext,
395 global: &GlobalScope,
396 ) -> DomRoot<ReadableStreamDefaultReader> {
397 let closed_promise = Promise::new(cx, global);
398 reflect_dom_object(cx, Box::new(Self::new_inherited(&closed_promise)), global)
399 }
400
401 pub(crate) fn set_up(
403 &self,
404 cx: &mut JSContext,
405 stream: &ReadableStream,
406 global: &GlobalScope,
407 ) -> Fallible<()> {
408 if stream.is_locked() {
410 return Err(Error::Type(c"stream is locked".to_owned()));
411 }
412 self.generic_initialize(cx, global, stream);
415
416 self.read_requests.borrow_mut().clear();
418
419 Ok(())
420 }
421
422 pub(crate) fn close(&self, cx: &mut js::context::JSContext) {
424 self.closed_promise.borrow().resolve_native(cx, &());
426 rooted!(&in(cx) let mut read_requests = self.take_read_requests());
429 for request in read_requests.iter() {
432 request.close_steps(cx);
434 }
435 }
436
437 pub(crate) fn add_read_request(&self, read_request: &ReadRequest) {
439 self.read_requests
440 .borrow_mut()
441 .push_back(read_request.clone());
442 }
443
444 pub(crate) fn get_num_read_requests(&self) -> usize {
446 self.read_requests.borrow().len()
447 }
448
449 pub(crate) fn error(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) {
451 self.closed_promise.borrow().reject_native(cx, &e);
453
454 self.closed_promise.borrow().set_promise_is_handled(cx);
456
457 self.error_read_requests(cx, e);
459 }
460
461 pub(crate) fn remove_read_request(&self) -> ReadRequest {
463 self.read_requests
464 .borrow_mut()
465 .pop_front()
466 .expect("Reader must have read request when remove is called into.")
467 }
468
469 pub(crate) fn release(&self, cx: &mut js::context::JSContext) -> Fallible<()> {
471 self.generic_release(cx).expect("Generic release failed");
473 rooted!(&in(cx) let mut error = UndefinedValue());
475 Error::Type(c"Reader is released".to_owned()).to_jsval(
476 cx,
477 &self.global(),
478 error.handle_mut(),
479 );
480
481 self.error_read_requests(cx, error.handle());
483 Ok(())
484 }
485
486 fn take_read_requests(&self) -> VecDeque<ReadRequest> {
487 mem::take(&mut *self.read_requests.borrow_mut())
488 }
489
490 fn error_read_requests(&self, cx: &mut js::context::JSContext, rval: SafeHandleValue) {
492 rooted!(&in(cx) let mut read_requests = self.take_read_requests());
494
495 for request in read_requests.iter() {
497 request.error_steps(cx, rval);
498 }
499 }
500
501 pub(crate) fn read(&self, cx: &mut js::context::JSContext, read_request: &ReadRequest) {
503 assert!(self.stream.get().is_some());
507
508 let stream = self.stream.get().unwrap();
509
510 stream.set_is_disturbed(true);
512 if stream.is_closed() {
514 read_request.close_steps(cx);
515 } else if stream.is_errored() {
516 rooted!(&in(cx) let mut error = UndefinedValue());
519 stream.get_stored_error(error.handle_mut());
520 read_request.error_steps(cx, error.handle());
521 } else {
522 assert!(stream.is_readable());
525 stream.perform_pull_steps(cx, read_request);
527 }
528 }
529
530 #[allow(clippy::too_many_arguments)]
533 pub(crate) fn byte_tee_append_native_handler_to_closed_promise(
534 &self,
535 cx: &mut js::context::JSContext,
536 branch_1: &ReadableStream,
537 branch_2: &ReadableStream,
538 canceled_1: Rc<Cell<bool>>,
539 canceled_2: Rc<Cell<bool>>,
540 cancel_promise: &RootedPromise,
541 reader_version: Rc<Cell<u64>>,
542 expected_version: u64,
543 ) {
544 let branch_1_controller = branch_1.get_byte_controller();
546 let branch_2_controller = branch_2.get_byte_controller();
547
548 let global = self.global();
549 let handler = PromiseNativeHandler::new(
550 cx,
551 &global,
552 None,
553 Some(Box::new(ByteTeeClosedPromiseRejectionHandler {
554 branch_1_controller: Dom::from_ref(&branch_1_controller),
555 branch_2_controller: Dom::from_ref(&branch_2_controller),
556 canceled_1,
557 canceled_2,
558 cancel_promise: cancel_promise.to_traced(),
559 reader_version,
560 expected_version,
561 })),
562 );
563
564 let mut realm = enter_auto_realm(cx, &*global);
565 let cx = &mut realm.current_realm();
566
567 self.closed_promise
568 .borrow()
569 .append_native_handler(cx, &handler);
570 }
571
572 pub(crate) fn default_tee_append_native_handler_to_closed_promise(
574 &self,
575 cx: &mut js::context::JSContext,
576 branch_1: &ReadableStream,
577 branch_2: &ReadableStream,
578 canceled_1: Rc<Cell<bool>>,
579 canceled_2: Rc<Cell<bool>>,
580 cancel_promise: &RootedPromise,
581 ) {
582 let branch_1_controller = branch_1.get_default_controller();
583
584 let branch_2_controller = branch_2.get_default_controller();
585
586 let global = self.global();
587 let handler = PromiseNativeHandler::new(
588 cx,
589 &global,
590 None,
591 Some(Box::new(DefaultTeeClosedPromiseRejectionHandler {
592 branch_1_controller: Dom::from_ref(&branch_1_controller),
593 branch_2_controller: Dom::from_ref(&branch_2_controller),
594 canceled_1,
595 canceled_2,
596 cancel_promise: cancel_promise.to_traced(),
597 })),
598 );
599
600 let mut realm = enter_auto_realm(cx, &*global);
601 let cx = &mut realm.current_realm();
602
603 self.closed_promise
604 .borrow()
605 .append_native_handler(cx, &handler);
606 }
607
608 pub(crate) fn read_all_bytes(
610 &self,
611 cx: &mut js::context::JSContext,
612 success_steps: Rc<ReadAllBytesSuccessSteps>,
613 failure_steps: Rc<ReadAllBytesFailureSteps>,
614 ) {
615 read_loop(cx, self, success_steps, failure_steps);
620 }
621
622 pub(crate) fn process_read_requests(
624 &self,
625 cx: &mut js::context::JSContext,
626 controller: &ReadableByteStreamController,
627 ) -> Fallible<()> {
628 while !self.read_requests.borrow().is_empty() {
630 if controller.get_queue_total_size() == 0.0 {
632 return Ok(());
633 }
634
635 rooted!(&in(cx) let read_request = self.remove_read_request());
638
639 controller
641 .fill_read_request_from_queue(cx, &read_request)
642 .expect("Fill read request from queue failed");
643 }
644 Ok(())
645 }
646}
647
648impl ReadableStreamDefaultReaderMethods<crate::DomTypeHolder> for ReadableStreamDefaultReader {
649 fn Constructor(
651 cx: &mut JSContext,
652 global: &GlobalScope,
653 proto: Option<SafeHandleObject>,
654 stream: &ReadableStream,
655 ) -> Fallible<DomRoot<Self>> {
656 let reader = Self::new_with_proto(cx, global, proto);
657
658 reader.set_up(cx, stream, global)?;
660
661 Ok(reader)
662 }
663
664 fn Read(&self, cx: &mut js::context::JSContext) -> RootedPromise {
666 if self.stream.get().is_none() {
668 rooted!(&in(cx) let mut error = UndefinedValue());
669 Error::Type(c"stream is undefined".to_owned()).to_jsval(
670 cx,
671 &self.global(),
672 error.handle_mut(),
673 );
674 return Promise::new_rejected(cx, &self.global(), error.handle());
675 }
676 let promise = Promise::new(cx, &self.global());
678
679 rooted!(&in(cx) let read_request = ReadRequest::Read(promise.to_traced()));
690
691 self.read(cx, &read_request);
693
694 promise
696 }
697
698 fn ReleaseLock(&self, cx: &mut js::context::JSContext) -> Fallible<()> {
700 if self.stream.get().is_none() {
701 return Ok(());
703 }
704
705 self.release(cx)
707 }
708
709 fn Closed(&self, cx: &JSContext) -> RootedPromise {
711 self.closed(cx)
712 }
713
714 fn Cancel(&self, cx: &mut js::context::JSContext, reason: SafeHandleValue) -> RootedPromise {
716 self.generic_cancel(cx, &self.global(), reason)
717 }
718}
719
720impl ReadableStreamGenericReader for ReadableStreamDefaultReader {
721 fn get_closed_promise(&self, cx: &JSContext) -> RootedPromise {
722 self.closed_promise.borrow().root(cx)
723 }
724
725 fn set_closed_promise(&self, promise: &RootedPromise) {
726 *self.closed_promise.borrow_mut() = promise.to_traced();
727 }
728
729 fn set_stream(&self, stream: Option<&ReadableStream>) {
730 self.stream.set(stream);
731 }
732
733 fn get_stream(&self) -> Option<DomRoot<ReadableStream>> {
734 self.stream.get()
735 }
736
737 fn as_default_reader(&self) -> Option<&ReadableStreamDefaultReader> {
738 Some(self)
739 }
740}