+1 -1

include/streams/readable.h

reviewed

··· 53 53 ant_value_t rs_cancel_resolve(ant_t *js, ant_value_t *args, int nargs); 54 54 ant_value_t readable_stream_cancel(ant_t *js, ant_value_t stream_obj, ant_value_t reason); 55 55 ant_value_t rs_create_stream(ant_t *js, ant_value_t pull_fn, ant_value_t cancel_fn, double hwm); 56 56 + ant_value_t rs_controller_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t chunk); 56 57 57 58 void rs_controller_close(ant_t *js, ant_value_t ctrl_obj); 58 59 void rs_default_controller_clear_algorithms(ant_value_t ctrl_obj); 59 60 void rs_ctrl_queue_push(ant_t *js, ant_value_t ctrl_obj, ant_value_t value); 60 60 - void rs_controller_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t chunk); 61 61 void rs_default_controller_call_pull_if_needed(ant_t *js, ant_value_t controller_obj); 62 62 void rs_default_reader_error_read_requests(ant_t *js, ant_value_t reader_obj, ant_value_t e); 63 63 void rs_fulfill_read_request(ant_t *js, ant_value_t stream_obj, ant_value_t chunk, bool done);

+13 -8

src/streams/readable.c

reviewed

··· 450 450 if (is_err(size_result)) { 451 451 ant_value_t thrown = js->thrown_value; 452 452 ant_value_t err = is_object_type(thrown) ? thrown : size_result; 453 453 + if (stream && stream->state == RS_STATE_ERRORED) 454 454 + err = rs_stream_error(stream_obj); 453 455 readable_stream_error(js, stream_obj, err); 454 454 - if (is_object_type(thrown)) return js_throw(js, thrown); 455 455 - return size_result; 456 456 + return js_throw(js, err); 456 457 } 457 458 if (vtype(size_result) == T_NUM) chunk_size = js_getnum(size_result); 458 459 else chunk_size = js_to_number(js, size_result); ··· 495 496 return js_mkundef(); 496 497 } 497 498 498 498 - void rs_controller_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t chunk) { 499 499 + ant_value_t rs_controller_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t chunk) { 499 500 rs_controller_t *ctrl = rs_get_controller(ctrl_obj); 500 500 - if (!ctrl) return; 501 501 + if (!ctrl) return js_mkundef(); 501 502 502 503 ant_value_t stream_obj = rs_ctrl_stream(ctrl_obj); 503 504 rs_stream_t *stream = rs_get_stream(stream_obj); 504 504 - if (!rs_default_controller_can_close_or_enqueue(ctrl, stream)) return; 505 505 + if (!rs_default_controller_can_close_or_enqueue(ctrl, stream)) 506 506 + return js_mkerr_typed(js, JS_ERR_TYPE, "The stream is not in a state that permits enqueue"); 505 507 506 508 ant_value_t reader_obj = rs_stream_reader(stream_obj); 507 509 if (is_object_type(reader_obj) && rs_reader_has_reqs(js, reader_obj)) { 508 510 rs_fulfill_read_request(js, stream_obj, chunk, false); 509 511 rs_default_controller_call_pull_if_needed(js, ctrl_obj); 510 510 - return; 512 512 + return js_mkundef(); 511 513 } 512 514 513 515 double chunk_size = 1; ··· 518 520 if (is_err(size_result)) { 519 521 ant_value_t thrown = js->thrown_value; 520 522 ant_value_t err = is_object_type(thrown) ? thrown : size_result; 523 523 + if (stream && stream->state == RS_STATE_ERRORED) 524 524 + err = rs_stream_error(stream_obj); 521 525 js->thrown_exists = false; 522 526 js->thrown_value = js_mkundef(); 523 527 js->thrown_stack = js_mkundef(); 524 528 readable_stream_error(js, stream_obj, err); 525 525 - return; 529 529 + return js_throw(js, err); 526 530 } 527 531 if (vtype(size_result) == T_NUM) chunk_size = js_getnum(size_result); 528 532 else chunk_size = js_to_number(js, size_result); ··· 532 536 ant_value_t err = js_make_error_silent(js, JS_ERR_RANGE, 533 537 "The return value of a queuing strategy's size function must be a finite, non-NaN, non-negative number"); 534 538 readable_stream_error(js, stream_obj, err); 535 535 - return; 539 539 + return js_throw(js, err); 536 540 } 537 541 538 542 rs_ctrl_queue_push(js, ctrl_obj, chunk); ··· 546 550 ctrl->queue_sizes[ctrl->queue_sizes_len++] = chunk_size; 547 551 ctrl->queue_total_size += chunk_size; 548 552 rs_default_controller_call_pull_if_needed(js, ctrl_obj); 553 553 + return js_mkundef(); 549 554 } 550 555 551 556 void rs_controller_close(ant_t *js, ant_value_t ctrl_obj) {

+300 -55

src/streams/transform.c

reviewed

··· 113 113 return js_get_slot(ctrl_obj, SLOT_RS_PULL); 114 114 } 115 115 116 116 + static inline ant_value_t ts_writable_stored_error(ant_value_t ts_obj) { 117 117 + return js_get_slot(ts_writable(ts_obj), SLOT_BUFFER); 118 118 + } 119 119 + 120 120 + static inline ant_value_t ts_cancel_promise(ant_value_t ts_obj) { 121 121 + return js_get_slot(ts_obj, SLOT_RS_CANCEL); 122 122 + } 123 123 + 124 124 + static inline bool ts_cancel_started_by_abort(ant_value_t ts_obj) { 125 125 + return js_get_slot(ts_obj, SLOT_WS_ABORT) == js_true; 126 126 + } 127 127 + 128 128 + static inline bool ts_is_flushing(ant_value_t ts_obj) { 129 129 + return js_get_slot(ts_obj, SLOT_WS_CLOSE) == js_true; 130 130 + } 131 131 + 132 132 + static inline ant_value_t ts_cancel_settle_error(ant_value_t ts_obj) { 133 133 + return js_get_slot(ts_obj, SLOT_RS_SIZE); 134 134 + } 135 135 + 136 136 + static inline bool ts_cancel_joined_abort(ant_value_t ts_obj) { 137 137 + return js_get_slot(ts_obj, SLOT_WS_WRITE) == js_true; 138 138 + } 139 139 + 140 140 + static inline bool ts_cancel_has_user_handler(ant_value_t ts_obj) { 141 141 + return js_get_slot(ts_obj, SLOT_WS_SIGNAL) == js_true; 142 142 + } 143 143 + 144 144 + static inline void ts_set_flushing(ant_value_t ts_obj, bool flushing) { 145 145 + js_set_slot(ts_obj, SLOT_WS_CLOSE, flushing ? js_true : js_false); 146 146 + } 147 147 + 148 148 + static inline void ts_set_cancel_state(ant_value_t ts_obj, ant_value_t promise, bool started_by_abort, bool has_user_handler) { 149 149 + js_set_slot(ts_obj, SLOT_RS_CANCEL, promise); 150 150 + js_set_slot(ts_obj, SLOT_WS_ABORT, started_by_abort ? js_true : js_false); 151 151 + js_set_slot(ts_obj, SLOT_RS_SIZE, js_mkundef()); 152 152 + js_set_slot(ts_obj, SLOT_WS_WRITE, js_false); 153 153 + js_set_slot(ts_obj, SLOT_WS_SIGNAL, has_user_handler ? js_true : js_false); 154 154 + } 155 155 + 156 156 + static inline void ts_clear_cancel_state(ant_value_t ts_obj, ant_value_t promise) { 157 157 + if (ts_cancel_promise(ts_obj) == promise) { 158 158 + js_set_slot(ts_obj, SLOT_RS_CANCEL, js_mkundef()); 159 159 + js_set_slot(ts_obj, SLOT_WS_ABORT, js_false); 160 160 + js_set_slot(ts_obj, SLOT_WS_SIGNAL, js_false); 161 161 + }} 162 162 + 116 163 static void ts_ctrl_clear_algorithms(ant_value_t ctrl_obj) { 117 164 js_set_slot(ctrl_obj, SLOT_ENTRIES, js_mkundef()); 118 165 js_set_slot(ctrl_obj, SLOT_CTOR, js_mkundef()); ··· 197 244 ts_set_bp_flag(ts_obj, backpressure); 198 245 } 199 246 200 200 - static void ts_ctrl_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t chunk) { 247 247 + static ant_value_t ts_ctrl_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t chunk) { 201 248 ant_value_t ts_obj = ts_ctrl_stream(ctrl_obj); 202 249 ant_value_t readable = ts_readable(ts_obj); 203 250 ant_value_t rs_ctrl = rs_stream_controller(js, readable); 204 251 rs_stream_t *rs = rs_get_stream(readable); 205 205 - if (!rs || rs->state != RS_STATE_READABLE) return; 252 252 + if (!rs || rs->state != RS_STATE_READABLE) 253 253 + return js_mkerr_typed(js, JS_ERR_TYPE, "Readable side is not in a readable state"); 206 254 207 207 - rs_controller_enqueue(js, rs_ctrl, chunk); 255 255 + ant_value_t enqueue_result = rs_controller_enqueue(js, rs_ctrl, chunk); 256 256 + if (is_err(enqueue_result)) { 257 257 + ant_value_t err = ts_take_thrown_or(js, enqueue_result); 258 258 + ts_error_writable_and_unblock_write(js, ts_obj, err); 259 259 + return js_throw(js, err); 260 260 + } 208 261 209 262 rs_controller_t *rc = rs_get_controller(rs_ctrl); 210 210 - bool bp = rc && (rc->queue_total_size > 0); 263 263 + bool bp = rc && ((rc->strategy_hwm - rc->queue_total_size) <= 0); 211 264 if (bp != ts_get_backpressure(ts_obj)) 212 265 ts_set_backpressure(js, ts_obj, bp); 266 266 + return js_mkundef(); 213 267 } 214 268 215 269 static void ts_ctrl_error(ant_t *js, ant_value_t ctrl_obj, ant_value_t e) { 216 270 ant_value_t ts_obj = ts_ctrl_stream(ctrl_obj); 271 271 + if (vtype(ts_cancel_promise(ts_obj)) == T_PROMISE && ts_cancel_has_user_handler(ts_obj)) 272 272 + js_set_slot(ts_obj, SLOT_RS_SIZE, e); 217 273 ts_error(js, ts_obj, e); 218 274 ts_error_writable_and_unblock_write(js, ts_obj, e); 219 275 } ··· 278 334 ts_chain_thenable(js, resolved, res_fn, rej_fn); 279 335 } 280 336 } else { 281 281 - ts_ctrl_enqueue(js, ctrl_obj, chunk); 337 337 + ant_value_t enqueue_result = ts_ctrl_enqueue(js, ctrl_obj, chunk); 338 338 + if (is_err(enqueue_result)) { 339 339 + ant_value_t err = ts_take_thrown_or(js, enqueue_result); 340 340 + js_reject_promise(js, p, err); 341 341 + return p; 342 342 + } 282 343 js_resolve_promise(js, p, js_mkundef()); 283 344 } 284 345 285 346 return p; 286 347 } 287 348 349 349 + static ant_value_t ts_cancel_base_resolve(ant_t *js, ant_value_t *args, int nargs) { 350 350 + ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 351 351 + ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 352 352 + ant_value_t ts_obj = js_get_slot(wrapper, SLOT_ENTRIES); 353 353 + ts_clear_cancel_state(ts_obj, p); 354 354 + js_resolve_promise(js, p, js_mkundef()); 355 355 + return js_mkundef(); 356 356 + } 357 357 + 358 358 + static ant_value_t ts_cancel_base_reject(ant_t *js, ant_value_t *args, int nargs) { 359 359 + ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 360 360 + ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 361 361 + ant_value_t ts_obj = js_get_slot(wrapper, SLOT_ENTRIES); 362 362 + ant_value_t err = (nargs > 0) ? args[0] : js_mkundef(); 363 363 + ts_clear_cancel_state(ts_obj, p); 364 364 + js_reject_promise(js, p, err); 365 365 + return js_mkundef(); 366 366 + } 367 367 + 368 368 + static ant_value_t ts_run_cancel_algorithm(ant_t *js, ant_value_t ts_obj, ant_value_t cancel_fn, ant_value_t reason, bool started_by_abort) { 369 369 + ant_value_t existing = ts_cancel_promise(ts_obj); 370 370 + if (vtype(existing) == T_PROMISE) return existing; 371 371 + 372 372 + ant_value_t p = js_mkpromise(js); 373 373 + promise_mark_handled(p); 374 374 + ts_set_cancel_state(ts_obj, p, started_by_abort, is_callable(cancel_fn)); 375 375 + ts_ctrl_clear_algorithms(ts_controller(ts_obj)); 376 376 + 377 377 + ant_value_t result = js_mkundef(); 378 378 + if (is_callable(cancel_fn)) { 379 379 + ant_value_t cancel_args[1] = { reason }; 380 380 + result = sv_vm_call(js->vm, js, cancel_fn, ts_ctrl_transformer(ts_controller(ts_obj)), cancel_args, 1, NULL, false); 381 381 + } 382 382 + 383 383 + if (is_err(result)) { 384 384 + ant_value_t err = ts_take_thrown_or(js, result); 385 385 + ts_clear_cancel_state(ts_obj, p); 386 386 + js_reject_promise(js, p, err); 387 387 + return p; 388 388 + } 389 389 + 390 390 + ant_value_t wrapper = js_mkobj(js); 391 391 + js_set_slot(wrapper, SLOT_DATA, p); 392 392 + js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); 393 393 + ant_value_t res_fn = js_heavy_mkfun(js, ts_cancel_base_resolve, wrapper); 394 394 + ant_value_t rej_fn = js_heavy_mkfun(js, ts_cancel_base_reject, wrapper); 395 395 + 396 396 + if (ts_is_thenable(js, result)) ts_chain_thenable(js, result, res_fn, rej_fn); 397 397 + else { 398 398 + ant_value_t resolved = js_mkpromise(js); 399 399 + js_resolve_promise(js, resolved, js_mkundef()); 400 400 + ts_chain_thenable(js, resolved, res_fn, rej_fn); 401 401 + } 402 402 + 403 403 + return p; 404 404 + } 405 405 + 406 406 + static ant_value_t ts_source_cancel_resolve(ant_t *js, ant_value_t *args, int nargs) { 407 407 + ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 408 408 + ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 409 409 + ant_value_t ts_obj = js_get_slot(wrapper, SLOT_ENTRIES); 410 410 + ant_value_t reason = js_get_slot(wrapper, SLOT_BUFFER); 411 411 + bool started_by_abort = js_get_slot(wrapper, SLOT_CTOR) == js_true; 412 412 + 413 413 + ant_value_t readable = ts_readable(ts_obj); 414 414 + rs_stream_t *rs = rs_get_stream(readable); 415 415 + ant_value_t settle_error = ts_cancel_settle_error(ts_obj); 416 416 + if (is_object_type(settle_error)) { 417 417 + js_reject_promise(js, p, settle_error); 418 418 + return js_mkundef(); 419 419 + } 420 420 + if (rs && rs->state == RS_STATE_ERRORED) { 421 421 + js_reject_promise(js, p, rs_stream_error(readable)); 422 422 + return js_mkundef(); 423 423 + } 424 424 + 425 425 + ant_value_t writable = ts_writable(ts_obj); 426 426 + ws_stream_t *ws = ws_get_stream(writable); 427 427 + if (!started_by_abort && ws && (ws->has_pending_abort || ts_cancel_joined_abort(ts_obj))) { 428 428 + js_reject_promise(js, p, ts_writable_stored_error(ts_obj)); 429 429 + return js_mkundef(); 430 430 + } 431 431 + 432 432 + if (ws && ws->state == WS_STATE_WRITABLE) 433 433 + ts_error_writable_and_unblock_write(js, ts_obj, reason); 434 434 + 435 435 + js_resolve_promise(js, p, js_mkundef()); 436 436 + return js_mkundef(); 437 437 + } 438 438 + 439 439 + static ant_value_t ts_source_cancel_reject(ant_t *js, ant_value_t *args, int nargs) { 440 440 + ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 441 441 + ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 442 442 + ant_value_t ts_obj = js_get_slot(wrapper, SLOT_ENTRIES); 443 443 + ant_value_t err = (nargs > 0) ? args[0] : js_mkundef(); 444 444 + ts_error_writable_and_unblock_write(js, ts_obj, err); 445 445 + js_reject_promise(js, p, err); 446 446 + return js_mkundef(); 447 447 + } 448 448 + 449 449 + static ant_value_t ts_abort_cancel_resolve(ant_t *js, ant_value_t *args, int nargs) { 450 450 + ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 451 451 + ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 452 452 + ant_value_t ts_obj = js_get_slot(wrapper, SLOT_ENTRIES); 453 453 + ant_value_t reason = js_get_slot(wrapper, SLOT_BUFFER); 454 454 + bool started_by_abort = js_get_slot(wrapper, SLOT_CTOR) == js_true; 455 455 + 456 456 + ant_value_t readable = ts_readable(ts_obj); 457 457 + rs_stream_t *rs = rs_get_stream(readable); 458 458 + if (rs && rs->state == RS_STATE_ERRORED) { 459 459 + js_reject_promise(js, p, rs_stream_error(readable)); 460 460 + return js_mkundef(); 461 461 + } 462 462 + 463 463 + if (!started_by_abort) { 464 464 + js_reject_promise(js, p, ts_writable_stored_error(ts_obj)); 465 465 + return js_mkundef(); 466 466 + } 467 467 + 468 468 + ts_error(js, ts_obj, reason); 469 469 + js_resolve_promise(js, p, js_mkundef()); 470 470 + return js_mkundef(); 471 471 + } 472 472 + 473 473 + static ant_value_t ts_abort_cancel_reject(ant_t *js, ant_value_t *args, int nargs) { 474 474 + ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 475 475 + ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 476 476 + ant_value_t ts_obj = js_get_slot(wrapper, SLOT_ENTRIES); 477 477 + ant_value_t err = (nargs > 0) ? args[0] : js_mkundef(); 478 478 + ts_error(js, ts_obj, err); 479 479 + js_reject_promise(js, p, err); 480 480 + return js_mkundef(); 481 481 + } 482 482 + 483 483 + static ant_value_t ts_close_cancel_resolve(ant_t *js, ant_value_t *args, int nargs) { 484 484 + ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 485 485 + ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 486 486 + ant_value_t ts_obj = js_get_slot(wrapper, SLOT_ENTRIES); 487 487 + ant_value_t readable = ts_readable(ts_obj); 488 488 + rs_stream_t *rs = rs_get_stream(readable); 489 489 + ant_value_t settle_error = ts_cancel_settle_error(ts_obj); 490 490 + if (is_object_type(settle_error)) { 491 491 + js_reject_promise(js, p, settle_error); 492 492 + return js_mkundef(); 493 493 + } 494 494 + if (rs && rs->state == RS_STATE_ERRORED) { 495 495 + js_reject_promise(js, p, rs_stream_error(readable)); 496 496 + return js_mkundef(); 497 497 + } 498 498 + js_resolve_promise(js, p, js_mkundef()); 499 499 + return js_mkundef(); 500 500 + } 501 501 + 502 502 + static ant_value_t ts_close_cancel_reject(ant_t *js, ant_value_t *args, int nargs) { 503 503 + ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 504 504 + ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 505 505 + ant_value_t err = (nargs > 0) ? args[0] : js_mkundef(); 506 506 + js_reject_promise(js, p, err); 507 507 + return js_mkundef(); 508 508 + } 509 509 + 288 510 static ant_value_t ts_sink_write_bp_resolve(ant_t *js, ant_value_t *args, int nargs) { 289 511 ant_value_t wrapper = js_get_slot(js->current_func, SLOT_DATA); 290 512 ant_value_t ctrl_obj = js_get_slot(wrapper, SLOT_DATA); ··· 293 515 ws_stream_t *ws = ws_get_stream(ts_writable(ts_obj)); 294 516 295 517 if (ws && ws->state == WS_STATE_ERRORING) { 296 296 - ant_value_t err = js_make_error_silent(js, JS_ERR_TYPE, "WritableStream is in erroring state"); 518 518 + ant_value_t err = ts_writable_stored_error(ts_obj); 519 519 + if (!is_object_type(err)) 520 520 + err = js_make_error_silent(js, JS_ERR_TYPE, "WritableStream is in erroring state"); 297 521 ant_value_t fp = ts_ctrl_finish_promise(ctrl_obj); 298 522 if (vtype(fp) == T_PROMISE) js_reject_promise(js, fp, err); 299 523 return js_mkundef(); ··· 351 575 ant_value_t ctrl_obj = ts_controller(ts_obj); 352 576 ant_value_t cancel_fn = ts_ctrl_cancel_fn(ctrl_obj); 353 577 ant_value_t reason = (nargs > 0) ? args[0] : js_mkundef(); 354 354 - 355 355 - if (is_callable(cancel_fn)) { 356 356 - ant_value_t cancel_args[1] = { reason }; 357 357 - ant_value_t result = sv_vm_call(js->vm, js, cancel_fn, ts_ctrl_transformer(ctrl_obj), cancel_args, 1, NULL, false); 358 358 - if (is_err(result)) reason = ts_take_thrown_or(js, reason); 359 359 - } 578 578 + bool joined_source_cancel = vtype(ts_cancel_promise(ts_obj)) == T_PROMISE && !ts_cancel_started_by_abort(ts_obj); 579 579 + if (joined_source_cancel) js_set_slot(ts_obj, SLOT_WS_WRITE, js_true); 360 580 361 361 - ts_error(js, ts_obj, reason); 362 581 ant_value_t p = js_mkpromise(js); 363 363 - js_resolve_promise(js, p, js_mkundef()); 582 582 + ant_value_t base = ts_run_cancel_algorithm(js, ts_obj, cancel_fn, reason, true); 583 583 + ant_value_t wrapper = js_mkobj(js); 584 584 + js_set_slot(wrapper, SLOT_DATA, p); 585 585 + js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); 586 586 + js_set_slot(wrapper, SLOT_BUFFER, reason); 587 587 + js_set_slot(wrapper, SLOT_CTOR, ts_cancel_started_by_abort(ts_obj) ? js_true : js_false); 588 588 + ant_value_t res_fn = js_heavy_mkfun(js, ts_abort_cancel_resolve, wrapper); 589 589 + ant_value_t rej_fn = js_heavy_mkfun(js, ts_abort_cancel_reject, wrapper); 590 590 + ts_chain_thenable(js, base, res_fn, rej_fn); 364 591 return p; 365 592 } 366 593 ··· 371 598 372 599 ant_value_t readable = ts_readable(ts_obj); 373 600 rs_stream_t *rs = rs_get_stream(readable); 374 374 - if (rs && rs->state == RS_STATE_READABLE) js_resolve_promise(js, p, js_mkundef()); 375 375 - else { 376 376 - ant_value_t err = js_make_error_silent(js, JS_ERR_TYPE, "TransformStream readable side is not in a readable state"); 377 377 - js_reject_promise(js, p, err); 601 601 + ant_value_t rs_ctrl = rs_stream_controller(js, readable); 602 602 + ts_set_flushing(ts_obj, false); 603 603 + 604 604 + if (rs && rs->state == RS_STATE_READABLE) { 605 605 + rs_controller_close(js, rs_ctrl); 606 606 + js_resolve_promise(js, p, js_mkundef()); 607 607 + } else if (rs && rs->state == RS_STATE_CLOSED) { 608 608 + js_resolve_promise(js, p, js_mkundef()); 609 609 + } else if (rs && rs->state == RS_STATE_ERRORED) { 610 610 + js_reject_promise(js, p, rs_stream_error(readable)); 611 611 + } else { 612 612 + js_reject_promise(js, p, js_make_error_silent(js, JS_ERR_TYPE, "TransformStream readable side is not in a readable state")); 378 613 } 379 614 380 615 return js_mkundef(); ··· 385 620 ant_value_t p = js_get_slot(wrapper, SLOT_DATA); 386 621 ant_value_t ts_obj = js_get_slot(wrapper, SLOT_ENTRIES); 387 622 ant_value_t e = (nargs > 0) ? args[0] : js_mkundef(); 623 623 + ts_set_flushing(ts_obj, false); 388 624 ts_error(js, ts_obj, e); 389 625 js_reject_promise(js, p, e); 390 626 return js_mkundef(); ··· 396 632 ant_value_t readable = ts_readable(ts_obj); 397 633 398 634 ant_value_t flush_fn = ts_ctrl_flush_fn(ctrl_obj); 399 399 - ts_ctrl_clear_algorithms(ctrl_obj); 400 400 - 401 635 ant_value_t p = js_mkpromise(js); 402 636 promise_mark_handled(p); 403 637 638 638 + ant_value_t cancel_p = ts_cancel_promise(ts_obj); 639 639 + if (vtype(cancel_p) == T_PROMISE) { 640 640 + ant_value_t wrapper = js_mkobj(js); 641 641 + js_set_slot(wrapper, SLOT_DATA, p); 642 642 + js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); 643 643 + ant_value_t res_fn = js_heavy_mkfun(js, ts_close_cancel_resolve, wrapper); 644 644 + ant_value_t rej_fn = js_heavy_mkfun(js, ts_close_cancel_reject, wrapper); 645 645 + ts_chain_thenable(js, cancel_p, res_fn, rej_fn); 646 646 + return p; 647 647 + } 648 648 + 649 649 + ts_ctrl_clear_algorithms(ctrl_obj); 650 650 + ts_set_flushing(ts_obj, true); 651 651 + 404 652 if (is_callable(flush_fn)) { 405 653 ant_value_t flush_args[1] = { ctrl_obj }; 406 654 ant_value_t result = sv_vm_call(js->vm, js, flush_fn, ts_ctrl_transformer(ctrl_obj), flush_args, 1, NULL, false); ··· 408 656 if (is_err(result)) { 409 657 ant_value_t err = ts_take_thrown_or(js, result); 410 658 ts_error(js, ts_obj, err); 659 659 + ts_set_flushing(ts_obj, false); 411 660 js_reject_promise(js, p, err); 412 661 return p; 413 662 } ··· 430 679 ant_value_t rs_ctrl = rs_stream_controller(js, readable); 431 680 rs_controller_close(js, rs_ctrl); 432 681 } 682 682 + ts_set_flushing(ts_obj, false); 433 683 js_resolve_promise(js, p, js_mkundef()); 434 684 } 435 685 ··· 463 713 ant_value_t cancel_fn = ts_ctrl_cancel_fn(ctrl_obj); 464 714 ant_value_t reason = (nargs > 0) ? args[0] : js_mkundef(); 465 715 466 466 - ts_ctrl_clear_algorithms(ctrl_obj); 467 467 - 468 468 - if (is_callable(cancel_fn)) { 469 469 - ant_value_t cancel_args[1] = { reason }; 470 470 - ant_value_t result = sv_vm_call(js->vm, js, cancel_fn, ts_ctrl_transformer(ctrl_obj), cancel_args, 1, NULL, false); 471 471 - 472 472 - if (is_err(result)) reason = ts_take_thrown_or(js, reason); 473 473 - else if (ts_is_thenable(js, result)) { 474 474 - ant_value_t p = js_mkpromise(js); 475 475 - ant_value_t wrapper = js_mkobj(js); 476 476 - js_set_slot(wrapper, SLOT_DATA, p); 477 477 - js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); 478 478 - 479 479 - ant_value_t res_fn = js_heavy_mkfun(js, ts_source_pull_resolve, p); 480 480 - ant_value_t rej_fn = js_heavy_mkfun(js, ts_sink_close_reject, wrapper); 481 481 - ts_chain_thenable(js, result, res_fn, rej_fn); 482 482 - 483 483 - ts_error_writable_and_unblock_write(js, ts_obj, reason); 484 484 - return p; 485 485 - } 716 716 + if (ts_is_flushing(ts_obj)) { 717 717 + ant_value_t p = js_mkpromise(js); 718 718 + js_resolve_promise(js, p, js_mkundef()); 719 719 + return p; 486 720 } 487 721 488 488 - ts_error_writable_and_unblock_write(js, ts_obj, reason); 489 489 - 490 722 ant_value_t p = js_mkpromise(js); 491 491 - js_resolve_promise(js, p, js_mkundef()); 723 723 + ant_value_t base = ts_run_cancel_algorithm(js, ts_obj, cancel_fn, reason, false); 724 724 + ant_value_t wrapper = js_mkobj(js); 725 725 + js_set_slot(wrapper, SLOT_DATA, p); 726 726 + js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); 727 727 + js_set_slot(wrapper, SLOT_BUFFER, reason); 728 728 + js_set_slot(wrapper, SLOT_CTOR, ts_cancel_started_by_abort(ts_obj) ? js_true : js_false); 729 729 + ant_value_t res_fn = js_heavy_mkfun(js, ts_source_cancel_resolve, wrapper); 730 730 + ant_value_t rej_fn = js_heavy_mkfun(js, ts_source_cancel_reject, wrapper); 731 731 + ts_chain_thenable(js, base, res_fn, rej_fn); 492 732 return p; 493 733 } 494 734 ··· 512 752 return js_mkerr_typed(js, JS_ERR_TYPE, "Readable side is not in a readable state"); 513 753 514 754 ant_value_t chunk = (nargs > 0) ? args[0] : js_mkundef(); 515 515 - ts_ctrl_enqueue(js, js->this_val, chunk); 516 516 - return js_mkundef(); 755 755 + return ts_ctrl_enqueue(js, js->this_val, chunk); 517 756 } 518 757 519 758 static ant_value_t js_ts_ctrl_error(ant_t *js, ant_value_t *args, int nargs) { ··· 590 829 transformer = args[0]; 591 830 592 831 if (is_object_type(transformer)) { 593 593 - ant_value_t rt_val = js_get(js, transformer, "readableType"); 832 832 + ant_value_t rt_val = js_getprop_fallback(js, transformer, "readableType"); 594 833 if (!is_undefined(rt_val)) 595 834 return js_mkerr_typed(js, JS_ERR_RANGE, "readableType is not supported"); 596 596 - ant_value_t wt_val = js_get(js, transformer, "writableType"); 835 835 + ant_value_t wt_val = js_getprop_fallback(js, transformer, "writableType"); 597 836 if (!is_undefined(wt_val)) 598 837 return js_mkerr_typed(js, JS_ERR_RANGE, "writableType is not supported"); 599 838 } ··· 609 848 double writable_hwm = 1; 610 849 ant_value_t writable_size_fn = js_mkundef(); 611 850 if (is_object_type(writable_strategy)) { 612 612 - ant_value_t hwm_val = js_get(js, writable_strategy, "highWaterMark"); 851 851 + ant_value_t hwm_val = js_getprop_fallback(js, writable_strategy, "highWaterMark"); 613 852 if (is_err(hwm_val)) return hwm_val; 614 853 if (!is_undefined(hwm_val)) { 615 854 writable_hwm = js_to_number(js, hwm_val); 616 855 if (writable_hwm != writable_hwm || writable_hwm < 0) 617 856 return js_mkerr_typed(js, JS_ERR_RANGE, "Invalid highWaterMark"); 618 857 } 619 619 - ant_value_t s = js_get(js, writable_strategy, "size"); 858 858 + ant_value_t s = js_getprop_fallback(js, writable_strategy, "size"); 620 859 if (is_err(s)) return s; 621 860 if (!is_undefined(s)) { 622 861 if (!is_callable(s)) return js_mkerr_typed(js, JS_ERR_TYPE, "size must be a function"); ··· 627 866 double readable_hwm = 0; 628 867 ant_value_t readable_size_fn = js_mkundef(); 629 868 if (is_object_type(readable_strategy)) { 630 630 - ant_value_t hwm_val = js_get(js, readable_strategy, "highWaterMark"); 869 869 + ant_value_t hwm_val = js_getprop_fallback(js, readable_strategy, "highWaterMark"); 631 870 if (is_err(hwm_val)) return hwm_val; 632 871 if (!is_undefined(hwm_val)) { 633 872 readable_hwm = js_to_number(js, hwm_val); 634 873 if (readable_hwm != readable_hwm || readable_hwm < 0) 635 874 return js_mkerr_typed(js, JS_ERR_RANGE, "Invalid highWaterMark"); 636 875 } 637 637 - ant_value_t s = js_get(js, readable_strategy, "size"); 876 876 + ant_value_t s = js_getprop_fallback(js, readable_strategy, "size"); 638 877 if (is_err(s)) return s; 639 878 if (!is_undefined(s)) { 640 879 if (!is_callable(s)) return js_mkerr_typed(js, JS_ERR_TYPE, "size must be a function"); ··· 648 887 ant_value_t start_fn = js_mkundef(); 649 888 650 889 if (is_object_type(transformer)) { 651 651 - ant_value_t tv = js_get(js, transformer, "transform"); 890 890 + ant_value_t tv = js_getprop_fallback(js, transformer, "transform"); 652 891 if (is_err(tv)) return tv; 653 892 if (!is_undefined(tv)) { 654 893 if (!is_callable(tv)) return js_mkerr_typed(js, JS_ERR_TYPE, "transform must be a function"); 655 894 transform_fn = tv; 656 895 } 657 657 - ant_value_t fv = js_get(js, transformer, "flush"); 896 896 + ant_value_t fv = js_getprop_fallback(js, transformer, "flush"); 658 897 if (is_err(fv)) return fv; 659 898 if (!is_undefined(fv)) { 660 899 if (!is_callable(fv)) return js_mkerr_typed(js, JS_ERR_TYPE, "flush must be a function"); 661 900 flush_fn = fv; 662 901 } 663 663 - ant_value_t cv = js_get(js, transformer, "cancel"); 902 902 + ant_value_t cv = js_getprop_fallback(js, transformer, "cancel"); 664 903 if (is_err(cv)) return cv; 665 904 if (!is_undefined(cv)) { 666 905 if (!is_callable(cv)) return js_mkerr_typed(js, JS_ERR_TYPE, "cancel must be a function"); 667 906 cancel_fn = cv; 668 907 } 669 669 - ant_value_t sv = js_get(js, transformer, "start"); 908 908 + ant_value_t sv = js_getprop_fallback(js, transformer, "start"); 670 909 if (is_err(sv)) return sv; 671 910 if (!is_undefined(sv)) { 672 911 if (!is_callable(sv)) return js_mkerr_typed(js, JS_ERR_TYPE, "start must be a function"); ··· 688 927 js_set_slot(ctrl_obj, SLOT_SETTLED, transformer); 689 928 js_set_slot(ctrl_obj, SLOT_RS_PULL, js_mkundef()); 690 929 js_set_slot(ts_obj, SLOT_DEFAULT, ctrl_obj); 930 930 + js_set_slot(ts_obj, SLOT_RS_CANCEL, js_mkundef()); 931 931 + js_set_slot(ts_obj, SLOT_WS_ABORT, js_false); 932 932 + js_set_slot(ts_obj, SLOT_WS_CLOSE, js_false); 933 933 + js_set_slot(ts_obj, SLOT_RS_SIZE, js_mkundef()); 934 934 + js_set_slot(ts_obj, SLOT_WS_WRITE, js_false); 935 935 + js_set_slot(ts_obj, SLOT_WS_SIGNAL, js_false); 691 936 692 937 ant_value_t sink_write = js_heavy_mkfun(js, ts_sink_write, ts_obj); 693 938 ant_value_t sink_abort = js_heavy_mkfun(js, ts_sink_abort, ts_obj);