MIRROR: javascript for 🐜's, a tiny runtime with big ambitions
improve ReadableStream
+112
-49
+3
include/streams/readable.h
+3
include/streams/readable.h
+109
-49
src/streams/readable.c
+109
-49
src/streams/readable.c
···
42
42
ant_extra_slot_t *entries = (ant_extra_slot_t *)obj->extra_slots;
43
43
for (uint8_t i = 0; i < obj->extra_count; i++) {
44
44
if (entries[i].slot == SLOT_DATA && vtype(entries[i].value) == T_NUM) {
45
-
free((rs_controller_t *)(uintptr_t)(size_t)js_getnum(entries[i].value));
45
+
rs_controller_t *ctrl = (rs_controller_t *)(uintptr_t)(size_t)js_getnum(entries[i].value);
46
+
free(ctrl->queue_sizes);
47
+
free(ctrl);
46
48
return;
47
49
}}
48
50
}
49
51
50
-
static ant_value_t rs_stream_controller(ant_t *js, ant_value_t stream_obj) {
52
+
static inline ant_value_t rs_stream_controller(ant_t *js, ant_value_t stream_obj) {
51
53
return js_get_slot(stream_obj, SLOT_ENTRIES);
52
54
}
53
55
54
-
static ant_value_t rs_stream_reader(ant_value_t stream_obj) {
56
+
static inline ant_value_t rs_stream_reader(ant_value_t stream_obj) {
55
57
return js_get_slot(stream_obj, SLOT_CTOR);
56
58
}
57
59
58
-
static ant_value_t rs_stream_error(ant_value_t stream_obj) {
60
+
static inline ant_value_t rs_stream_error(ant_value_t stream_obj) {
59
61
return js_get_slot(stream_obj, SLOT_BUFFER);
60
62
}
61
63
62
-
static ant_value_t rs_ctrl_stream(ant_value_t ctrl_obj) {
64
+
static inline ant_value_t rs_ctrl_stream(ant_value_t ctrl_obj) {
63
65
return js_get_slot(ctrl_obj, SLOT_ENTRIES);
64
66
}
65
67
66
-
static ant_value_t rs_ctrl_pull(ant_value_t ctrl_obj) {
68
+
static inline ant_value_t rs_ctrl_pull(ant_value_t ctrl_obj) {
67
69
return js_get_slot(ctrl_obj, SLOT_RS_PULL);
68
70
}
69
71
70
-
static ant_value_t rs_ctrl_cancel(ant_value_t ctrl_obj) {
72
+
static inline ant_value_t rs_ctrl_cancel(ant_value_t ctrl_obj) {
71
73
return js_get_slot(ctrl_obj, SLOT_RS_CANCEL);
72
74
}
73
75
74
-
static ant_value_t rs_ctrl_size(ant_value_t ctrl_obj) {
76
+
static inline ant_value_t rs_ctrl_size(ant_value_t ctrl_obj) {
75
77
return js_get_slot(ctrl_obj, SLOT_RS_SIZE);
76
78
}
77
79
78
-
static ant_value_t rs_ctrl_queue(ant_t *js, ant_value_t ctrl_obj) {
80
+
static inline ant_value_t rs_ctrl_queue(ant_t *js, ant_value_t ctrl_obj) {
79
81
return js_get_slot(ctrl_obj, SLOT_BUFFER);
80
82
}
81
83
82
-
static ant_value_t rs_reader_stream(ant_value_t reader_obj) {
84
+
static inline ant_value_t rs_reader_stream(ant_value_t reader_obj) {
83
85
return js_get_slot(reader_obj, SLOT_ENTRIES);
84
86
}
85
87
86
-
static ant_value_t rs_reader_closed(ant_value_t reader_obj) {
88
+
static inline ant_value_t rs_reader_closed(ant_value_t reader_obj) {
87
89
return js_get_slot(reader_obj, SLOT_RS_CLOSED);
88
90
}
89
91
90
-
static ant_value_t rs_reader_reqs(ant_value_t reader_obj) {
92
+
static inline ant_value_t rs_reader_reqs(ant_value_t reader_obj) {
91
93
return js_get_slot(reader_obj, SLOT_BUFFER);
92
94
}
93
95
···
217
219
218
220
static void readable_stream_error(ant_t *js, ant_value_t stream_obj, ant_value_t e) {
219
221
rs_stream_t *stream = rs_get_stream(stream_obj);
220
-
if (!stream) return;
222
+
if (!stream || stream->state != RS_STATE_READABLE) return;
221
223
stream->state = RS_STATE_ERRORED;
222
224
js_set_slot(stream_obj, SLOT_BUFFER, e);
223
225
···
226
228
js_reject_promise(js, rs_reader_closed(reader_obj), e);
227
229
rs_default_reader_error_read_requests(js, reader_obj, e);
228
230
}
231
+
}
232
+
233
+
static ant_value_t rs_cancel_resolve(ant_t *js, ant_value_t *args, int nargs) {
234
+
ant_value_t p = js_get_slot(js->current_func, SLOT_DATA);
235
+
js_resolve_promise(js, p, js_mkundef());
236
+
return js_mkundef();
237
+
}
238
+
239
+
static ant_value_t rs_cancel_reject(ant_t *js, ant_value_t *args, int nargs) {
240
+
ant_value_t p = js_get_slot(js->current_func, SLOT_DATA);
241
+
js_reject_promise(js, p, nargs > 0 ? args[0] : js_mkundef());
242
+
return js_mkundef();
229
243
}
230
244
231
245
static ant_value_t readable_stream_cancel(ant_t *js, ant_value_t stream_obj, ant_value_t reason) {
···
250
264
ant_value_t cancel_fn = rs_ctrl_cancel(ctrl_obj);
251
265
ant_value_t result = js_mkundef();
252
266
if (is_callable(cancel_fn)) {
253
-
ant_value_t args[1] = { reason };
254
-
result = sv_vm_call(js->vm, js, cancel_fn, ctrl_obj, args, 1, NULL, false);
267
+
ant_value_t cancel_args[1] = { reason };
268
+
result = sv_vm_call(js->vm, js, cancel_fn, ctrl_obj, cancel_args, 1, NULL, false);
255
269
}
270
+
256
271
rs_default_controller_clear_algorithms(ctrl_obj);
257
-
258
272
ant_value_t p = js_mkpromise(js);
259
-
if (is_err(result))
260
-
js_reject_promise(js, p, result);
261
-
else
262
-
js_resolve_promise(js, p, js_mkundef());
273
+
274
+
if (is_err(result)) {
275
+
ant_value_t thrown = js->thrown_value;
276
+
js_reject_promise(js, p, is_object_type(thrown) ? thrown : result);
277
+
} else if (vtype(result) == T_PROMISE) {
278
+
ant_value_t res_fn = js_heavy_mkfun(js, rs_cancel_resolve, p);
279
+
ant_value_t rej_fn = js_heavy_mkfun(js, rs_cancel_reject, p);
280
+
ant_value_t then_fn = js_get(js, result, "then");
281
+
if (is_callable(then_fn)) {
282
+
ant_value_t then_args[2] = { res_fn, rej_fn };
283
+
sv_vm_call(js->vm, js, then_fn, result, then_args, 2, NULL, false);
284
+
}
285
+
} else js_resolve_promise(js, p, js_mkundef());
286
+
263
287
return p;
264
288
}
265
289
···
352
376
rs_controller_t *ctrl = rs_get_controller(ctrl_obj);
353
377
if (ctrl && rs_ctrl_queue_len(js, ctrl_obj) > 0) {
354
378
ant_value_t chunk = rs_ctrl_queue_shift(js, ctrl_obj);
355
-
ctrl->queue_total_size -= 1;
379
+
double chunk_size = 1;
380
+
if (ctrl->queue_sizes_len > 0) {
381
+
chunk_size = ctrl->queue_sizes[0];
382
+
ctrl->queue_sizes_len--;
383
+
memmove(ctrl->queue_sizes, ctrl->queue_sizes + 1, ctrl->queue_sizes_len * sizeof(double));
384
+
}
385
+
ctrl->queue_total_size -= chunk_size;
356
386
if (ctrl->queue_total_size < 0) ctrl->queue_total_size = 0;
357
-
if (ctrl->close_requested && rs_ctrl_queue_len(js, ctrl_obj) == 0) {
358
-
rs_default_controller_clear_algorithms(ctrl_obj);
359
-
readable_stream_close(js, stream_obj);
360
-
} else {
361
-
rs_default_controller_call_pull_if_needed(js, ctrl_obj);
362
-
}
387
+
bool should_close = ctrl->close_requested && rs_ctrl_queue_len(js, ctrl_obj) == 0;
363
388
ant_value_t p = js_mkpromise(js);
364
389
js_resolve_promise(js, p, js_iter_result(js, true, chunk));
390
+
if (should_close) {
391
+
rs_default_controller_clear_algorithms(ctrl_obj);
392
+
readable_stream_close(js, stream_obj);
393
+
} else rs_default_controller_call_pull_if_needed(js, ctrl_obj);
394
+
365
395
return p;
366
396
}
367
397
···
418
448
ant_value_t size_args[1] = { chunk };
419
449
ant_value_t size_result = sv_vm_call(js->vm, js, size_fn, js_mkundef(), size_args, 1, NULL, false);
420
450
if (is_err(size_result)) {
421
-
readable_stream_error(js, stream_obj, size_result);
451
+
ant_value_t thrown = js->thrown_value;
452
+
ant_value_t err = is_object_type(thrown) ? thrown : size_result;
453
+
readable_stream_error(js, stream_obj, err);
454
+
if (is_object_type(thrown)) return js_throw(js, thrown);
422
455
return size_result;
423
456
}
424
457
if (vtype(size_result) == T_NUM) chunk_size = js_getnum(size_result);
···
426
459
}
427
460
428
461
if (chunk_size < 0 || chunk_size != chunk_size || chunk_size == (double)INFINITY) {
429
-
ant_value_t err = js_mkerr_typed(js, JS_ERR_RANGE,
462
+
js_mkerr_typed(js, JS_ERR_RANGE,
430
463
"The return value of a queuing strategy's size function must be a finite, non-NaN, non-negative number");
464
+
ant_value_t err = is_object_type(js->thrown_value) ? js->thrown_value : js_mkundef();
431
465
readable_stream_error(js, stream_obj, err);
432
-
return err;
466
+
return js_throw(js, err);
433
467
}
434
468
435
469
rs_ctrl_queue_push(js, js->this_val, chunk);
470
+
if (ctrl->queue_sizes_len >= ctrl->queue_sizes_cap) {
471
+
uint32_t new_cap = ctrl->queue_sizes_cap ? ctrl->queue_sizes_cap * 2 : 8;
472
+
double *ns = realloc(ctrl->queue_sizes, new_cap * sizeof(double));
473
+
if (ns) { ctrl->queue_sizes = ns; ctrl->queue_sizes_cap = new_cap; }
474
+
}
475
+
if (ctrl->queue_sizes_len < ctrl->queue_sizes_cap)
476
+
ctrl->queue_sizes[ctrl->queue_sizes_len++] = chunk_size;
436
477
ctrl->queue_total_size += chunk_size;
437
478
rs_default_controller_call_pull_if_needed(js, js->this_val);
438
479
return js_mkundef();
···
444
485
ant_value_t stream_obj = rs_ctrl_stream(js->this_val);
445
486
rs_stream_t *stream = rs_get_stream(stream_obj);
446
487
if (!stream || stream->state != RS_STATE_READABLE) return js_mkundef();
488
+
447
489
ant_value_t e = (nargs > 0) ? args[0] : js_mkundef();
448
490
ctrl->queue_total_size = 0;
491
+
ctrl->queue_sizes_len = 0;
449
492
rs_default_controller_clear_algorithms(js->this_val);
450
493
readable_stream_error(js, stream_obj, e);
494
+
451
495
return js_mkundef();
452
496
}
453
497
···
469
513
if (!stream) return js_mkundef();
470
514
471
515
if (rs_reader_has_reqs(js, js->this_val)) {
472
-
ant_value_t err = js_mkerr_typed(js, JS_ERR_TYPE, "Reader was released");
473
-
rs_default_reader_error_read_requests(js, js->this_val, err);
516
+
js_mkerr_typed(js, JS_ERR_TYPE, "Reader was released");
517
+
rs_default_reader_error_read_requests(js, js->this_val, js->thrown_value);
474
518
}
475
519
476
-
if (stream->state == RS_STATE_READABLE) {
477
-
ant_value_t err = js_mkerr_typed(js, JS_ERR_TYPE, "Reader was released while stream is readable");
478
-
js_reject_promise(js, rs_reader_closed(js->this_val), err);
479
-
}
520
+
ant_value_t new_closed = js_mkpromise(js);
521
+
js_mkerr_typed(js, JS_ERR_TYPE, "Reader was released");
522
+
ant_value_t release_err = js->thrown_value;
523
+
524
+
if (stream->state == RS_STATE_READABLE)
525
+
js_reject_promise(js, rs_reader_closed(js->this_val), release_err);
526
+
527
+
js_reject_promise(js, new_closed, release_err);
528
+
js_set_slot(js->this_val, SLOT_RS_CLOSED, new_closed);
480
529
481
530
js_set_slot(stream_obj, SLOT_CTOR, js_mkundef());
482
531
js_set_slot(js->this_val, SLOT_ENTRIES, js_mkundef());
···
485
534
486
535
static ant_value_t js_rs_reader_cancel(ant_t *js, ant_value_t *args, int nargs) {
487
536
ant_value_t stream_obj = rs_reader_stream(js->this_val);
488
-
if (!is_object_type(stream_obj))
489
-
return js_mkerr_typed(js, JS_ERR_TYPE, "Cannot cancel a released reader");
537
+
if (!is_object_type(stream_obj)) {
538
+
ant_value_t p = js_mkpromise(js);
539
+
js_mkerr_typed(js, JS_ERR_TYPE, "Cannot cancel a released reader");
540
+
js_reject_promise(js, p, js->thrown_value);
541
+
return p;
542
+
}
490
543
ant_value_t reason = (nargs > 0) ? args[0] : js_mkundef();
491
544
return readable_stream_cancel(js, stream_obj, reason);
492
545
}
···
530
583
static ant_value_t js_rs_cancel(ant_t *js, ant_value_t *args, int nargs) {
531
584
rs_stream_t *stream = rs_get_stream(js->this_val);
532
585
if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream");
533
-
if (is_object_type(rs_stream_reader(js->this_val)))
534
-
return js_mkerr_typed(js, JS_ERR_TYPE, "Cannot cancel a locked ReadableStream");
586
+
if (is_object_type(rs_stream_reader(js->this_val))) {
587
+
ant_value_t p = js_mkpromise(js);
588
+
js_mkerr_typed(js, JS_ERR_TYPE, "Cannot cancel a locked ReadableStream");
589
+
js_reject_promise(js, p, js->thrown_value);
590
+
return p;
591
+
}
535
592
ant_value_t reason = (nargs > 0) ? args[0] : js_mkundef();
536
593
return readable_stream_cancel(js, js->this_val, reason);
537
594
}
···
543
600
if (nargs > 0 && is_object_type(args[0])) {
544
601
ant_value_t mode = js_get(js, args[0], "mode");
545
602
if (!is_undefined(mode)) {
546
-
if (vtype(mode) == T_STR) {
547
-
size_t mode_len;
548
-
const char *mode_str = js_getstr(js, mode, &mode_len);
549
-
if (mode_str && mode_len == 4 && memcmp(mode_str, "byob", 4) == 0)
550
-
return js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStreamBYOBReader is not yet implemented");
603
+
ant_value_t mode_str_v = mode;
604
+
if (vtype(mode) != T_STR) {
605
+
mode_str_v = js_tostring_val(js, mode);
606
+
if (is_err(mode_str_v)) return mode_str_v;
551
607
}
608
+
size_t mode_len;
609
+
const char *mode_str = js_getstr(js, mode_str_v, &mode_len);
610
+
if (mode_str && mode_len == 4 && memcmp(mode_str, "byob", 4) == 0)
611
+
return js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStreamBYOBReader is not yet implemented");
552
612
return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid reader mode");
553
613
}
554
614
}
···
655
715
double hwm = 1;
656
716
if (is_object_type(strategy)) {
657
717
ant_value_t hwm_val = js_get(js, strategy, "highWaterMark");
718
+
if (is_err(hwm_val)) return hwm_val;
658
719
if (!is_undefined(hwm_val)) {
659
720
hwm = js_to_number(js, hwm_val);
660
-
if (hwm != hwm || hwm < 0)
661
-
return js_mkerr_typed(js, JS_ERR_RANGE, "Invalid highWaterMark");
721
+
if (hwm != hwm || hwm < 0) return js_mkerr_typed(js, JS_ERR_RANGE, "Invalid highWaterMark");
662
722
}
663
723
}
664
724
665
725
ant_value_t size_fn = js_mkundef();
666
726
if (is_object_type(strategy)) {
667
727
ant_value_t s = js_get(js, strategy, "size");
728
+
if (is_err(s)) return s;
668
729
if (!is_undefined(s)) {
669
-
if (!is_callable(s))
670
-
return js_mkerr_typed(js, JS_ERR_TYPE, "size must be a function");
730
+
if (!is_callable(s)) return js_mkerr_typed(js, JS_ERR_TYPE, "size must be a function");
671
731
size_fn = s;
672
732
}
673
733
}