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