script/dom/stream/readablebytestreamcontroller.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 http://mozilla.org/MPL/2.0/. */
4
5use std::cell::Cell;
6use std::cmp::min;
7use std::collections::VecDeque;
8use std::rc::Rc;
9
10use dom_struct::dom_struct;
11use js::context::JSContext;
12use js::jsapi::{Heap, Type};
13use js::jsval::UndefinedValue;
14use js::realm::CurrentRealm;
15use js::rust::{HandleObject, HandleValue as SafeHandleValue, HandleValue};
16use js::typedarray::{ArrayBufferU8, ArrayBufferViewU8};
17use script_bindings::cell::DomRefCell;
18use script_bindings::reflector::{Reflector, reflect_dom_object};
19
20use super::readablestreambyobreader::ReadIntoRequest;
21use super::readablestreamdefaultreader::ReadRequest;
22use super::underlyingsourcecontainer::{UnderlyingSourceContainer, UnderlyingSourceType};
23use crate::dom::bindings::buffer_source::{
24 Constructor, HeapBufferSource, byte_size, create_array_buffer_with_size,
25 create_buffer_source_with_constructor,
26};
27use crate::dom::bindings::codegen::Bindings::ReadableByteStreamControllerBinding::ReadableByteStreamControllerMethods;
28use crate::dom::bindings::codegen::UnionTypes::ReadableStreamDefaultControllerOrReadableByteStreamController as Controller;
29use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible};
30use crate::dom::bindings::reflector::DomGlobal;
31use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
32use crate::dom::bindings::trace::RootedTraceableBox;
33use crate::dom::globalscope::GlobalScope;
34use crate::dom::promise::Promise;
35use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
36use crate::dom::stream::readablestream::ReadableStream;
37use crate::dom::stream::readablestreambyobrequest::ReadableStreamBYOBRequest;
38use crate::realms::enter_auto_realm;
39
40/// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry>
41#[derive(JSTraceable, MallocSizeOf)]
42#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
43pub(crate) struct QueueEntry {
44 /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-buffer>
45 #[ignore_malloc_size_of = "HeapBufferSource"]
46 buffer: HeapBufferSource<ArrayBufferU8>,
47 /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-byte-offset>
48 byte_offset: usize,
49 /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-byte-length>
50 byte_length: usize,
51}
52
53impl js::gc::Rootable for QueueEntry {}
54
55impl QueueEntry {
56 pub(crate) fn new(
57 buffer: RootedTraceableBox<HeapBufferSource<ArrayBufferU8>>,
58 byte_offset: usize,
59 byte_length: usize,
60 ) -> QueueEntry {
61 QueueEntry {
62 buffer: *buffer.into_box(),
63 byte_offset,
64 byte_length,
65 }
66 }
67}
68
69#[derive(Debug, Eq, JSTraceable, MallocSizeOf, PartialEq)]
70pub(crate) enum ReaderType {
71 /// <https://streams.spec.whatwg.org/#readablestreambyobreader>
72 Byob,
73 /// <https://streams.spec.whatwg.org/#readablestreamdefaultreader>
74 Default,
75}
76
77/// <https://streams.spec.whatwg.org/#pull-into-descriptor>
78#[derive(Eq, JSTraceable, MallocSizeOf, PartialEq)]
79#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
80pub(crate) struct PullIntoDescriptor {
81 #[ignore_malloc_size_of = "HeapBufferSource"]
82 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-buffer>
83 buffer: HeapBufferSource<ArrayBufferU8>,
84 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-buffer-byte-length>
85 buffer_byte_length: u64,
86 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-byte-offset>
87 byte_offset: u64,
88 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-byte-length>
89 byte_length: u64,
90 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-bytes-filled>
91 bytes_filled: Cell<u64>,
92 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-minimum-fill>
93 minimum_fill: u64,
94 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-element-size>
95 element_size: u64,
96 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-view-constructor>
97 view_constructor: Constructor,
98 /// <https://streams.spec.whatwg.org/#pull-into-descriptor-reader-type>
99 reader_type: Option<ReaderType>,
100}
101
102impl js::gc::Rootable for PullIntoDescriptor {}
103
104/// The fulfillment handler for
105/// <https://streams.spec.whatwg.org/#dom-underlyingsource-start>
106#[derive(Clone, JSTraceable, MallocSizeOf)]
107#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
108struct StartAlgorithmFulfillmentHandler {
109 controller: Dom<ReadableByteStreamController>,
110}
111
112impl Callback for StartAlgorithmFulfillmentHandler {
113 /// Continuation of <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
114 /// Upon fulfillment of startPromise,
115 fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
116 // Set controller.[[started]] to true.
117 self.controller.started.set(true);
118
119 // Assert: controller.[[pulling]] is false.
120 assert!(!self.controller.pulling.get());
121
122 // Assert: controller.[[pullAgain]] is false.
123 assert!(!self.controller.pull_again.get());
124
125 // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
126 self.controller.call_pull_if_needed(cx);
127 }
128}
129
130/// The rejection handler for
131/// <https://streams.spec.whatwg.org/#dom-underlyingsource-start>
132#[derive(Clone, JSTraceable, MallocSizeOf)]
133#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
134struct StartAlgorithmRejectionHandler {
135 controller: Dom<ReadableByteStreamController>,
136}
137
138impl Callback for StartAlgorithmRejectionHandler {
139 /// Continuation of <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
140 /// Upon rejection of startPromise with reason r,
141 fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
142 // Perform ! ReadableByteStreamControllerError(controller, r).
143 self.controller.error(cx, v);
144 }
145}
146
147/// The fulfillment handler for
148/// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
149#[derive(Clone, JSTraceable, MallocSizeOf)]
150#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
151struct PullAlgorithmFulfillmentHandler {
152 controller: Dom<ReadableByteStreamController>,
153}
154
155impl Callback for PullAlgorithmFulfillmentHandler {
156 /// Continuation of <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
157 /// Upon fulfillment of pullPromise
158 fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
159 // Set controller.[[pulling]] to false.
160 self.controller.pulling.set(false);
161
162 // If controller.[[pullAgain]] is true,
163 if self.controller.pull_again.get() {
164 // Set controller.[[pullAgain]] to false.
165 self.controller.pull_again.set(false);
166
167 // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
168 self.controller.call_pull_if_needed(cx);
169 }
170 }
171}
172
173/// The rejection handler for
174/// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
175#[derive(Clone, JSTraceable, MallocSizeOf)]
176#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
177struct PullAlgorithmRejectionHandler {
178 controller: Dom<ReadableByteStreamController>,
179}
180
181impl Callback for PullAlgorithmRejectionHandler {
182 /// Continuation of <https://streams.spec.whatwg.org/#readable-stream-byte-controller-call-pull-if-needed>
183 /// Upon rejection of pullPromise with reason e.
184 fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
185 // Perform ! ReadableByteStreamControllerError(controller, e).
186 self.controller.error(cx, v);
187 }
188}
189
190/// <https://streams.spec.whatwg.org/#readablebytestreamcontroller>
191#[dom_struct]
192pub(crate) struct ReadableByteStreamController {
193 reflector_: Reflector,
194 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-autoallocatechunksize>
195 auto_allocate_chunk_size: Option<u64>,
196 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-stream>
197 stream: MutNullableDom<ReadableStream>,
198 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-strategyhwm>
199 strategy_hwm: f64,
200 /// A mutable reference to the underlying source is used to implement these two
201 /// internal slots:
202 ///
203 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pullalgorithm>
204 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-cancelalgorithm>
205 underlying_source: MutNullableDom<UnderlyingSourceContainer>,
206 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-queue>
207 queue: DomRefCell<VecDeque<QueueEntry>>,
208 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-queuetotalsize>
209 queue_total_size: Cell<f64>,
210 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-byobrequest>
211 byob_request: MutNullableDom<ReadableStreamBYOBRequest>,
212 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pendingpullintos>
213 pending_pull_intos: DomRefCell<Vec<PullIntoDescriptor>>,
214 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-closerequested>
215 close_requested: Cell<bool>,
216 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-started>
217 started: Cell<bool>,
218 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pulling>
219 pulling: Cell<bool>,
220 /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pullalgorithm>
221 pull_again: Cell<bool>,
222}
223
224impl ReadableByteStreamController {
225 fn new_inherited(
226 underlying_source_container: &UnderlyingSourceContainer,
227 strategy_hwm: f64,
228 ) -> ReadableByteStreamController {
229 let auto_allocate_chunk_size = underlying_source_container.auto_allocate_chunk_size();
230 ReadableByteStreamController {
231 reflector_: Reflector::new(),
232 byob_request: MutNullableDom::new(None),
233 stream: MutNullableDom::new(None),
234 underlying_source: MutNullableDom::new(Some(underlying_source_container)),
235 auto_allocate_chunk_size,
236 pending_pull_intos: DomRefCell::new(Vec::new()),
237 strategy_hwm,
238 close_requested: Default::default(),
239 queue: DomRefCell::new(Default::default()),
240 queue_total_size: Default::default(),
241 started: Default::default(),
242 pulling: Default::default(),
243 pull_again: Default::default(),
244 }
245 }
246
247 pub(crate) fn new(
248 cx: &mut JSContext,
249 underlying_source_type: UnderlyingSourceType,
250 strategy_hwm: f64,
251 global: &GlobalScope,
252 ) -> DomRoot<ReadableByteStreamController> {
253 let underlying_source_container =
254 UnderlyingSourceContainer::new(cx, global, underlying_source_type);
255 reflect_dom_object(
256 cx,
257 Box::new(ReadableByteStreamController::new_inherited(
258 &underlying_source_container,
259 strategy_hwm,
260 )),
261 global,
262 )
263 }
264
265 #[allow(dead_code)]
266 pub(crate) fn set_stream(&self, stream: &ReadableStream) {
267 self.stream.set(Some(stream))
268 }
269
270 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-pull-into>
271 pub(crate) fn perform_pull_into(
272 &self,
273 cx: &mut JSContext,
274 read_into_request: &ReadIntoRequest,
275 view: &HeapBufferSource<ArrayBufferViewU8>,
276 min: u64,
277 ) {
278 // Let stream be controller.[[stream]].
279 let stream = self.stream.get().unwrap();
280
281 // Let elementSize be 1.
282 let mut element_size = 1;
283
284 // Let ctor be %DataView%.
285 let mut ctor = Constructor::DataView;
286
287 // If view has a [[TypedArrayName]] internal slot (i.e., it is not a DataView),
288 if view.has_typed_array_name() {
289 // Set elementSize to the element size specified in the
290 // typed array constructors table for view.[[TypedArrayName]].
291 let view_typw = view.get_array_buffer_view_type();
292 element_size = byte_size(view_typw);
293
294 // Set ctor to the constructor specified in the typed array constructors table for view.[[TypedArrayName]].
295 ctor = Constructor::Name(view_typw);
296 }
297
298 // Let minimumFill be min × elementSize.
299 let minimum_fill = min * element_size;
300
301 // Assert: minimumFill ≥ 0 and minimumFill ≤ view.[[ByteLength]].
302 assert!(minimum_fill <= (view.byte_length() as u64));
303
304 // Assert: the remainder after dividing minimumFill by elementSize is 0.
305 assert_eq!(minimum_fill % element_size, 0);
306
307 // Let byteOffset be view.[[ByteOffset]].
308 let byte_offset = view.get_byte_offset();
309
310 // Let byteLength be view.[[ByteLength]].
311 let byte_length = view.byte_length();
312
313 // Let bufferResult be TransferArrayBuffer(view.[[ViewedArrayBuffer]]).
314 match view
315 .get_array_buffer_view_buffer(cx)
316 .transfer_array_buffer(cx)
317 {
318 Ok(buffer) => {
319 // Let buffer be bufferResult.[[Value]].
320 // Let pullIntoDescriptor be a new pull-into descriptor with
321 // buffer buffer
322 // buffer byte length buffer.[[ArrayBufferByteLength]]
323 // byte offset byteOffset
324 // byte length byteLength
325 // bytes filled 0
326 // minimum fill minimumFill
327 // element size elementSize
328 // view constructor ctor
329 // reader type "byob"
330 let buffer_byte_length = buffer.byte_length();
331 let pull_into_descriptor = RootedTraceableBox::new(PullIntoDescriptor {
332 buffer: *buffer.into_box(),
333 buffer_byte_length: buffer_byte_length as u64,
334 byte_offset: byte_offset as u64,
335 byte_length: byte_length as u64,
336 bytes_filled: Cell::new(0),
337 minimum_fill,
338 element_size,
339 view_constructor: ctor.clone(),
340 reader_type: Some(ReaderType::Byob),
341 });
342
343 // If controller.[[pendingPullIntos]] is not empty,
344 {
345 let mut pending_pull_intos = self.pending_pull_intos.safe_borrow_mut(cx);
346 if !pending_pull_intos.is_empty() {
347 // Append pullIntoDescriptor to controller.[[pendingPullIntos]].
348 pending_pull_intos.push(*pull_into_descriptor.into_box());
349
350 // Perform ! ReadableStreamAddReadIntoRequest(stream, readIntoRequest).
351 stream.add_read_into_request(read_into_request);
352
353 // Return.
354 return;
355 }
356 }
357
358 // If stream.[[state]] is "closed",
359 if stream.is_closed() {
360 // Let emptyView be ! Construct(ctor, « pullIntoDescriptor’s buffer,
361 // pullIntoDescriptor’s byte offset, 0 »).
362 if let Ok(empty_view) = create_buffer_source_with_constructor(
363 cx,
364 &ctor,
365 &pull_into_descriptor.buffer,
366 pull_into_descriptor.byte_offset as usize,
367 0,
368 ) {
369 // Perform readIntoRequest’s close steps, given emptyView.
370 let result = RootedTraceableBox::new(Heap::default());
371 rooted!(&in(cx) let mut view_value = UndefinedValue());
372 empty_view.get_buffer_view_value(cx, view_value.handle_mut());
373 result.set(*view_value);
374
375 read_into_request.close_steps(cx, Some(result));
376
377 // Return.
378 return;
379 } else {
380 return;
381 }
382 }
383
384 // If controller.[[queueTotalSize]] > 0,
385 if self.queue_total_size.get() > 0.0 {
386 // If ! ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(
387 // controller, pullIntoDescriptor) is true,
388 if self.fill_pull_into_descriptor_from_queue(cx, &pull_into_descriptor) {
389 // Let filledView be ! ReadableByteStreamControllerConvertPullIntoDescriptor(
390 // pullIntoDescriptor).
391 if let Ok(filled_view) =
392 self.convert_pull_into_descriptor(cx, &pull_into_descriptor)
393 {
394 // Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
395 self.handle_queue_drain(cx);
396
397 // Perform readIntoRequest’s chunk steps, given filledView.
398 let result = RootedTraceableBox::new(Heap::default());
399 rooted!(&in(cx) let mut view_value = UndefinedValue());
400 filled_view.get_buffer_view_value(cx, view_value.handle_mut());
401 result.set(*view_value);
402 read_into_request.chunk_steps(cx, result);
403
404 // Return.
405 return;
406 } else {
407 return;
408 }
409 }
410
411 // If controller.[[closeRequested]] is true,
412 if self.close_requested.get() {
413 // Let e be a new TypeError exception.
414 rooted!(&in(cx) let mut error = UndefinedValue());
415 Error::Type(c"close requested".to_owned()).to_jsval(
416 cx,
417 &self.global(),
418 error.handle_mut(),
419 );
420
421 // Perform ! ReadableByteStreamControllerError(controller, e).
422 self.error(cx, error.handle());
423
424 // Perform readIntoRequest’s error steps, given e.
425 read_into_request.error_steps(cx, error.handle());
426
427 // Return.
428 return;
429 }
430 }
431
432 // Append pullIntoDescriptor to controller.[[pendingPullIntos]].
433 {
434 self.pending_pull_intos
435 .safe_borrow_mut(cx)
436 .push(*pull_into_descriptor.into_box());
437 }
438 // Perform ! ReadableStreamAddReadIntoRequest(stream, readIntoRequest).
439 stream.add_read_into_request(read_into_request);
440
441 // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
442 self.call_pull_if_needed(cx);
443 },
444 Err(error) => {
445 // If bufferResult is an abrupt completion,
446
447 // Perform readIntoRequest’s error steps, given bufferResult.[[Value]].
448 rooted!(&in(cx) let mut rval = UndefinedValue());
449 error.to_jsval(cx, &self.global(), rval.handle_mut());
450 read_into_request.error_steps(cx, rval.handle());
451
452 // Return.
453 },
454 }
455 }
456
457 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond>
458 pub(crate) fn respond(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
459 let heap_buffer = {
460 // Assert: controller.[[pendingPullIntos]] is not empty.
461 let pending_pull_intos = self.pending_pull_intos.borrow();
462 assert!(!pending_pull_intos.is_empty());
463
464 // Let firstDescriptor be controller.[[pendingPullIntos]][0].
465 let first_descriptor = pending_pull_intos.first().unwrap();
466
467 // Let state be controller.[[stream]].[[state]].
468 let stream = self.stream.get().unwrap();
469
470 // If state is "closed",
471 if stream.is_closed() {
472 // If bytesWritten is not 0, throw a TypeError exception.
473 if bytes_written != 0 {
474 return Err(Error::Type(
475 c"bytesWritten not zero on closed stream".to_owned(),
476 ));
477 }
478 } else {
479 // Assert: state is "readable".
480 assert!(stream.is_readable());
481
482 // If bytesWritten is 0, throw a TypeError exception.
483 if bytes_written == 0 {
484 return Err(Error::Type(c"bytesWritten is 0".to_owned()));
485 }
486
487 // If firstDescriptor’s bytes filled + bytesWritten > firstDescriptor’s byte length,
488 // throw a RangeError exception.
489 if first_descriptor.bytes_filled.get() + bytes_written >
490 first_descriptor.byte_length
491 {
492 return Err(Error::Range(
493 c"bytes filled + bytesWritten > byte length".to_owned(),
494 ));
495 }
496 }
497
498 first_descriptor
499 .buffer
500 .transfer_array_buffer(cx)
501 .expect("TransferArrayBuffer failed")
502 };
503 // Set firstDescriptor’s buffer to ! TransferArrayBuffer(firstDescriptor’s buffer).
504 self.pending_pull_intos
505 .safe_borrow_mut(cx)
506 .first_mut()
507 .unwrap()
508 .buffer = *(heap_buffer.into_box());
509
510 // Perform ? ReadableByteStreamControllerRespondInternal(controller, bytesWritten).
511 self.respond_internal(cx, bytes_written)
512 }
513
514 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-internal>
515 fn respond_internal(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
516 {
517 // Let firstDescriptor be controller.[[pendingPullIntos]][0].
518 let pending_pull_intos = self.pending_pull_intos.borrow();
519 let first_descriptor = pending_pull_intos.first().unwrap();
520
521 // Assert: ! CanTransferArrayBuffer(firstDescriptor’s buffer) is true
522 assert!(first_descriptor.buffer.can_transfer_array_buffer(cx));
523 }
524
525 // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
526 self.invalidate_byob_request();
527
528 // Let state be controller.[[stream]].[[state]].
529 let stream = self.stream.get().unwrap();
530
531 // If state is "closed",
532 if stream.is_closed() {
533 // Assert: bytesWritten is 0.
534 assert_eq!(bytes_written, 0);
535
536 // Perform ! ReadableByteStreamControllerRespondInClosedState(controller, firstDescriptor).
537 self.respond_in_closed_state(cx)
538 .expect("respond_in_closed_state failed");
539 } else {
540 // Assert: state is "readable".
541 assert!(stream.is_readable());
542
543 // Assert: bytesWritten > 0.
544 assert!(bytes_written > 0);
545
546 // Perform ? ReadableByteStreamControllerRespondInReadableState(controller, bytesWritten, firstDescriptor).
547 self.respond_in_readable_state(cx, bytes_written)?;
548 }
549
550 // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
551 self.call_pull_if_needed(cx);
552
553 Ok(())
554 }
555
556 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-in-closed-state>
557 fn respond_in_closed_state(&self, cx: &mut JSContext) -> Fallible<()> {
558 let pending_pull_intos = self.pending_pull_intos.borrow();
559 let first_descriptor = pending_pull_intos.first().unwrap();
560
561 // Assert: the remainder after dividing firstDescriptor’s bytes filled
562 // by firstDescriptor’s element size is 0.
563 assert_eq!(
564 first_descriptor.bytes_filled.get() % first_descriptor.element_size,
565 0
566 );
567
568 // If firstDescriptor’s reader type is "none",
569 // perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
570 let reader_type = first_descriptor.reader_type.is_none();
571
572 // needed to drop the borrow and avoid BorrowMutError
573 drop(pending_pull_intos);
574
575 if reader_type {
576 self.shift_pending_pull_into();
577 }
578
579 // Let stream be controller.[[stream]].
580 let stream = self.stream.get().unwrap();
581
582 // If ! ReadableStreamHasBYOBReader(stream) is true,
583 if stream.has_byob_reader() {
584 // Let filledPullIntos be a new empty list.
585 rooted!(&in(cx) let mut filled_pull_intos = Vec::new());
586
587 // While filledPullIntos’s size < ! ReadableStreamGetNumReadIntoRequests(stream),
588 while filled_pull_intos.len() < stream.get_num_read_into_requests() {
589 // Let pullIntoDescriptor be ! ReadableByteStreamControllerShiftPendingPullInto(controller).
590 // Append pullIntoDescriptor to filledPullIntos.
591 filled_pull_intos.push(self.shift_pending_pull_into());
592 }
593
594 // For each filledPullInto of filledPullIntos,
595 for filled_pull_into in &*filled_pull_intos {
596 // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(stream, filledPullInto).
597 self.commit_pull_into_descriptor(cx, filled_pull_into)
598 .expect("commit_pull_into_descriptor failed");
599 }
600 }
601
602 Ok(())
603 }
604
605 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-in-readable-state>
606 fn respond_in_readable_state(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
607 let pending_pull_intos = self.pending_pull_intos.borrow();
608 let first_descriptor = pending_pull_intos.first().unwrap();
609
610 // Assert: pullIntoDescriptor’s bytes filled + bytesWritten ≤ pullIntoDescriptor’s byte length.
611 assert!(
612 first_descriptor.bytes_filled.get() + bytes_written <= first_descriptor.byte_length
613 );
614
615 // Perform ! ReadableByteStreamControllerFillHeadPullIntoDescriptor(
616 // controller, bytesWritten, pullIntoDescriptor).
617 self.fill_head_pull_into_descriptor(bytes_written, first_descriptor);
618
619 // If pullIntoDescriptor’s reader type is "none",
620 if first_descriptor.reader_type.is_none() {
621 // needed to drop the borrow and avoid BorrowMutError
622 drop(pending_pull_intos);
623
624 // Perform ? ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(controller, pullIntoDescriptor).
625 self.enqueue_detached_pull_into_to_queue(cx)?;
626
627 // Let filledPullIntos be the result of performing
628 // ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
629 rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));
630
631 // For each filledPullInto of filledPullIntos,
632 for filled_pull_into in &*filled_pull_intos {
633 // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]]
634 // , filledPullInto).
635 self.commit_pull_into_descriptor(cx, filled_pull_into)
636 .expect("commit_pull_into_descriptor failed");
637 }
638
639 // Return.
640 return Ok(());
641 }
642
643 // If pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill, return.
644 if first_descriptor.bytes_filled.get() < first_descriptor.minimum_fill {
645 return Ok(());
646 }
647
648 // needed to drop the borrow and avoid BorrowMutError
649 drop(pending_pull_intos);
650
651 // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
652 rooted!(&in(cx) let pull_into_descriptor = self.shift_pending_pull_into());
653
654 // Let remainderSize be the remainder after dividing pullIntoDescriptor’s bytes
655 // filled by pullIntoDescriptor’s element size.
656 let remainder_size =
657 pull_into_descriptor.bytes_filled.get() % pull_into_descriptor.element_size;
658
659 // If remainderSize > 0,
660 if remainder_size > 0 {
661 // Let end be pullIntoDescriptor’s byte offset + pullIntoDescriptor’s bytes filled.
662 let end = pull_into_descriptor.byte_offset + pull_into_descriptor.bytes_filled.get();
663
664 // Perform ? ReadableByteStreamControllerEnqueueClonedChunkToQueue(controller,
665 // pullIntoDescriptor’s buffer, end − remainderSize, remainderSize).
666 self.enqueue_cloned_chunk_to_queue(
667 cx,
668 &pull_into_descriptor.buffer,
669 end - remainder_size,
670 remainder_size,
671 )?;
672 }
673
674 // Set pullIntoDescriptor’s bytes filled to pullIntoDescriptor’s bytes filled − remainderSize.
675 pull_into_descriptor
676 .bytes_filled
677 .set(pull_into_descriptor.bytes_filled.get() - remainder_size);
678
679 // Let filledPullIntos be the result of performing
680 // ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
681 rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));
682
683 // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]], pullIntoDescriptor).
684 self.commit_pull_into_descriptor(cx, &pull_into_descriptor)
685 .expect("commit_pull_into_descriptor failed");
686
687 // For each filledPullInto of filledPullIntos,
688 for filled_pull_into in &*filled_pull_intos {
689 // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]], filledPullInto).
690 self.commit_pull_into_descriptor(cx, filled_pull_into)
691 .expect("commit_pull_into_descriptor failed");
692 }
693
694 Ok(())
695 }
696
697 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-with-new-view>
698 pub(crate) fn respond_with_new_view(
699 &self,
700 cx: &mut JSContext,
701 view: &HeapBufferSource<ArrayBufferViewU8>,
702 ) -> Fallible<()> {
703 let view_byte_length;
704
705 let view = {
706 // Assert: controller.[[pendingPullIntos]] is not empty.
707 let pending_pull_intos = self.pending_pull_intos.borrow();
708 assert!(!pending_pull_intos.is_empty());
709
710 // Assert: ! IsDetachedBuffer(view.[[ViewedArrayBuffer]]) is false.
711 assert!(!view.is_detached_buffer(cx));
712
713 // Let firstDescriptor be controller.[[pendingPullIntos]][0].
714 let first_descriptor = pending_pull_intos.first().unwrap();
715
716 // Let state be controller.[[stream]].[[state]].
717 let stream = self.stream.get().unwrap();
718
719 // If state is "closed",
720 if stream.is_closed() {
721 // If view.[[ByteLength]] is not 0, throw a TypeError exception.
722 if view.byte_length() != 0 {
723 return Err(Error::Type(c"view byte length is not 0".to_owned()));
724 }
725 } else {
726 // Assert: state is "readable".
727 assert!(stream.is_readable());
728
729 // If view.[[ByteLength]] is 0, throw a TypeError exception.
730 if view.byte_length() == 0 {
731 return Err(Error::Type(c"view byte length is 0".to_owned()));
732 }
733 }
734
735 // If firstDescriptor’s byte offset + firstDescriptor’ bytes filled is not view.[[ByteOffset]],
736 // throw a RangeError exception.
737 if first_descriptor.byte_offset + first_descriptor.bytes_filled.get() !=
738 (view.get_byte_offset() as u64)
739 {
740 return Err(Error::Range(
741 c"firstDescriptor's byte offset + bytes filled is not view byte offset"
742 .to_owned(),
743 ));
744 }
745
746 // If firstDescriptor’s buffer byte length is not view.[[ViewedArrayBuffer]].[[ByteLength]],
747 // throw a RangeError exception.
748 if first_descriptor.buffer_byte_length !=
749 (view.viewed_buffer_array_byte_length(cx) as u64)
750 {
751 return Err(Error::Range(
752 c"firstDescriptor's buffer byte length is not view viewed buffer array byte length"
753 .to_owned(),
754 ));
755 }
756
757 // If firstDescriptor’s bytes filled + view.[[ByteLength]] > firstDescriptor’s byte length,
758 // throw a RangeError exception.
759 if first_descriptor.bytes_filled.get() + (view.byte_length()) as u64 >
760 first_descriptor.byte_length
761 {
762 return Err(Error::Range(
763 c"bytes filled + view byte length > byte length".to_owned(),
764 ));
765 }
766
767 // Let viewByteLength be view.[[ByteLength]].
768 view_byte_length = view.byte_length();
769
770 view.get_array_buffer_view_buffer(cx)
771 .transfer_array_buffer(cx)?
772 };
773 // Set firstDescriptor’s buffer to ? TransferArrayBuffer(view.[[ViewedArrayBuffer]]).
774 self.pending_pull_intos
775 .safe_borrow_mut(cx)
776 .first_mut()
777 .unwrap()
778 .buffer = *view.into_box();
779
780 // Perform ? ReadableByteStreamControllerRespondInternal(controller, viewByteLength).
781 self.respond_internal(cx, view_byte_length as u64)
782 }
783
784 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-get-desired-size>
785 pub(crate) fn get_desired_size(&self) -> Option<f64> {
786 // Let state be controller.[[stream]].[[state]].
787 let stream = self.stream.get()?;
788
789 // If state is "errored", return null.
790 if stream.is_errored() {
791 return None;
792 }
793
794 // If state is "closed", return 0.
795 if stream.is_closed() {
796 return Some(0.0);
797 }
798
799 // Return controller.[[strategyHWM]] − controller.[[queueTotalSize]].
800 Some(self.strategy_hwm - self.queue_total_size.get())
801 }
802
803 /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollergetbyobrequest>
804 pub(crate) fn get_byob_request(
805 &self,
806 cx: &mut js::context::JSContext,
807 ) -> Fallible<Option<DomRoot<ReadableStreamBYOBRequest>>> {
808 // If controller.[[byobRequest]] is null and controller.[[pendingPullIntos]] is not empty,
809 let pending_pull_intos = self.pending_pull_intos.borrow();
810 if self.byob_request.get().is_none() && !pending_pull_intos.is_empty() {
811 // Let firstDescriptor be controller.[[pendingPullIntos]][0].
812 let first_descriptor = pending_pull_intos.first().unwrap();
813 // Let view be ! Construct(%Uint8Array%, « firstDescriptor’s buffer,
814 // firstDescriptor’s byte offset + firstDescriptor’s bytes filled,
815 // firstDescriptor’s byte length − firstDescriptor’s bytes filled »).
816
817 let byte_offset = first_descriptor.byte_offset + first_descriptor.bytes_filled.get();
818 let byte_length = first_descriptor.byte_length - first_descriptor.bytes_filled.get();
819
820 let view = create_buffer_source_with_constructor(
821 cx,
822 &Constructor::Name(Type::Uint8),
823 &first_descriptor.buffer,
824 byte_offset as usize,
825 byte_length as usize,
826 )
827 .expect("Construct Uint8Array failed");
828
829 // Let byobRequest be a new ReadableStreamBYOBRequest.
830 let byob_request = ReadableStreamBYOBRequest::new(cx, &self.global());
831
832 // Set byobRequest.[[controller]] to controller.
833 byob_request.set_controller(Some(&DomRoot::from_ref(self)));
834
835 // Set byobRequest.[[view]] to view.
836 byob_request.set_view(Some(view));
837
838 // Set controller.[[byobRequest]] to byobRequest.
839 self.byob_request.set(Some(&byob_request));
840 }
841
842 // Return controller.[[byobRequest]].
843 Ok(self.byob_request.get())
844 }
845
846 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-close>
847 pub(crate) fn close(&self, cx: &mut JSContext) -> Fallible<()> {
848 // Let stream be controller.[[stream]].
849 let stream = self.stream.get().unwrap();
850
851 // If controller.[[closeRequested]] is true or stream.[[state]] is not "readable", return.
852 if self.close_requested.get() || !stream.is_readable() {
853 return Ok(());
854 }
855
856 // If controller.[[queueTotalSize]] > 0,
857 if self.queue_total_size.get() > 0.0 {
858 // Set controller.[[closeRequested]] to true.
859 self.close_requested.set(true);
860 // Return.
861 return Ok(());
862 }
863
864 {
865 // If controller.[[pendingPullIntos]] is not empty,
866 let pending_pull_intos = self.pending_pull_intos.borrow();
867 if !pending_pull_intos.is_empty() {
868 // Let firstPendingPullInto be controller.[[pendingPullIntos]][0].
869 let first_pending_pull_into = pending_pull_intos.first().unwrap();
870
871 // If the remainder after dividing firstPendingPullInto’s bytes filled by
872 // firstPendingPullInto’s element size is not 0,
873 if !first_pending_pull_into
874 .bytes_filled
875 .get()
876 .is_multiple_of(first_pending_pull_into.element_size)
877 {
878 // needed to drop the borrow and avoid BorrowMutError
879 drop(pending_pull_intos);
880
881 // Let e be a new TypeError exception.
882 let e = Error::Type(
883 c"remainder after dividing firstPendingPullInto's bytes
884 filled by firstPendingPullInto's element size is not 0"
885 .to_owned(),
886 );
887
888 // Perform ! ReadableByteStreamControllerError(controller, e).
889 rooted!(&in(cx) let mut error = UndefinedValue());
890 e.clone().to_jsval(cx, &self.global(), error.handle_mut());
891 self.error(cx, error.handle());
892
893 // Throw e.
894 return Err(e);
895 }
896 }
897 }
898
899 // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
900 self.clear_algorithms();
901
902 // Perform ! ReadableStreamClose(stream).
903 stream.close(cx);
904 Ok(())
905 }
906
907 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-error>
908 pub(crate) fn error(&self, cx: &mut JSContext, e: SafeHandleValue) {
909 // Let stream be controller.[[stream]].
910 let stream = self.stream.get().unwrap();
911
912 // If stream.[[state]] is not "readable", return.
913 if !stream.is_readable() {
914 return;
915 }
916
917 // Perform ! ReadableByteStreamControllerClearPendingPullIntos(controller).
918 self.clear_pending_pull_intos();
919
920 // Perform ! ResetQueue(controller).
921 self.reset_queue();
922
923 // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
924 self.clear_algorithms();
925
926 // Perform ! ReadableStreamError(stream, e).
927 stream.error(cx, e);
928 }
929
930 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-clear-algorithms>
931 fn clear_algorithms(&self) {
932 // Set controller.[[pullAlgorithm]] to undefined.
933 // Set controller.[[cancelAlgorithm]] to undefined.
934 self.underlying_source.set(None);
935 }
936
937 /// <https://streams.spec.whatwg.org/#reset-queue>
938 pub(crate) fn reset_queue(&self) {
939 // Assert: container has [[queue]] and [[queueTotalSize]] internal slots.
940
941 // Set container.[[queue]] to a new empty list.
942 self.queue.borrow_mut().clear();
943
944 // Set container.[[queueTotalSize]] to 0.
945 self.queue_total_size.set(0.0);
946 }
947
948 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-clear-pending-pull-intos>
949 pub(crate) fn clear_pending_pull_intos(&self) {
950 // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
951 self.invalidate_byob_request();
952
953 // Set controller.[[pendingPullIntos]] to a new empty list.
954 self.pending_pull_intos.borrow_mut().clear();
955 }
956
957 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-invalidate-byob-request>
958 pub(crate) fn invalidate_byob_request(&self) {
959 if let Some(byob_request) = self.byob_request.get() {
960 // Set controller.[[byobRequest]].[[controller]] to undefined.
961 byob_request.set_controller(None);
962
963 // Set controller.[[byobRequest]].[[view]] to null.
964 byob_request.set_view(None);
965
966 // Set controller.[[byobRequest]] to null.
967 self.byob_request.set(None);
968 }
969 // If controller.[[byobRequest]] is null, return.
970 }
971
972 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-enqueue>
973 pub(crate) fn enqueue(
974 &self,
975 cx: &mut JSContext,
976 chunk: RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>,
977 ) -> Fallible<()> {
978 // Let stream be controller.[[stream]].
979 let stream = self.stream.get().unwrap();
980
981 // If controller.[[closeRequested]] is true or stream.[[state]] is not "readable", return.
982 if self.close_requested.get() || !stream.is_readable() {
983 return Ok(());
984 }
985
986 // Let buffer be chunk.[[ViewedArrayBuffer]].
987 let buffer = chunk.get_array_buffer_view_buffer(cx);
988
989 // Let byteOffset be chunk.[[ByteOffset]].
990 let byte_offset = chunk.get_byte_offset();
991
992 // Let byteLength be chunk.[[ByteLength]].
993 let byte_length = chunk.byte_length();
994
995 // If ! IsDetachedBuffer(buffer) is true, throw a TypeError exception.
996 if buffer.is_detached_buffer(cx) {
997 return Err(Error::Type(c"buffer is detached".to_owned()));
998 }
999
1000 // Let transferredBuffer be ? TransferArrayBuffer(buffer).
1001 let transferred_buffer = buffer.transfer_array_buffer(cx)?;
1002
1003 // If controller.[[pendingPullIntos]] is not empty,
1004
1005 let pending_pull_intos = self.pending_pull_intos.borrow();
1006 if !pending_pull_intos.is_empty() {
1007 let heap_buffer = {
1008 // Let firstPendingPullInto be controller.[[pendingPullIntos]][0].
1009 let first_descriptor = pending_pull_intos.first().unwrap();
1010 // If ! IsDetachedBuffer(firstPendingPullInto’s buffer) is true, throw a TypeError exception.
1011 if first_descriptor.buffer.is_detached_buffer(cx) {
1012 return Err(Error::Type(c"buffer is detached".to_owned()));
1013 }
1014
1015 // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
1016 self.invalidate_byob_request();
1017
1018 first_descriptor
1019 .buffer
1020 .transfer_array_buffer(cx)
1021 .expect("TransferArrayBuffer failed")
1022 };
1023
1024 drop(pending_pull_intos);
1025 // Set firstPendingPullInto’s buffer to ! TransferArrayBuffer(firstPendingPullInto’s buffer).
1026
1027 self.pending_pull_intos
1028 .safe_borrow_mut(cx)
1029 .first_mut()
1030 .unwrap()
1031 .buffer = *heap_buffer.into_box();
1032
1033 // If firstPendingPullInto’s reader type is "none",
1034 if self
1035 .pending_pull_intos
1036 .borrow()
1037 .first()
1038 .unwrap()
1039 .reader_type
1040 .is_none()
1041 {
1042 // perform ? ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(
1043 // controller, firstPendingPullInto).
1044 self.enqueue_detached_pull_into_to_queue(cx)?;
1045 }
1046 }
1047
1048 // If ! ReadableStreamHasDefaultReader(stream) is true,
1049 if stream.has_default_reader() {
1050 // Perform ! ReadableByteStreamControllerProcessReadRequestsUsingQueue(controller).
1051 self.process_read_requests_using_queue(cx)
1052 .expect("process_read_requests_using_queue failed");
1053
1054 // If ! ReadableStreamGetNumReadRequests(stream) is 0,
1055 if stream.get_num_read_requests() == 0 {
1056 // Assert: controller.[[pendingPullIntos]] is empty.
1057 {
1058 assert!(self.pending_pull_intos.borrow().is_empty());
1059 }
1060
1061 // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(
1062 // controller, transferredBuffer, byteOffset, byteLength).
1063 self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);
1064 } else {
1065 // Assert: controller.[[queue]] is empty.
1066 assert!(self.queue.borrow().is_empty());
1067
1068 // If controller.[[pendingPullIntos]] is not empty,
1069
1070 let pending_pull_intos = self.pending_pull_intos.borrow();
1071 if !pending_pull_intos.is_empty() {
1072 // Assert: controller.[[pendingPullIntos]][0]'s reader type is "default".
1073 assert!(matches!(
1074 pending_pull_intos.first().unwrap().reader_type,
1075 Some(ReaderType::Default)
1076 ));
1077
1078 // needed to drop the borrow and avoid BorrowMutError
1079 drop(pending_pull_intos);
1080
1081 // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1082 self.shift_pending_pull_into();
1083 }
1084
1085 // Let transferredView be ! Construct(%Uint8Array%, « transferredBuffer, byteOffset, byteLength »).
1086 let transferred_view = create_buffer_source_with_constructor(
1087 cx,
1088 &Constructor::Name(Type::Uint8),
1089 &transferred_buffer,
1090 byte_offset,
1091 byte_length,
1092 )
1093 .expect("Construct Uint8Array failed");
1094
1095 // Perform ! ReadableStreamFulfillReadRequest(stream, transferredView, false).
1096 rooted!(&in(cx) let mut view_value = UndefinedValue());
1097 transferred_view.get_buffer_view_value(cx, view_value.handle_mut());
1098 stream.fulfill_read_request(cx, view_value.handle(), false);
1099 }
1100 // Otherwise, if ! ReadableStreamHasBYOBReader(stream) is true,
1101 } else if stream.has_byob_reader() {
1102 // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(
1103 // controller, transferredBuffer, byteOffset, byteLength).
1104 self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);
1105
1106 // Let filledPullIntos be the result of performing !
1107 // ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
1108 rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));
1109
1110 // For each filledPullInto of filledPullIntos,
1111 // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(stream, filledPullInto).
1112 for filled_pull_into in &*filled_pull_intos {
1113 self.commit_pull_into_descriptor(cx, filled_pull_into)
1114 .expect("commit_pull_into_descriptor failed");
1115 }
1116 } else {
1117 // Assert: ! IsReadableStreamLocked(stream) is false.
1118 assert!(!stream.is_locked());
1119
1120 // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue
1121 // (controller, transferredBuffer, byteOffset, byteLength).
1122 self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);
1123 }
1124
1125 // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
1126 self.call_pull_if_needed(cx);
1127
1128 Ok(())
1129 }
1130
1131 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-commit-pull-into-descriptor>
1132 fn commit_pull_into_descriptor(
1133 &self,
1134 cx: &mut JSContext,
1135 pull_into_descriptor: &PullIntoDescriptor,
1136 ) -> Fallible<()> {
1137 // Assert: stream.[[state]] is not "errored".
1138 let stream = self.stream.get().unwrap();
1139 assert!(!stream.is_errored());
1140
1141 // Assert: pullIntoDescriptor.reader type is not "none".
1142 assert!(pull_into_descriptor.reader_type.is_some());
1143
1144 // Let done be false.
1145 let mut done = false;
1146
1147 // If stream.[[state]] is "closed",
1148 if stream.is_closed() {
1149 // Assert: the remainder after dividing pullIntoDescriptor’s bytes filled
1150 // by pullIntoDescriptor’s element size is 0.
1151 assert!(
1152 pull_into_descriptor
1153 .bytes_filled
1154 .get()
1155 .is_multiple_of(pull_into_descriptor.element_size)
1156 );
1157
1158 // Set done to true.
1159 done = true;
1160 }
1161
1162 // Let filledView be ! ReadableByteStreamControllerConvertPullIntoDescriptor(pullIntoDescriptor).
1163 let filled_view = self
1164 .convert_pull_into_descriptor(cx, pull_into_descriptor)
1165 .expect("convert_pull_into_descriptor failed");
1166
1167 rooted!(&in(cx) let mut view_value = UndefinedValue());
1168 filled_view.get_buffer_view_value(cx, view_value.handle_mut());
1169
1170 // If pullIntoDescriptor’s reader type is "default",
1171 if matches!(pull_into_descriptor.reader_type, Some(ReaderType::Default)) {
1172 // Perform ! ReadableStreamFulfillReadRequest(stream, filledView, done).
1173
1174 stream.fulfill_read_request(cx, view_value.handle(), done);
1175 } else {
1176 // Assert: pullIntoDescriptor’s reader type is "byob".
1177 assert!(matches!(
1178 pull_into_descriptor.reader_type,
1179 Some(ReaderType::Byob)
1180 ));
1181
1182 // Perform ! ReadableStreamFulfillReadIntoRequest(stream, filledView, done).
1183 stream.fulfill_read_into_request(cx, view_value.handle(), done);
1184 }
1185 Ok(())
1186 }
1187
1188 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-convert-pull-into-descriptor>
1189 pub(crate) fn convert_pull_into_descriptor(
1190 &self,
1191 cx: &mut js::context::JSContext,
1192 pull_into_descriptor: &PullIntoDescriptor,
1193 ) -> Fallible<RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>> {
1194 // Let bytesFilled be pullIntoDescriptor’s bytes filled.
1195 let bytes_filled = pull_into_descriptor.bytes_filled.get();
1196
1197 // Let elementSize be pullIntoDescriptor’s element size.
1198 let element_size = pull_into_descriptor.element_size;
1199
1200 // Assert: bytesFilled ≤ pullIntoDescriptor’s byte length.
1201 assert!(bytes_filled <= pull_into_descriptor.byte_length);
1202
1203 // Assert: the remainder after dividing bytesFilled by elementSize is 0.
1204 assert!(bytes_filled.is_multiple_of(element_size));
1205
1206 // Let buffer be ! TransferArrayBuffer(pullIntoDescriptor’s buffer).
1207 let buffer = pull_into_descriptor
1208 .buffer
1209 .transfer_array_buffer(cx)
1210 .expect("TransferArrayBuffer failed");
1211
1212 // Return ! Construct(pullIntoDescriptor’s view constructor,
1213 // « buffer, pullIntoDescriptor’s byte offset, bytesFilled ÷ elementSize »).
1214 Ok(create_buffer_source_with_constructor(
1215 cx,
1216 &pull_into_descriptor.view_constructor,
1217 &buffer,
1218 pull_into_descriptor.byte_offset as usize,
1219 (bytes_filled / element_size) as usize,
1220 )
1221 .expect("Construct view failed"))
1222 }
1223
1224 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-process-pull-into-descriptors-using-queue>
1225 pub(crate) fn process_pull_into_descriptors_using_queue(
1226 &self,
1227 cx: &mut js::context::JSContext,
1228 ) -> Vec<PullIntoDescriptor> {
1229 // Assert: controller.[[closeRequested]] is false.
1230 assert!(!self.close_requested.get());
1231
1232 // Let filledPullIntos be a new empty list.
1233 rooted!(&in(cx) let mut filled_pull_intos = Vec::new());
1234
1235 // While controller.[[pendingPullIntos]] is not empty,
1236 loop {
1237 // If controller.[[queueTotalSize]] is 0, then break.
1238 if self.queue_total_size.get() == 0.0 {
1239 break;
1240 }
1241
1242 // Let pullIntoDescriptor be controller.[[pendingPullIntos]][0].
1243 let fill_pull_result = {
1244 let pending_pull_intos = self.pending_pull_intos.borrow();
1245 let Some(pull_into_descriptor) = pending_pull_intos.first() else {
1246 break;
1247 };
1248 self.fill_pull_into_descriptor_from_queue(cx, pull_into_descriptor)
1249 };
1250
1251 // If ! ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller, pullIntoDescriptor) is true,
1252 if fill_pull_result {
1253 // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1254 // Append pullIntoDescriptor to filledPullIntos.
1255 filled_pull_intos.push(self.shift_pending_pull_into());
1256 }
1257 }
1258
1259 // Return filledPullIntos.
1260 filled_pull_intos.take()
1261 }
1262
1263 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-fill-pull-into-descriptor-from-queue>
1264 pub(crate) fn fill_pull_into_descriptor_from_queue(
1265 &self,
1266 cx: &mut js::context::JSContext,
1267 pull_into_descriptor: &PullIntoDescriptor,
1268 ) -> bool {
1269 // Let maxBytesToCopy be min(controller.[[queueTotalSize]],
1270 // pullIntoDescriptor’s byte length − pullIntoDescriptor’s bytes filled).
1271 let max_bytes_to_copy = min(
1272 self.queue_total_size.get() as usize,
1273 (pull_into_descriptor.byte_length - pull_into_descriptor.bytes_filled.get()) as usize,
1274 );
1275
1276 // Let maxBytesFilled be pullIntoDescriptor’s bytes filled + maxBytesToCopy.
1277 let max_bytes_filled = pull_into_descriptor.bytes_filled.get() as usize + max_bytes_to_copy;
1278
1279 // Let totalBytesToCopyRemaining be maxBytesToCopy.
1280 let mut total_bytes_to_copy_remaining = max_bytes_to_copy;
1281
1282 // Let ready be false.
1283 let mut ready = false;
1284
1285 // Assert: ! IsDetachedBuffer(pullIntoDescriptor’s buffer) is false.
1286 assert!(!pull_into_descriptor.buffer.is_detached_buffer(cx));
1287
1288 // Assert: pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill.
1289 assert!(pull_into_descriptor.bytes_filled.get() < pull_into_descriptor.minimum_fill);
1290
1291 // Let remainderBytes be the remainder after dividing maxBytesFilled by pullIntoDescriptor’s element size.
1292 let remainder_bytes = max_bytes_filled % pull_into_descriptor.element_size as usize;
1293
1294 // Let maxAlignedBytes be maxBytesFilled − remainderBytes.
1295 let max_aligned_bytes = max_bytes_filled - remainder_bytes;
1296
1297 // If maxAlignedBytes ≥ pullIntoDescriptor’s minimum fill,
1298 if max_aligned_bytes >= pull_into_descriptor.minimum_fill as usize {
1299 // Set totalBytesToCopyRemaining to maxAlignedBytes − pullIntoDescriptor’s bytes filled.
1300 total_bytes_to_copy_remaining =
1301 max_aligned_bytes - (pull_into_descriptor.bytes_filled.get() as usize);
1302
1303 // Set ready to true.
1304 ready = true;
1305 }
1306
1307 // Let queue be controller.[[queue]].
1308 // While totalBytesToCopyRemaining > 0,
1309 while total_bytes_to_copy_remaining > 0 {
1310 // Let headOfQueue be queue[0].
1311 let queue = self.queue.borrow();
1312 let head_of_queue = queue.front().unwrap();
1313
1314 // Let bytesToCopy be min(totalBytesToCopyRemaining, headOfQueue’s byte length).
1315 let bytes_to_copy = total_bytes_to_copy_remaining.min(head_of_queue.byte_length);
1316
1317 // Let destStart be pullIntoDescriptor’s byte offset + pullIntoDescriptor’s bytes filled.
1318 let dest_start =
1319 pull_into_descriptor.byte_offset + pull_into_descriptor.bytes_filled.get();
1320
1321 // Let descriptorBuffer be pullIntoDescriptor’s buffer.
1322 let descriptor_buffer = &pull_into_descriptor.buffer;
1323
1324 // Let queueBuffer be headOfQueue’s buffer.
1325 let queue_buffer = &head_of_queue.buffer;
1326
1327 // Let queueByteOffset be headOfQueue’s byte offset.
1328 let queue_byte_offset = head_of_queue.byte_offset;
1329
1330 // Assert: ! CanCopyDataBlockBytes(descriptorBuffer, destStart,
1331 // queueBuffer, queueByteOffset, bytesToCopy) is true.
1332 assert!(descriptor_buffer.can_copy_data_block_bytes(
1333 cx,
1334 dest_start as usize,
1335 queue_buffer,
1336 queue_byte_offset,
1337 bytes_to_copy
1338 ));
1339
1340 // Perform ! CopyDataBlockBytes(descriptorBuffer.[[ArrayBufferData]], destStart,
1341 // queueBuffer.[[ArrayBufferData]], queueByteOffset, bytesToCopy).
1342 descriptor_buffer.copy_data_block_bytes(
1343 cx,
1344 dest_start as usize,
1345 queue_buffer,
1346 queue_byte_offset,
1347 bytes_to_copy,
1348 );
1349
1350 let head_of_queue_byte_length = head_of_queue.byte_length;
1351 // Remove the borrow on self.queue
1352 drop(queue);
1353
1354 // If headOfQueue’s byte length is bytesToCopy,
1355 if head_of_queue_byte_length == bytes_to_copy {
1356 // Remove queue[0].
1357 self.queue.safe_borrow_mut(cx).pop_front().unwrap();
1358 } else {
1359 let mut queue = self.queue.safe_borrow_mut(cx);
1360 let head_of_queue = queue.front_mut().unwrap();
1361 // Set headOfQueue’s byte offset to headOfQueue’s byte offset + bytesToCopy.
1362 head_of_queue.byte_offset += bytes_to_copy;
1363
1364 // Set headOfQueue’s byte length to headOfQueue’s byte length − bytesToCopy.
1365 head_of_queue.byte_length -= bytes_to_copy;
1366 }
1367
1368 // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − bytesToCopy.
1369 self.queue_total_size
1370 .set(self.queue_total_size.get() - (bytes_to_copy as f64));
1371
1372 // Perform ! ReadableByteStreamControllerFillHeadPullIntoDescriptor(
1373 // controller, bytesToCopy, pullIntoDescriptor).
1374 self.fill_head_pull_into_descriptor(bytes_to_copy as u64, pull_into_descriptor);
1375
1376 // Set totalBytesToCopyRemaining to totalBytesToCopyRemaining − bytesToCopy.
1377 total_bytes_to_copy_remaining -= bytes_to_copy;
1378 }
1379
1380 // If ready is false,
1381 if !ready {
1382 // Assert: controller.[[queueTotalSize]] is 0.
1383 assert!(self.queue_total_size.get() == 0.0);
1384
1385 // Assert: pullIntoDescriptor’s bytes filled > 0.
1386 assert!(pull_into_descriptor.bytes_filled.get() > 0);
1387
1388 // Assert: pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill.
1389 assert!(pull_into_descriptor.bytes_filled.get() < pull_into_descriptor.minimum_fill);
1390 }
1391
1392 // Return ready.
1393 ready
1394 }
1395
1396 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-fill-head-pull-into-descriptor>
1397 pub(crate) fn fill_head_pull_into_descriptor(
1398 &self,
1399 bytes_copied: u64,
1400 pull_into_descriptor: &PullIntoDescriptor,
1401 ) {
1402 // Assert: either controller.[[pendingPullIntos]] is empty,
1403 // or controller.[[pendingPullIntos]][0] is pullIntoDescriptor.
1404 {
1405 let pending_pull_intos = self.pending_pull_intos.borrow();
1406 assert!(
1407 pending_pull_intos.is_empty() ||
1408 pending_pull_intos.first().unwrap() == pull_into_descriptor
1409 );
1410 }
1411
1412 // Assert: controller.[[byobRequest]] is null.
1413 assert!(self.byob_request.get().is_none());
1414
1415 // Set pullIntoDescriptor’s bytes filled to bytes filled + size.
1416 pull_into_descriptor
1417 .bytes_filled
1418 .set(pull_into_descriptor.bytes_filled.get() + bytes_copied);
1419 }
1420
1421 /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerenqueuedetachedpullintotoqueue>
1422 pub(crate) fn enqueue_detached_pull_into_to_queue(&self, cx: &mut JSContext) -> Fallible<()> {
1423 // first_descriptor: &PullIntoDescriptor,
1424 let pending_pull_intos = self.pending_pull_intos.borrow();
1425 let first_descriptor = pending_pull_intos.first().unwrap();
1426
1427 // Assert: pullIntoDescriptor’s reader type is "none".
1428 assert!(first_descriptor.reader_type.is_none());
1429
1430 // If pullIntoDescriptor’s bytes filled > 0, perform ?
1431 // ReadableByteStreamControllerEnqueueClonedChunkToQueue(controller,
1432 // pullIntoDescriptor’s buffer, pullIntoDescriptor’s byte offset, pullIntoDescriptor’s bytes filled).
1433
1434 if first_descriptor.bytes_filled.get() > 0 {
1435 self.enqueue_cloned_chunk_to_queue(
1436 cx,
1437 &first_descriptor.buffer,
1438 first_descriptor.byte_offset,
1439 first_descriptor.bytes_filled.get(),
1440 )?;
1441 }
1442
1443 // needed to drop the borrow and avoid BorrowMutError
1444 drop(pending_pull_intos);
1445
1446 // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1447 self.shift_pending_pull_into();
1448
1449 Ok(())
1450 }
1451
1452 /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerenqueueclonedchunktoqueue>
1453 pub(crate) fn enqueue_cloned_chunk_to_queue(
1454 &self,
1455 cx: &mut JSContext,
1456 buffer: &HeapBufferSource<ArrayBufferU8>,
1457 byte_offset: u64,
1458 byte_length: u64,
1459 ) -> Fallible<()> {
1460 // Let cloneResult be CloneArrayBuffer(buffer, byteOffset, byteLength, %ArrayBuffer%).
1461 if let Ok(clone_result) =
1462 buffer.clone_array_buffer(cx, byte_offset as usize, byte_length as usize)
1463 {
1464 // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue
1465 // (controller, cloneResult.[[Value]], 0, byteLength).
1466 self.enqueue_chunk_to_queue(clone_result, 0, byte_length as usize);
1467
1468 Ok(())
1469 } else {
1470 // If cloneResult is an abrupt completion,
1471
1472 // Perform ! ReadableByteStreamControllerError(controller, cloneResult.[[Value]]).
1473 rooted!(&in(cx) let mut rval = UndefinedValue());
1474 let error = Error::Type(c"can not clone array buffer".to_owned());
1475 error
1476 .clone()
1477 .to_jsval(cx, &self.global(), rval.handle_mut());
1478 self.error(cx, rval.handle());
1479
1480 // Return cloneResult.
1481 Err(error)
1482 }
1483 }
1484
1485 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-enqueue-chunk-to-queue>
1486 pub(crate) fn enqueue_chunk_to_queue(
1487 &self,
1488 buffer: RootedTraceableBox<HeapBufferSource<ArrayBufferU8>>,
1489 byte_offset: usize,
1490 byte_length: usize,
1491 ) {
1492 // Let entry be a new ReadableByteStreamQueueEntry object.
1493 // Append entry to controller.[[queue]].
1494 self.queue
1495 .borrow_mut()
1496 .push_back(QueueEntry::new(buffer, byte_offset, byte_length));
1497
1498 // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] + byteLength.
1499 self.queue_total_size
1500 .set(self.queue_total_size.get() + byte_length as f64);
1501 }
1502
1503 pub(crate) fn in_memory(&self) -> bool {
1504 let Some(underlying_source) = self.underlying_source.get() else {
1505 return false;
1506 };
1507 underlying_source.in_memory()
1508 }
1509
1510 pub(crate) fn get_in_memory_bytes(&self, cx: &mut JSContext) -> Option<Vec<u8>> {
1511 let underlying_source = self.underlying_source.get()?;
1512 if !underlying_source.in_memory() {
1513 return None;
1514 }
1515
1516 self.queue.borrow().iter().try_fold(
1517 Vec::with_capacity(self.queue_total_size.get() as usize),
1518 |mut bytes, entry| {
1519 let mut chunk = vec![0; entry.byte_length];
1520 entry
1521 .buffer
1522 .copy_data_to(
1523 cx,
1524 &mut chunk,
1525 entry.byte_offset,
1526 entry.byte_offset + entry.byte_length,
1527 )
1528 .ok()?;
1529 bytes.extend(chunk);
1530 Some(bytes)
1531 },
1532 )
1533 }
1534
1535 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-shift-pending-pull-into>
1536 pub(crate) fn shift_pending_pull_into(&self) -> PullIntoDescriptor {
1537 // Assert: controller.[[byobRequest]] is null.
1538 assert!(self.byob_request.get().is_none());
1539
1540 // Let descriptor be controller.[[pendingPullIntos]][0].
1541 // Remove descriptor from controller.[[pendingPullIntos]].
1542 // Return descriptor.
1543 self.pending_pull_intos.borrow_mut().remove(0)
1544 }
1545
1546 /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerprocessreadrequestsusingqueue>
1547 pub(crate) fn process_read_requests_using_queue(&self, cx: &mut JSContext) -> Fallible<()> {
1548 // Let reader be controller.[[stream]].[[reader]].
1549 // Assert: reader implements ReadableStreamDefaultReader.
1550 let reader = self.stream.get().unwrap().get_default_reader();
1551
1552 // Step 3
1553 reader.process_read_requests(cx, self)
1554 }
1555
1556 /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerfillreadrequestfromqueue>
1557 pub(crate) fn fill_read_request_from_queue(
1558 &self,
1559 cx: &mut JSContext,
1560 read_request: &ReadRequest,
1561 ) -> Fallible<()> {
1562 // Assert: controller.[[queueTotalSize]] > 0.
1563 assert!(self.queue_total_size.get() > 0.0);
1564 // Also assert that the queue has a non-zero length;
1565 assert!(!self.queue.borrow().is_empty());
1566
1567 // Let entry be controller.[[queue]][0].
1568 // Remove entry from controller.[[queue]].
1569 rooted!(&in(cx) let entry = self.remove_entry());
1570
1571 // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − entry’s byte length.
1572 self.queue_total_size
1573 .set(self.queue_total_size.get() - entry.byte_length as f64);
1574
1575 // Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
1576 self.handle_queue_drain(cx);
1577
1578 // Let view be ! Construct(%Uint8Array%, « entry’s buffer, entry’s byte offset, entry’s byte length »).
1579 let view = create_buffer_source_with_constructor(
1580 cx,
1581 &Constructor::Name(Type::Uint8),
1582 &entry.buffer,
1583 entry.byte_offset,
1584 entry.byte_length,
1585 )
1586 .expect("Construct Uint8Array failed");
1587
1588 // Perform readRequest’s chunk steps, given view.
1589 let result = RootedTraceableBox::new(Heap::default());
1590 rooted!(&in(cx) let mut view_value = UndefinedValue());
1591 view.get_buffer_view_value(cx, view_value.handle_mut());
1592 result.set(*view_value);
1593
1594 read_request.chunk_steps(cx, result, &self.global());
1595
1596 Ok(())
1597 }
1598
1599 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-handle-queue-drain>
1600 pub(crate) fn handle_queue_drain(&self, cx: &mut JSContext) {
1601 // Assert: controller.[[stream]].[[state]] is "readable".
1602 assert!(self.stream.get().unwrap().is_readable());
1603
1604 // If controller.[[queueTotalSize]] is 0 and controller.[[closeRequested]] is true,
1605 if self.queue_total_size.get() == 0.0 && self.close_requested.get() {
1606 // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
1607 self.clear_algorithms();
1608
1609 // Perform ! ReadableStreamClose(controller.[[stream]]).
1610 self.stream.get().unwrap().close(cx);
1611 } else {
1612 // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
1613 self.call_pull_if_needed(cx);
1614 }
1615 }
1616
1617 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
1618 fn call_pull_if_needed(&self, cx: &mut JSContext) {
1619 // Let shouldPull be ! ReadableByteStreamControllerShouldCallPull(controller).
1620 let should_pull = self.should_call_pull();
1621 // If shouldPull is false, return.
1622 if !should_pull {
1623 return;
1624 }
1625
1626 // If controller.[[pulling]] is true,
1627 if self.pulling.get() {
1628 // Set controller.[[pullAgain]] to true.
1629 self.pull_again.set(true);
1630
1631 // Return.
1632 return;
1633 }
1634
1635 // Assert: controller.[[pullAgain]] is false.
1636 assert!(!self.pull_again.get());
1637
1638 // Set controller.[[pulling]] to true.
1639 self.pulling.set(true);
1640
1641 // Let pullPromise be the result of performing controller.[[pullAlgorithm]].
1642 // Continues into the resolve and reject handling of the native handler.
1643 let global = self.global();
1644 let rooted_controller = DomRoot::from_ref(self);
1645 let controller = Controller::ReadableByteStreamController(rooted_controller.clone());
1646
1647 if let Some(underlying_source) = self.underlying_source.get() {
1648 let handler = PromiseNativeHandler::new(
1649 cx,
1650 &global,
1651 Some(Box::new(PullAlgorithmFulfillmentHandler {
1652 controller: Dom::from_ref(&rooted_controller),
1653 })),
1654 Some(Box::new(PullAlgorithmRejectionHandler {
1655 controller: Dom::from_ref(&rooted_controller),
1656 })),
1657 );
1658
1659 let mut realm = enter_auto_realm(cx, &*global);
1660 let cx = &mut realm.current_realm();
1661
1662 let result = underlying_source
1663 .call_pull_algorithm(cx, controller)
1664 .unwrap_or_else(|| {
1665 let promise = Promise::new_resolved(cx, &global, ());
1666 Ok(promise)
1667 });
1668 let promise = result.unwrap_or_else(|error| {
1669 rooted!(&in(cx) let mut rval = UndefinedValue());
1670 // TODO: check if `self.global()` is the right globalscope.
1671 error.to_jsval(cx, &global, rval.handle_mut());
1672 Promise::new_rejected(cx, &global, rval.handle())
1673 });
1674 promise.append_native_handler(cx, &handler);
1675 }
1676 }
1677
1678 /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-should-call-pull>
1679 fn should_call_pull(&self) -> bool {
1680 // Let stream be controller.[[stream]].
1681 // Note: the spec does not assert that stream is not undefined here,
1682 // so we return false if it is.
1683 let stream = self.stream.get().unwrap();
1684
1685 // If stream.[[state]] is not "readable", return false.
1686 if !stream.is_readable() {
1687 return false;
1688 }
1689
1690 // If controller.[[closeRequested]] is true, return false.
1691 if self.close_requested.get() {
1692 return false;
1693 }
1694
1695 // If controller.[[started]] is false, return false.
1696 if !self.started.get() {
1697 return false;
1698 }
1699
1700 // If ! ReadableStreamHasDefaultReader(stream) is true and ! ReadableStreamGetNumReadRequests(stream) > 0
1701 // , return true.
1702 if stream.has_default_reader() && stream.get_num_read_requests() > 0 {
1703 return true;
1704 }
1705
1706 // If ! ReadableStreamHasBYOBReader(stream) is true and ! ReadableStreamGetNumReadIntoRequests(stream) > 0
1707 // , return true.
1708 if stream.has_byob_reader() && stream.get_num_read_into_requests() > 0 {
1709 return true;
1710 }
1711
1712 // Let desiredSize be ! ReadableByteStreamControllerGetDesiredSize(controller).
1713 let desired_size = self.get_desired_size();
1714
1715 // Assert: desiredSize is not null.
1716 assert!(desired_size.is_some());
1717
1718 // If desiredSize > 0, return true.
1719 if desired_size.unwrap() > 0. {
1720 return true;
1721 }
1722
1723 // Return false.
1724 false
1725 }
1726 /// <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
1727 pub(crate) fn setup(
1728 &self,
1729 cx: &mut JSContext,
1730 global: &GlobalScope,
1731 stream: &ReadableStream,
1732 ) -> Fallible<()> {
1733 // Assert: stream.[[controller]] is undefined.
1734 stream.assert_no_controller();
1735
1736 // If autoAllocateChunkSize is not undefined,
1737 if self.auto_allocate_chunk_size.is_some() {
1738 // Assert: ! IsInteger(autoAllocateChunkSize) is true. Implicit
1739 // Assert: autoAllocateChunkSize is positive. (Implicit by type.)
1740 }
1741
1742 // Set controller.[[stream]] to stream.
1743 self.stream.set(Some(stream));
1744
1745 // Set controller.[[pullAgain]] and controller.[[pulling]] to false.
1746 self.pull_again.set(false);
1747 self.pulling.set(false);
1748
1749 // Set controller.[[byobRequest]] to null.
1750 self.byob_request.set(None);
1751
1752 // Perform ! ResetQueue(controller).
1753 self.reset_queue();
1754
1755 // Set controller.[[closeRequested]] and controller.[[started]] to false.
1756 self.close_requested.set(false);
1757 self.started.set(false);
1758
1759 // Set controller.[[strategyHWM]] to highWaterMark.
1760 // Set controller.[[pullAlgorithm]] to pullAlgorithm.
1761 // Set controller.[[cancelAlgorithm]] to cancelAlgorithm.
1762 // Set controller.[[autoAllocateChunkSize]] to autoAllocateChunkSize.
1763 // Set controller.[[pendingPullIntos]] to a new empty list.
1764 // Note: the above steps are done in `new`.
1765
1766 // Set stream.[[controller]] to controller.
1767 let rooted_byte_controller = DomRoot::from_ref(self);
1768 stream.set_byte_controller(&rooted_byte_controller);
1769
1770 if let Some(underlying_source) = rooted_byte_controller.underlying_source.get() {
1771 // Let startResult be the result of performing startAlgorithm. (This might throw an exception.)
1772 let start_result = underlying_source
1773 .call_start_algorithm(
1774 cx,
1775 Controller::ReadableByteStreamController(rooted_byte_controller.clone()),
1776 )
1777 .unwrap_or_else(|| Ok(Promise::new_resolved(cx, global, ())));
1778
1779 // Let startPromise be a promise resolved with startResult.
1780 let start_promise = start_result?;
1781
1782 // Upon fulfillment of startPromise, Upon rejection of startPromise with reason r,
1783 let handler = PromiseNativeHandler::new(
1784 cx,
1785 global,
1786 Some(Box::new(StartAlgorithmFulfillmentHandler {
1787 controller: Dom::from_ref(&rooted_byte_controller),
1788 })),
1789 Some(Box::new(StartAlgorithmRejectionHandler {
1790 controller: Dom::from_ref(&rooted_byte_controller),
1791 })),
1792 );
1793 let mut realm = enter_auto_realm(cx, global);
1794 let cx = &mut realm.current_realm();
1795 start_promise.append_native_handler(cx, &handler);
1796 };
1797
1798 Ok(())
1799 }
1800
1801 // <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontroller-releasesteps
1802 pub(crate) fn perform_release_steps(&self) -> Fallible<()> {
1803 // If this.[[pendingPullIntos]] is not empty,
1804 let mut pending_pull_intos = self.pending_pull_intos.borrow_mut();
1805 if !pending_pull_intos.is_empty() {
1806 // Let firstPendingPullInto be this.[[pendingPullIntos]][0].
1807 let mut first_pending_pull_into = RootedTraceableBox::new(pending_pull_intos.remove(0));
1808
1809 // Set firstPendingPullInto’s reader type to "none".
1810 first_pending_pull_into.reader_type = None;
1811
1812 // Set this.[[pendingPullIntos]] to the list « firstPendingPullInto »
1813 pending_pull_intos.clear();
1814 pending_pull_intos.push(*first_pending_pull_into.into_box());
1815 }
1816 Ok(())
1817 }
1818
1819 /// <https://streams.spec.whatwg.org/#rbs-controller-private-cancel>
1820 pub(crate) fn perform_cancel_steps(
1821 &self,
1822 cx: &mut JSContext,
1823 global: &GlobalScope,
1824 reason: SafeHandleValue,
1825 ) -> Rc<Promise> {
1826 // Perform ! ReadableByteStreamControllerClearPendingPullIntos(this).
1827 self.clear_pending_pull_intos();
1828
1829 // Perform ! ResetQueue(this).
1830 self.reset_queue();
1831
1832 let underlying_source = self
1833 .underlying_source
1834 .get()
1835 .expect("Controller should have a source when the cancel steps are called into.");
1836
1837 // Let result be the result of performing this.[[cancelAlgorithm]], passing in reason.
1838 let result = underlying_source
1839 .call_cancel_algorithm(cx, global, reason)
1840 .unwrap_or_else(|| {
1841 let promise = Promise::new(cx, global);
1842 promise.resolve_native(cx, &());
1843 Ok(promise)
1844 });
1845
1846 let promise = result.unwrap_or_else(|error| {
1847 rooted!(&in(cx) let mut rval = UndefinedValue());
1848 error.to_jsval(cx, global, rval.handle_mut());
1849 let promise = Promise::new(cx, global);
1850 promise.reject_native(cx, &rval.handle());
1851 promise
1852 });
1853
1854 // Perform ! ReadableByteStreamControllerClearAlgorithms(this).
1855 self.clear_algorithms();
1856
1857 // Return result(the promise).
1858 promise
1859 }
1860
1861 /// <https://streams.spec.whatwg.org/#rbs-controller-private-pull>
1862 pub(crate) fn perform_pull_steps(&self, cx: &mut JSContext, read_request: &ReadRequest) {
1863 // Let stream be this.[[stream]].
1864 let stream = self.stream.get().unwrap();
1865
1866 // Assert: ! ReadableStreamHasDefaultReader(stream) is true.
1867 assert!(stream.has_default_reader());
1868
1869 // If this.[[queueTotalSize]] > 0,
1870 if self.queue_total_size.get() > 0.0 {
1871 // Assert: ! ReadableStreamGetNumReadRequests(stream) is 0.
1872 assert_eq!(stream.get_num_read_requests(), 0);
1873
1874 // Perform ! ReadableByteStreamControllerFillReadRequestFromQueue(this, readRequest).
1875 let _ = self.fill_read_request_from_queue(cx, read_request);
1876
1877 // Return.
1878 return;
1879 }
1880
1881 // Let autoAllocateChunkSize be this.[[autoAllocateChunkSize]].
1882 let auto_allocate_chunk_size = self.auto_allocate_chunk_size;
1883
1884 // If autoAllocateChunkSize is not undefined,
1885 if let Some(auto_allocate_chunk_size) = auto_allocate_chunk_size {
1886 // create_array_buffer_with_size
1887 // Let buffer be Construct(%ArrayBuffer%, « autoAllocateChunkSize »).
1888 match create_array_buffer_with_size(cx, auto_allocate_chunk_size as usize) {
1889 Ok(buffer) => {
1890 // Let pullIntoDescriptor be a new pull-into descriptor with
1891 // buffer buffer.[[Value]]
1892 // buffer byte length autoAllocateChunkSize
1893 // byte offset 0
1894 // byte length autoAllocateChunkSize
1895 // bytes filled 0
1896 // minimum fill 1
1897 // element size 1
1898 // view constructor %Uint8Array%
1899 // reader type "default"
1900
1901 // Append pullIntoDescriptor to this.[[pendingPullIntos]].
1902 self.pending_pull_intos
1903 .safe_borrow_mut(cx)
1904 .push(PullIntoDescriptor {
1905 buffer: *buffer.into_box(),
1906 buffer_byte_length: auto_allocate_chunk_size,
1907 byte_length: auto_allocate_chunk_size,
1908 byte_offset: 0,
1909 bytes_filled: Cell::new(0),
1910 minimum_fill: 1,
1911 element_size: 1,
1912 view_constructor: Constructor::Name(Type::Uint8),
1913 reader_type: Some(ReaderType::Default),
1914 });
1915 },
1916 Err(error) => {
1917 // If buffer is an abrupt completion,
1918 // Perform readRequest’s error steps, given buffer.[[Value]].
1919
1920 rooted!(&in(cx) let mut rval = UndefinedValue());
1921 error.to_jsval(cx, &self.global(), rval.handle_mut());
1922 read_request.error_steps(cx, rval.handle());
1923
1924 // Return.
1925 return;
1926 },
1927 }
1928 }
1929
1930 // Perform ! ReadableStreamAddReadRequest(stream, readRequest).
1931 stream.add_read_request(read_request);
1932
1933 // Perform ! ReadableByteStreamControllerCallPullIfNeeded(this).
1934 self.call_pull_if_needed(cx);
1935 }
1936
1937 /// Setting the JS object after the heap has settled down.
1938 pub(crate) fn set_underlying_source_this_object(&self, this_object: HandleObject) {
1939 if let Some(underlying_source) = self.underlying_source.get() {
1940 underlying_source.set_underlying_source_this_object(this_object);
1941 }
1942 }
1943
1944 pub(crate) fn remove_entry(&self) -> QueueEntry {
1945 self.queue
1946 .borrow_mut()
1947 .pop_front()
1948 .expect("Reader must have read request when remove is called into.")
1949 }
1950
1951 pub(crate) fn get_queue_total_size(&self) -> f64 {
1952 self.queue_total_size.get()
1953 }
1954
1955 pub(crate) fn get_pending_pull_intos_size(&self) -> usize {
1956 self.pending_pull_intos.borrow().len()
1957 }
1958}
1959
1960impl ReadableByteStreamControllerMethods<crate::DomTypeHolder> for ReadableByteStreamController {
1961 /// <https://streams.spec.whatwg.org/#rbs-controller-byob-request>
1962 fn GetByobRequest(
1963 &self,
1964 cx: &mut js::context::JSContext,
1965 ) -> Fallible<Option<DomRoot<ReadableStreamBYOBRequest>>> {
1966 // Return ! ReadableByteStreamControllerGetBYOBRequest(this).
1967 self.get_byob_request(cx)
1968 }
1969
1970 /// <https://streams.spec.whatwg.org/#rbs-controller-desired-size>
1971 fn GetDesiredSize(&self) -> Option<f64> {
1972 // Return ! ReadableByteStreamControllerGetDesiredSize(this).
1973 self.get_desired_size()
1974 }
1975
1976 /// <https://streams.spec.whatwg.org/#rbs-controller-close>
1977 fn Close(&self, cx: &mut JSContext) -> Fallible<()> {
1978 // If this.[[closeRequested]] is true, throw a TypeError exception.
1979 if self.close_requested.get() {
1980 return Err(Error::Type(c"closeRequested is true".to_owned()));
1981 }
1982
1983 // If this.[[stream]].[[state]] is not "readable", throw a TypeError exception.
1984 if !self.stream.get().unwrap().is_readable() {
1985 return Err(Error::Type(c"stream is not readable".to_owned()));
1986 }
1987
1988 // Perform ? ReadableByteStreamControllerClose(this).
1989 self.close(cx)
1990 }
1991
1992 /// <https://streams.spec.whatwg.org/#rbs-controller-enqueue>
1993 fn Enqueue(
1994 &self,
1995 cx: &mut JSContext,
1996 chunk: js::gc::CustomAutoRooterGuard<js::typedarray::ArrayBufferView>,
1997 ) -> Fallible<()> {
1998 let chunk = HeapBufferSource::<ArrayBufferViewU8>::from_view(cx, chunk);
1999
2000 // If chunk.[[ByteLength]] is 0, throw a TypeError exception.
2001 if chunk.byte_length() == 0 {
2002 return Err(Error::Type(c"chunk.ByteLength is 0".to_owned()));
2003 }
2004
2005 // If chunk.[[ViewedArrayBuffer]].[[ByteLength]] is 0, throw a TypeError exception.
2006 if chunk.viewed_buffer_array_byte_length(cx) == 0 {
2007 return Err(Error::Type(
2008 c"chunk.ViewedArrayBuffer.ByteLength is 0".to_owned(),
2009 ));
2010 }
2011
2012 // If this.[[closeRequested]] is true, throw a TypeError exception.
2013 if self.close_requested.get() {
2014 return Err(Error::Type(c"closeRequested is true".to_owned()));
2015 }
2016
2017 // If this.[[stream]].[[state]] is not "readable", throw a TypeError exception.
2018 if !self.stream.get().unwrap().is_readable() {
2019 return Err(Error::Type(c"stream is not readable".to_owned()));
2020 }
2021
2022 // Return ? ReadableByteStreamControllerEnqueue(this, chunk).
2023 self.enqueue(cx, chunk)
2024 }
2025
2026 /// <https://streams.spec.whatwg.org/#rbs-controller-error>
2027 fn Error(&self, cx: &mut JSContext, e: SafeHandleValue) -> Fallible<()> {
2028 // Perform ! ReadableByteStreamControllerError(this, e).
2029 self.error(cx, e);
2030 Ok(())
2031 }
2032}