diff --git a/include/modules/abort.h b/include/modules/abort.h index c021172..735633c 100644 --- a/include/modules/abort.h +++ b/include/modules/abort.h @@ -10,6 +10,7 @@ void gc_mark_abort(ant_t *js, gc_mark_fn mark); void signal_do_abort(ant_t *js, ant_value_t signal, ant_value_t reason); void abort_signal_add_listener(ant_t *js, ant_value_t signal, ant_value_t callback); +bool abort_signal_is_signal(ant_value_t signal); bool abort_signal_is_aborted(ant_value_t signal); ant_value_t abort_signal_get_reason(ant_value_t signal); diff --git a/include/streams/readable.h b/include/streams/readable.h index 3858ebe..b587014 100644 --- a/include/streams/readable.h +++ b/include/streams/readable.h @@ -38,6 +38,10 @@ extern ant_value_t g_controller_proto; void init_readable_stream_module(void); void gc_mark_readable_streams(ant_t *js, void (*mark)(ant_t *, ant_value_t)); +bool rs_is_stream(ant_value_t obj); +bool rs_is_reader(ant_value_t obj); +bool rs_is_controller(ant_value_t obj); + rs_stream_t *rs_get_stream(ant_value_t obj); rs_controller_t *rs_get_controller(ant_value_t obj); ant_offset_t rs_ctrl_queue_len(ant_t *js, ant_value_t ctrl_obj); diff --git a/include/streams/transform.h b/include/streams/transform.h index e025ceb..83a9edf 100644 --- a/include/streams/transform.h +++ b/include/streams/transform.h @@ -7,4 +7,10 @@ void init_transform_stream_module(void); void gc_mark_transform_streams(ant_t *js, void (*mark)(ant_t *, ant_value_t)); +bool ts_is_stream(ant_value_t obj); +bool ts_is_controller(ant_value_t obj); + +ant_value_t ts_stream_readable(ant_value_t ts_obj); +ant_value_t ts_stream_writable(ant_value_t ts_obj); + #endif diff --git a/include/streams/writable.h b/include/streams/writable.h index aeae48c..49e5b5c 100644 --- a/include/streams/writable.h +++ b/include/streams/writable.h @@ -36,6 +36,10 @@ extern ant_value_t g_ws_controller_proto; void init_writable_stream_module(void); void gc_mark_writable_streams(ant_t *js, void (*mark)(ant_t *, ant_value_t)); +bool ws_is_stream(ant_value_t obj); +bool ws_is_writer(ant_value_t obj); +bool ws_is_controller(ant_value_t obj); + ws_stream_t *ws_get_stream(ant_value_t obj); ws_controller_t *ws_get_controller(ant_value_t obj); diff --git a/src/ant.c b/src/ant.c index 836310c..a7ec95e 100644 --- a/src/ant.c +++ b/src/ant.c @@ -623,15 +623,16 @@ static ant_value_t to_string_val(ant_t *js, ant_value_t val) { bool js_truthy(ant_t *js, ant_value_t v) { static const void *dispatch[] = { - [T_OBJ] = &&l_true, - [T_FUNC] = &&l_true, - [T_CFUNC] = &&l_true, - [T_ARR] = &&l_true, - [T_SYMBOL] = &&l_true, - [T_BOOL] = &&l_bool, - [T_STR] = &&l_str, - [T_BIGINT] = &&l_bigint, - [T_NUM] = &&l_num, + [T_OBJ] = &&l_true, + [T_FUNC] = &&l_true, + [T_CFUNC] = &&l_true, + [T_ARR] = &&l_true, + [T_PROMISE] = &&l_true, + [T_SYMBOL] = &&l_true, + [T_BOOL] = &&l_bool, + [T_STR] = &&l_str, + [T_BIGINT] = &&l_bigint, + [T_NUM] = &&l_num, }; uint8_t t = vtype(v); @@ -5644,13 +5645,9 @@ static ant_value_t builtin_object_defineProperty(ant_t *js, ant_value_t *args, i if (has_get) js_set_sym_getter_desc(js, as_obj, prop, getter, desc_flags); if (has_set) js_set_sym_setter_desc(js, as_obj, prop, setter, desc_flags); } else { - if (has_get && has_set) { - js_set_accessor_desc(js, as_obj, prop_str, prop_len, getter, setter, desc_flags); - } else if (has_get) { - js_set_getter_desc(js, as_obj, prop_str, prop_len, getter, desc_flags); - } else { - js_set_setter_desc(js, as_obj, prop_str, prop_len, setter, desc_flags); - } + if (has_get && has_set) js_set_accessor_desc(js, as_obj, prop_str, prop_len, getter, setter, desc_flags); + else if (has_get) js_set_getter_desc(js, as_obj, prop_str, prop_len, getter, desc_flags); + else js_set_setter_desc(js, as_obj, prop_str, prop_len, setter, desc_flags); } } else { int desc_flags = diff --git a/src/modules/abort.c b/src/modules/abort.c index 8cb1d7c..f5f7ea2 100644 --- a/src/modules/abort.c +++ b/src/modules/abort.c @@ -395,6 +395,10 @@ bool abort_signal_is_aborted(ant_value_t signal) { return data && data->aborted; } +bool abort_signal_is_signal(ant_value_t signal) { + return get_signal_data(signal) != NULL; +} + ant_value_t abort_signal_get_reason(ant_value_t signal) { abort_signal_data_t *data = get_signal_data(signal); return data ? data->reason : js_mkundef(); diff --git a/src/modules/headers.c b/src/modules/headers.c index 8ea6fb8..02f57b8 100644 --- a/src/modules/headers.c +++ b/src/modules/headers.c @@ -585,12 +585,12 @@ void init_headers_module(void) { js_set(js, g_headers_proto, "getSetCookie", js_mkfun(js_headers_get_set_cookie)); js_set_sym(js, g_headers_proto, get_iterator_sym(), js_mkfun(js_headers_symbol_iterator)); - js_set_sym_getter_desc(js, g_headers_proto, get_toStringTag_sym(), js_mkfun(sym_this_cb), 0); + js_set_sym(js, g_headers_proto, get_toStringTag_sym(), js_mkstr(js, "Headers", 7)); ant_value_t ctor_obj = js_mkobj(js); js_set_slot(ctor_obj, SLOT_CFUNC, js_mkfun(js_headers_ctor)); js_mkprop_fast(js, ctor_obj, "prototype", 9, g_headers_proto); - js_mkprop_fast(js, ctor_obj, "name", 4, js_mkstr(js, "Headers", 7)); + js_mkprop_fast(js, ctor_obj, "name", 4, js_mkstr(js, "Headers", 7)); js_set_descriptor(js, ctor_obj, "name", 4, 0); ant_value_t ctor = js_obj_to_func(ctor_obj); diff --git a/src/streams/pipes.c b/src/streams/pipes.c index 80276cb..ebd1e02 100644 --- a/src/streams/pipes.c +++ b/src/streams/pipes.c @@ -12,21 +12,21 @@ #include "streams/readable.h" #include "streams/writable.h" -static void pipes_chain_thenable( +static void pipes_chain_promise( ant_t *js, ant_value_t value, ant_value_t on_resolve, ant_value_t on_reject ) { - ant_value_t thenable = value; - if (vtype(thenable) != T_PROMISE) { - thenable = js_mkpromise(js); - js_resolve_promise(js, thenable, value); + ant_value_t promise = value; + if (vtype(promise) != T_PROMISE) { + promise = js_mkpromise(js); + js_resolve_promise(js, promise, value); } - ant_value_t then_fn = js_get(js, thenable, "then"); + ant_value_t then_fn = js_get(js, promise, "then"); if (!is_callable(then_fn)) return; ant_value_t then_args[2] = { on_resolve, on_reject }; - sv_vm_call(js->vm, js, then_fn, thenable, then_args, 2, NULL, false); + sv_vm_call(js->vm, js, then_fn, promise, then_args, 2, NULL, false); } typedef struct { @@ -76,7 +76,7 @@ static ant_value_t pipe_state_promise(ant_value_t state) { static void pipes_release_reader(ant_t *js, ant_value_t reader_obj) { ant_value_t stream_obj = rs_reader_stream(reader_obj); - if (!is_object_type(stream_obj)) return; + if (!rs_is_stream(stream_obj)) return; if (rs_reader_has_reqs(js, reader_obj)) { ant_value_t release_err = js_make_error_silent(js, JS_ERR_TYPE, "Reader was released"); @@ -103,7 +103,7 @@ static void pipes_release_reader(ant_t *js, ant_value_t reader_obj) { static void pipes_release_writer(ant_t *js, ant_value_t writer_obj) { ant_value_t ws_obj = js_get_slot(writer_obj, SLOT_ENTRIES); - if (!is_object_type(ws_obj)) return; + if (!ws_is_stream(ws_obj)) return; ant_value_t rel_err = js_make_error_silent(js, JS_ERR_TYPE, "Writer was released"); ant_value_t ready = js_mkpromise(js); @@ -122,13 +122,13 @@ static void pipes_release_writer(ant_t *js, ant_value_t writer_obj) { static void pipes_release_locks(ant_t *js, ant_value_t state) { ant_value_t reader = pipe_state_reader(state); - if (is_object_type(reader)) { + if (rs_is_reader(reader)) { pipes_release_reader(js, reader); js_set_slot(state, SLOT_BUFFER, js_mkundef()); } ant_value_t writer = pipe_state_writer(state); - if (is_object_type(writer)) { + if (ws_is_writer(writer)) { pipes_release_writer(js, writer); js_set_slot(state, SLOT_DEFAULT, js_mkundef()); } @@ -230,6 +230,10 @@ static ant_value_t pipe_close_dest_reject(ant_t *js, ant_value_t *args, int narg return js_mkundef(); } +static ant_value_t pipe_ignore(ant_t *js, ant_value_t *args, int nargs) { + return js_mkundef(); +} + static ant_value_t pipe_read_resolve(ant_t *js, ant_value_t *args, int nargs) { ant_value_t state = js_get_slot(js->current_func, SLOT_DATA); pipe_state_t *pst = pipe_get_state(state); @@ -251,7 +255,7 @@ static ant_value_t pipe_read_resolve(ant_t *js, ant_value_t *args, int nargs) { ant_value_t close_promise = writable_stream_close(js, pipe_state_dest(state)); ant_value_t on_resolve = js_heavy_mkfun(js, pipe_close_dest_resolve, state); ant_value_t on_reject = js_heavy_mkfun(js, pipe_close_dest_reject, state); - pipes_chain_thenable(js, close_promise, on_resolve, on_reject); + pipes_chain_promise(js, close_promise, on_resolve, on_reject); return js_mkundef(); } @@ -259,7 +263,7 @@ static ant_value_t pipe_read_resolve(ant_t *js, ant_value_t *args, int nargs) { ant_value_t write_promise = ws_writer_write(js, pipe_state_writer(state), value); ant_value_t on_resolve = js_heavy_mkfun(js, pipe_write_resolve, state); ant_value_t on_reject = js_heavy_mkfun(js, pipe_dest_error, state); - pipes_chain_thenable(js, write_promise, on_resolve, on_reject); + pipes_chain_promise(js, write_promise, on_resolve, on_reject); return js_mkundef(); } @@ -281,7 +285,7 @@ static ant_value_t pipe_ready_resolve(ant_t *js, ant_value_t *args, int nargs) { ant_value_t read_promise = rs_default_reader_read(js, pipe_state_reader(state)); ant_value_t on_resolve = js_heavy_mkfun(js, pipe_read_resolve, state); ant_value_t on_reject = js_heavy_mkfun(js, pipe_source_error, state); - pipes_chain_thenable(js, read_promise, on_resolve, on_reject); + pipes_chain_promise(js, read_promise, on_resolve, on_reject); return js_mkundef(); } @@ -306,7 +310,7 @@ static void pipes_pump(ant_t *js, ant_value_t state) { ant_value_t ready = ws_writer_ready(writer); ant_value_t on_resolve = js_heavy_mkfun(js, pipe_ready_resolve, state); ant_value_t on_reject = js_heavy_mkfun(js, pipe_dest_error, state); - pipes_chain_thenable(js, ready, on_resolve, on_reject); + pipes_chain_promise(js, ready, on_resolve, on_reject); } static ant_value_t pipe_create_rejected(ant_t *js, ant_value_t error) { @@ -345,16 +349,16 @@ ant_value_t readable_stream_pipe_to( return pipe_create_rejected(js, js->thrown_value); } - if (is_object_type(rs_stream_reader(source))) { + if (rs_is_reader(rs_stream_reader(source))) { js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStream is already locked"); return pipe_create_rejected(js, js->thrown_value); } - if (is_object_type(ws_stream_writer(dest))) { + if (ws_is_writer(ws_stream_writer(dest))) { js_mkerr_typed(js, JS_ERR_TYPE, "WritableStream is already locked"); return pipe_create_rejected(js, js->thrown_value); } - if (!is_undefined(signal) && !is_object_type(signal)) { + if (!is_undefined(signal) && !abort_signal_is_signal(signal)) { js_mkerr_typed(js, JS_ERR_TYPE, "pipeTo option 'signal' must be an AbortSignal"); return pipe_create_rejected(js, js->thrown_value); } @@ -393,12 +397,18 @@ ant_value_t readable_stream_pipe_to( promise_mark_handled(js_get_slot(writer, SLOT_RS_CLOSED)); promise_mark_handled(js_get_slot(writer, SLOT_WS_READY)); - if (is_object_type(signal) && abort_signal_is_aborted(signal)) { + ant_value_t ignore_fn = js_heavy_mkfun(js, pipe_ignore, state); + ant_value_t source_closed_reject = js_heavy_mkfun(js, pipe_source_error, state); + ant_value_t dest_closed_reject = js_heavy_mkfun(js, pipe_dest_error, state); + pipes_chain_promise(js, rs_reader_closed(reader), ignore_fn, source_closed_reject); + pipes_chain_promise(js, js_get_slot(writer, SLOT_RS_CLOSED), ignore_fn, dest_closed_reject); + + if (abort_signal_is_signal(signal) && abort_signal_is_aborted(signal)) { pipes_shutdown_from_abort(js, state, abort_signal_get_reason(signal)); return promise; } - if (is_object_type(signal)) { + if (abort_signal_is_signal(signal)) { ant_value_t listener = js_heavy_mkfun(js, pipe_abort_listener, state); abort_signal_add_listener(js, signal, listener); } @@ -408,13 +418,7 @@ ant_value_t readable_stream_pipe_to( } static ant_value_t js_rs_pipe_to(ant_t *js, ant_value_t *args, int nargs) { - if (!is_object_type(js->this_val)) { - js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); - return pipe_create_rejected(js, js->thrown_value); - } - - rs_stream_t *stream = rs_get_stream(js->this_val); - if (!stream) { + if (!rs_is_stream(js->this_val)) { js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); return pipe_create_rejected(js, js->thrown_value); } @@ -430,11 +434,8 @@ static ant_value_t js_rs_pipe_to(ant_t *js, ant_value_t *args, int nargs) { static ant_value_t js_rs_pipe_through(ant_t *js, ant_value_t *args, int nargs) { ant_value_t source = js->this_val; - if (!is_object_type(source)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); - rs_stream_t *stream = rs_get_stream(source); - - if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); - if (is_object_type(rs_stream_reader(source))) + if (!rs_is_stream(source)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); + if (rs_is_reader(rs_stream_reader(source))) return js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStream is already locked"); if (nargs < 1 || !is_object_type(args[0])) return js_mkerr_typed(js, JS_ERR_TYPE, "pipeThrough requires a transform object"); @@ -443,11 +444,11 @@ static ant_value_t js_rs_pipe_through(ant_t *js, ant_value_t *args, int nargs) { ant_value_t writable = js_get(js, transform, "writable"); ant_value_t readable = js_get(js, transform, "readable"); - if (!is_object_type(writable) || !ws_get_stream(writable)) + if (!ws_is_stream(writable)) return js_mkerr_typed(js, JS_ERR_TYPE, "pipeThrough transform.writable must be a WritableStream"); - if (!is_object_type(readable) || !rs_get_stream(readable)) + if (!rs_is_stream(readable)) return js_mkerr_typed(js, JS_ERR_TYPE, "pipeThrough transform.readable must be a ReadableStream"); - if (is_object_type(ws_stream_writer(writable))) + if (ws_is_writer(ws_stream_writer(writable))) return js_mkerr_typed(js, JS_ERR_TYPE, "WritableStream is already locked"); bool prevent_close, prevent_abort, prevent_cancel; @@ -496,7 +497,7 @@ static ant_value_t tee_state_reader(ant_value_t state) { static void tee_release_reader(ant_t *js, ant_value_t state) { ant_value_t reader = tee_state_reader(state); - if (!is_object_type(reader)) return; + if (!rs_is_reader(reader)) return; pipes_release_reader(js, reader); js_set_slot(state, SLOT_BUFFER, js_mkundef()); } @@ -552,7 +553,7 @@ static void tee_enqueue_branch(ant_t *js, ant_value_t branch_stream, ant_value_t return; ant_value_t r = rs_stream_reader(branch_stream); - if (is_object_type(r) && rs_reader_has_reqs(js, r)) { + if (rs_is_reader(r) && rs_reader_has_reqs(js, r)) { rs_fulfill_read_request(js, branch_stream, value, false); rs_default_controller_call_pull_if_needed(js, ctrl); return; @@ -654,7 +655,7 @@ static void tee_pull(ant_t *js, ant_value_t state) { ant_value_t read_promise = rs_default_reader_read(js, tee_state_reader(state)); ant_value_t on_resolve = js_heavy_mkfun(js, tee_read_resolve, state); ant_value_t on_reject = js_heavy_mkfun(js, tee_read_reject, state); - pipes_chain_thenable(js, read_promise, on_resolve, on_reject); + pipes_chain_promise(js, read_promise, on_resolve, on_reject); } static ant_value_t tee_branch_pull(ant_t *js, ant_value_t *args, int nargs) { @@ -711,23 +712,21 @@ static ant_value_t tee_branch_cancel(ant_t *js, ant_value_t *args, int nargs) { ant_value_t cancel_promise = readable_stream_cancel(js, orig_stream, reasons); ant_value_t on_resolve = js_heavy_mkfun(js, tee_cancel_both_resolve, state); ant_value_t on_reject = js_heavy_mkfun(js, tee_cancel_both_reject, state); - pipes_chain_thenable(js, cancel_promise, on_resolve, on_reject); + pipes_chain_promise(js, cancel_promise, on_resolve, on_reject); } return promise; } static ant_value_t js_rs_tee(ant_t *js, ant_value_t *args, int nargs) { - if (!is_object_type(js->this_val)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); - rs_stream_t *stream = rs_get_stream(js->this_val); - - if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); - if (is_object_type(rs_stream_reader(js->this_val))) + if (!rs_is_stream(js->this_val)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); + if (rs_is_reader(rs_stream_reader(js->this_val))) return js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStream is already locked"); ant_value_t reader_args[1] = { js->this_val }; ant_value_t saved = js->new_target; js->new_target = g_reader_proto; + ant_value_t reader = js_rs_reader_ctor(js, reader_args, 1); js->new_target = saved; if (is_err(reader)) return reader; diff --git a/src/streams/readable.c b/src/streams/readable.c index 6e0ca98..9190fb2 100644 --- a/src/streams/readable.c +++ b/src/streams/readable.c @@ -17,6 +17,22 @@ ant_value_t g_rs_proto; ant_value_t g_reader_proto; ant_value_t g_controller_proto; +bool rs_is_stream(ant_value_t obj) { + return is_object_type(obj) && rs_get_stream(obj) != NULL; +} + +bool rs_is_reader(ant_value_t obj) { + return is_object_type(obj) + && vtype(js_get_slot(obj, SLOT_RS_CLOSED)) == T_PROMISE + && vtype(js_get_slot(obj, SLOT_BUFFER)) == T_ARR; +} + +bool rs_is_controller(ant_value_t obj) { + return is_object_type(obj) + && rs_get_controller(obj) != NULL + && rs_is_stream(js_get_slot(obj, SLOT_ENTRIES)); +} + rs_stream_t *rs_get_stream(ant_value_t obj) { ant_value_t s = js_get_slot(obj, SLOT_DATA); if (vtype(s) != T_NUM) return NULL; @@ -164,7 +180,7 @@ static bool rs_default_controller_should_call_pull(ant_t *js, rs_controller_t *c ant_value_t stream_obj = rs_ctrl_stream(ctrl_obj); ant_value_t reader_obj = rs_stream_reader(stream_obj); - if (is_object_type(reader_obj) && rs_reader_has_reqs(js, reader_obj)) return true; + if (rs_is_reader(reader_obj) && rs_reader_has_reqs(js, reader_obj)) return true; double desired = rs_default_controller_get_desired_size(ctrl, stream); return desired > 0; @@ -179,7 +195,7 @@ bool rs_default_controller_can_close_or_enqueue(rs_controller_t *ctrl, rs_stream void rs_fulfill_read_request(ant_t *js, ant_value_t stream_obj, ant_value_t chunk, bool done) { ant_value_t reader_obj = rs_stream_reader(stream_obj); - if (!is_object_type(reader_obj)) return; + if (!rs_is_reader(reader_obj)) return; ant_value_t promise = rs_reader_reqs_shift(js, reader_obj); if (vtype(promise) == T_UNDEF) return; ant_value_t result = js_iter_result(js, !done, chunk); @@ -202,7 +218,7 @@ void readable_stream_close(ant_t *js, ant_value_t stream_obj) { stream->state = RS_STATE_CLOSED; ant_value_t reader_obj = rs_stream_reader(stream_obj); - if (is_object_type(reader_obj)) { + if (rs_is_reader(reader_obj)) { ant_value_t arr = rs_reader_reqs(reader_obj); if (vtype(arr) == T_ARR) { ant_offset_t len = js_arr_len(js, arr); @@ -224,7 +240,7 @@ void readable_stream_error(ant_t *js, ant_value_t stream_obj, ant_value_t e) { js_set_slot(stream_obj, SLOT_BUFFER, e); ant_value_t reader_obj = rs_stream_reader(stream_obj); - if (is_object_type(reader_obj)) { + if (rs_is_reader(reader_obj)) { js_reject_promise(js, rs_reader_closed(reader_obj), e); rs_default_reader_error_read_requests(js, reader_obj, e); } @@ -436,7 +452,7 @@ static ant_value_t js_rs_controller_enqueue(ant_t *js, ant_value_t *args, int na ant_value_t chunk = (nargs > 0) ? args[0] : js_mkundef(); ant_value_t reader_obj = rs_stream_reader(stream_obj); - if (is_object_type(reader_obj) && rs_reader_has_reqs(js, reader_obj)) { + if (rs_is_reader(reader_obj) && rs_reader_has_reqs(js, reader_obj)) { rs_fulfill_read_request(js, stream_obj, chunk, false); rs_default_controller_call_pull_if_needed(js, js->this_val); return js_mkundef(); @@ -510,7 +526,7 @@ ant_value_t rs_controller_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t c return js_mkerr_typed(js, JS_ERR_TYPE, "The stream is not in a state that permits enqueue"); ant_value_t reader_obj = rs_stream_reader(stream_obj); - if (is_object_type(reader_obj) && rs_reader_has_reqs(js, reader_obj)) { + if (rs_is_reader(reader_obj) && rs_reader_has_reqs(js, reader_obj)) { rs_fulfill_read_request(js, stream_obj, chunk, false); rs_default_controller_call_pull_if_needed(js, ctrl_obj); return js_mkundef(); @@ -581,16 +597,15 @@ static ant_value_t js_rs_reader_get_closed(ant_t *js, ant_value_t *args, int nar static ant_value_t js_rs_reader_read(ant_t *js, ant_value_t *args, int nargs) { ant_value_t stream_obj = rs_reader_stream(js->this_val); - if (!is_object_type(stream_obj) || !rs_get_stream(stream_obj)) + if (!rs_is_stream(stream_obj)) return js_mkerr_typed(js, JS_ERR_TYPE, "Cannot read from a released reader"); return rs_default_reader_read(js, js->this_val); } static ant_value_t js_rs_reader_release_lock(ant_t *js, ant_value_t *args, int nargs) { ant_value_t stream_obj = rs_reader_stream(js->this_val); - if (!is_object_type(stream_obj)) return js_mkundef(); + if (!rs_is_stream(stream_obj)) return js_mkundef(); rs_stream_t *stream = rs_get_stream(stream_obj); - if (!stream) return js_mkundef(); if (rs_reader_has_reqs(js, js->this_val)) { ant_value_t release_err = js_make_error_silent(js, JS_ERR_TYPE, "Reader was released"); @@ -617,7 +632,7 @@ static ant_value_t js_rs_reader_release_lock(ant_t *js, ant_value_t *args, int n static ant_value_t js_rs_reader_cancel(ant_t *js, ant_value_t *args, int nargs) { ant_value_t stream_obj = rs_reader_stream(js->this_val); - if (!is_object_type(stream_obj)) { + if (!rs_is_stream(stream_obj)) { ant_value_t p = js_mkpromise(js); js_mkerr_typed(js, JS_ERR_TYPE, "Cannot cancel a released reader"); js_reject_promise(js, p, js->thrown_value); @@ -633,11 +648,11 @@ ant_value_t js_rs_reader_ctor(ant_t *js, ant_value_t *args, int nargs) { if (nargs < 1) return js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStreamDefaultReader requires a stream argument"); ant_value_t stream_obj = args[0]; - if (!is_object_type(stream_obj)) + if (!rs_is_stream(stream_obj)) return js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStreamDefaultReader argument must be a ReadableStream"); + rs_stream_t *stream = rs_get_stream(stream_obj); - if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStreamDefaultReader argument must be a ReadableStream"); - if (is_object_type(rs_stream_reader(stream_obj))) + if (rs_is_reader(rs_stream_reader(stream_obj))) return js_mkerr_typed(js, JS_ERR_TYPE, "ReadableStream is already locked to a reader"); ant_value_t closed = js_mkpromise(js); @@ -663,13 +678,13 @@ ant_value_t js_rs_reader_ctor(ant_t *js, ant_value_t *args, int nargs) { static ant_value_t js_rs_get_locked(ant_t *js, ant_value_t *args, int nargs) { rs_stream_t *stream = rs_get_stream(js->this_val); if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); - return js_bool(is_object_type(rs_stream_reader(js->this_val))); + return js_bool(rs_is_reader(rs_stream_reader(js->this_val))); } static ant_value_t js_rs_cancel(ant_t *js, ant_value_t *args, int nargs) { rs_stream_t *stream = rs_get_stream(js->this_val); if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid ReadableStream"); - if (is_object_type(rs_stream_reader(js->this_val))) { + if (rs_is_reader(rs_stream_reader(js->this_val))) { ant_value_t p = js_mkpromise(js); js_mkerr_typed(js, JS_ERR_TYPE, "Cannot cancel a locked ReadableStream"); js_reject_promise(js, p, js->thrown_value); diff --git a/src/streams/transform.c b/src/streams/transform.c index 95f97f5..f35a4a0 100644 --- a/src/streams/transform.c +++ b/src/streams/transform.c @@ -17,6 +17,30 @@ static ant_value_t g_ts_proto; static ant_value_t g_ts_ctrl_proto; +bool ts_is_controller(ant_value_t obj) { + if (!is_object_type(obj)) return false; + ant_value_t ts_obj = js_get_slot(obj, SLOT_DATA); + return is_object_type(ts_obj) + && vtype(js_get_slot(ts_obj, SLOT_DATA)) == T_NUM + && js_get_slot(ts_obj, SLOT_DEFAULT) == obj; +} + +bool ts_is_stream(ant_value_t obj) { + return is_object_type(obj) + && vtype(js_get_slot(obj, SLOT_DATA)) == T_NUM + && rs_is_stream(js_get_slot(obj, SLOT_ENTRIES)) + && ws_is_stream(js_get_slot(obj, SLOT_CTOR)) + && ts_is_controller(js_get_slot(obj, SLOT_DEFAULT)); +} + +ant_value_t ts_stream_readable(ant_value_t ts_obj) { + return js_get_slot(ts_obj, SLOT_ENTRIES); +} + +ant_value_t ts_stream_writable(ant_value_t ts_obj) { + return js_get_slot(ts_obj, SLOT_CTOR); +} + static void ts_ws_finalize(ant_t *js, ant_object_t *obj) { if (!obj->extra_slots) return; ant_extra_slot_t *entries = (ant_extra_slot_t *)obj->extra_slots; @@ -61,10 +85,6 @@ static void ts_rs_ctrl_finalize(ant_t *js, ant_object_t *obj) { }} } -static inline bool ts_is_valid(ant_value_t ts_obj) { - return vtype(js_get_slot(ts_obj, SLOT_DATA)) == T_NUM; -} - static inline bool ts_get_backpressure(ant_value_t ts_obj) { return js_getnum(js_get_slot(ts_obj, SLOT_DATA)) != 0; } @@ -74,11 +94,11 @@ static inline void ts_set_bp_flag(ant_value_t ts_obj, bool bp) { } static inline ant_value_t ts_readable(ant_value_t ts_obj) { - return js_get_slot(ts_obj, SLOT_ENTRIES); + return ts_stream_readable(ts_obj); } static inline ant_value_t ts_writable(ant_value_t ts_obj) { - return js_get_slot(ts_obj, SLOT_CTOR); + return ts_stream_writable(ts_obj); } static inline ant_value_t ts_bp_promise(ant_value_t ts_obj) { @@ -175,29 +195,19 @@ static ant_value_t ts_take_thrown_or(ant_t *js, ant_value_t fallback) { return err; } -static bool ts_is_thenable(ant_t *js, ant_value_t val) { - if (vtype(val) == T_PROMISE) return true; - if (!is_object_type(val)) return false; - ant_value_t then = js_get(js, val, "then"); - return is_callable(then); -} - -static void ts_chain_thenable(ant_t *js, ant_value_t val, ant_value_t res_fn, ant_value_t rej_fn) { - if (vtype(val) == T_PROMISE) { - ant_value_t then_fn = js_get(js, val, "then"); - if (is_callable(then_fn)) { - ant_value_t then_args[2] = { res_fn, rej_fn }; - ant_value_t then_result = sv_vm_call(js->vm, js, then_fn, val, then_args, 2, NULL, false); - promise_mark_handled(then_result); - }} else { - ant_value_t resolved = js_mkpromise(js); - js_resolve_promise(js, resolved, val); - ant_value_t then_fn = js_get(js, resolved, "then"); - if (is_callable(then_fn)) { - ant_value_t then_args[2] = { res_fn, rej_fn }; - ant_value_t then_result = sv_vm_call(js->vm, js, then_fn, resolved, then_args, 2, NULL, false); - promise_mark_handled(then_result); - }} +static void ts_chain_promise(ant_t *js, ant_value_t val, ant_value_t res_fn, ant_value_t rej_fn) { + ant_value_t promise = val; + if (vtype(promise) != T_PROMISE) { + promise = js_mkpromise(js); + js_resolve_promise(js, promise, val); + } + + ant_value_t then_fn = js_get(js, promise, "then"); + if (!is_callable(then_fn)) return; + + ant_value_t then_args[2] = { res_fn, rej_fn }; + ant_value_t then_result = sv_vm_call(js->vm, js, then_fn, promise, then_args, 2, NULL, false); + promise_mark_handled(then_result); } static void ts_error(ant_t *js, ant_value_t ts_obj, ant_value_t e) { @@ -325,14 +335,9 @@ static ant_value_t ts_ctrl_perform_transform(ant_t *js, ant_value_t ctrl_obj, an ant_value_t wrapper = js_mkobj(js); js_set_slot(wrapper, SLOT_DATA, p); js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); + ant_value_t rej_fn = js_heavy_mkfun(js, ts_transform_reject, wrapper); - - if (ts_is_thenable(js, result)) ts_chain_thenable(js, result, res_fn, rej_fn); - else { - ant_value_t resolved = js_mkpromise(js); - js_resolve_promise(js, resolved, js_mkundef()); - ts_chain_thenable(js, resolved, res_fn, rej_fn); - } + ts_chain_promise(js, result, res_fn, rej_fn); } else { ant_value_t enqueue_result = ts_ctrl_enqueue(js, ctrl_obj, chunk); if (is_err(enqueue_result)) { @@ -390,15 +395,10 @@ static ant_value_t ts_run_cancel_algorithm(ant_t *js, ant_value_t ts_obj, ant_va ant_value_t wrapper = js_mkobj(js); js_set_slot(wrapper, SLOT_DATA, p); js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); + ant_value_t res_fn = js_heavy_mkfun(js, ts_cancel_base_resolve, wrapper); ant_value_t rej_fn = js_heavy_mkfun(js, ts_cancel_base_reject, wrapper); - - if (ts_is_thenable(js, result)) ts_chain_thenable(js, result, res_fn, rej_fn); - else { - ant_value_t resolved = js_mkpromise(js); - js_resolve_promise(js, resolved, js_mkundef()); - ts_chain_thenable(js, resolved, res_fn, rej_fn); - } + ts_chain_promise(js, result, res_fn, rej_fn); return p; } @@ -448,13 +448,18 @@ static ant_value_t ts_abort_cancel_resolve(ant_t *js, ant_value_t *args, int nar ant_value_t readable = ts_readable(ts_obj); rs_stream_t *rs = rs_get_stream(readable); + if (rs && rs->state == RS_STATE_ERRORED) { js_reject_promise(js, p, rs_stream_error(readable)); return js_mkundef(); } + if (ts_cancel_joined_abort(ts_obj)) js_set_slot(ts_obj, SLOT_RS_SIZE, reason); ts_error(js, ts_obj, reason); - js_resolve_promise(js, p, js_mkundef()); + + if (ts_cancel_joined_abort(ts_obj)) js_reject_promise(js, p, reason); + else js_resolve_promise(js, p, js_mkundef()); + return js_mkundef(); } @@ -519,7 +524,7 @@ static ant_value_t ts_sink_write_bp_resolve(ant_t *js, ant_value_t *args, int na js_set_slot(rej_wrapper, SLOT_DATA, fp); js_set_slot(rej_wrapper, SLOT_ENTRIES, ts_obj); ant_value_t reject_fn = js_heavy_mkfun(js, ts_transform_reject, rej_wrapper); - ts_chain_thenable(js, transform_p, resolve_fn, reject_fn); + ts_chain_promise(js, transform_p, resolve_fn, reject_fn); } return js_mkundef(); } @@ -544,7 +549,7 @@ static ant_value_t ts_sink_write(ant_t *js, ant_value_t *args, int nargs) { js_set_slot(rej_wrapper, SLOT_ENTRIES, ts_obj); ant_value_t rej_fn = js_heavy_mkfun(js, ts_transform_reject, rej_wrapper); ant_value_t bp = ts_bp_promise(ts_obj); - ts_chain_thenable(js, bp, res_fn, rej_fn); + ts_chain_promise(js, bp, res_fn, rej_fn); } else { ant_value_t transform_p = ts_ctrl_perform_transform(js, ctrl_obj, chunk); ant_value_t resolve_fn = js_heavy_mkfun(js, ts_transform_resolve, finish_p); @@ -552,7 +557,7 @@ static ant_value_t ts_sink_write(ant_t *js, ant_value_t *args, int nargs) { js_set_slot(rej_wrapper, SLOT_DATA, finish_p); js_set_slot(rej_wrapper, SLOT_ENTRIES, ts_obj); ant_value_t reject_fn = js_heavy_mkfun(js, ts_transform_reject, rej_wrapper); - ts_chain_thenable(js, transform_p, resolve_fn, reject_fn); + ts_chain_promise(js, transform_p, resolve_fn, reject_fn); } return finish_p; @@ -563,8 +568,10 @@ static ant_value_t ts_sink_abort(ant_t *js, ant_value_t *args, int nargs) { ant_value_t ctrl_obj = ts_controller(ts_obj); ant_value_t cancel_fn = ts_ctrl_cancel_fn(ctrl_obj); ant_value_t reason = (nargs > 0) ? args[0] : js_mkundef(); + bool joined_source_cancel = vtype(ts_cancel_promise(ts_obj)) == T_PROMISE && !ts_cancel_started_by_abort(ts_obj); - if (joined_source_cancel) js_set_slot(ts_obj, SLOT_WS_WRITE, js_true); + if (joined_source_cancel && ts_cancel_has_user_handler(ts_obj)) + js_set_slot(ts_obj, SLOT_WS_WRITE, js_true); ant_value_t p = js_mkpromise(js); ant_value_t base = ts_run_cancel_algorithm(js, ts_obj, cancel_fn, reason, true); @@ -573,9 +580,11 @@ static ant_value_t ts_sink_abort(ant_t *js, ant_value_t *args, int nargs) { js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); js_set_slot(wrapper, SLOT_BUFFER, reason); js_set_slot(wrapper, SLOT_CTOR, ts_cancel_started_by_abort(ts_obj) ? js_true : js_false); + ant_value_t res_fn = js_heavy_mkfun(js, ts_abort_cancel_resolve, wrapper); ant_value_t rej_fn = js_heavy_mkfun(js, ts_abort_cancel_reject, wrapper); - ts_chain_thenable(js, base, res_fn, rej_fn); + ts_chain_promise(js, base, res_fn, rej_fn); + return p; } @@ -630,7 +639,7 @@ static ant_value_t ts_sink_close(ant_t *js, ant_value_t *args, int nargs) { js_set_slot(wrapper, SLOT_ENTRIES, ts_obj); ant_value_t res_fn = js_heavy_mkfun(js, ts_close_cancel_resolve, wrapper); ant_value_t rej_fn = js_heavy_mkfun(js, ts_close_cancel_reject, wrapper); - ts_chain_thenable(js, cancel_p, res_fn, rej_fn); + ts_chain_promise(js, cancel_p, res_fn, rej_fn); return p; } @@ -655,12 +664,7 @@ static ant_value_t ts_sink_close(ant_t *js, ant_value_t *args, int nargs) { ant_value_t res_fn = js_heavy_mkfun(js, ts_sink_close_resolve, wrapper); ant_value_t rej_fn = js_heavy_mkfun(js, ts_sink_close_reject, wrapper); - if (ts_is_thenable(js, result)) ts_chain_thenable(js, result, res_fn, rej_fn); - else { - ant_value_t resolved = js_mkpromise(js); - js_resolve_promise(js, resolved, js_mkundef()); - ts_chain_thenable(js, resolved, res_fn, rej_fn); - } + ts_chain_promise(js, result, res_fn, rej_fn); } else { rs_stream_t *rs = rs_get_stream(readable); if (rs && rs->state == RS_STATE_READABLE) { @@ -719,7 +723,7 @@ static ant_value_t ts_source_cancel(ant_t *js, ant_value_t *args, int nargs) { js_set_slot(wrapper, SLOT_CTOR, ts_cancel_started_by_abort(ts_obj) ? js_true : js_false); ant_value_t res_fn = js_heavy_mkfun(js, ts_source_cancel_resolve, wrapper); ant_value_t rej_fn = js_heavy_mkfun(js, ts_source_cancel_reject, wrapper); - ts_chain_thenable(js, base, res_fn, rej_fn); + ts_chain_promise(js, base, res_fn, rej_fn); return p; } @@ -758,12 +762,12 @@ static ant_value_t js_ts_ctrl_terminate(ant_t *js, ant_value_t *args, int nargs) } static ant_value_t js_ts_get_readable(ant_t *js, ant_value_t *args, int nargs) { - if (!ts_is_valid(js->this_val)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid TransformStream"); + if (!ts_is_stream(js->this_val)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid TransformStream"); return ts_readable(js->this_val); } static ant_value_t js_ts_get_writable(ant_t *js, ant_value_t *args, int nargs) { - if (!ts_is_valid(js->this_val)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid TransformStream"); + if (!ts_is_stream(js->this_val)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid TransformStream"); return ts_writable(js->this_val); } @@ -807,6 +811,8 @@ static ant_value_t ts_start_reject(ant_t *js, ant_value_t *args, int nargs) { ws_stream_t *ws = ws_get_stream(writable); if (ws && ws->state == WS_STATE_WRITABLE) ws_default_controller_error(js, ws_ctrl, e); + else if (ws && ws->state == WS_STATE_ERRORING) + writable_stream_finish_erroring(js, writable); return js_mkundef(); } diff --git a/src/streams/writable.c b/src/streams/writable.c index 73fb069..ee533f4 100644 --- a/src/streams/writable.c +++ b/src/streams/writable.c @@ -18,6 +18,22 @@ ant_value_t g_ws_writer_proto; ant_value_t g_ws_controller_proto; static ant_value_t g_close_sentinel; +bool ws_is_stream(ant_value_t obj) { + return is_object_type(obj) && ws_get_stream(obj) != NULL; +} + +bool ws_is_writer(ant_value_t obj) { + return is_object_type(obj) + && vtype(js_get_slot(obj, SLOT_RS_CLOSED)) == T_PROMISE + && vtype(js_get_slot(obj, SLOT_WS_READY)) == T_PROMISE; +} + +bool ws_is_controller(ant_value_t obj) { + return is_object_type(obj) + && ws_get_controller(obj) != NULL + && ws_is_stream(js_get_slot(obj, SLOT_ENTRIES)); +} + ws_stream_t *ws_get_stream(ant_value_t obj) { ant_value_t s = js_get_slot(obj, SLOT_DATA); if (vtype(s) != T_NUM) return NULL; @@ -179,30 +195,20 @@ static ant_value_t ws_write_reqs_shift(ant_t *js, ant_value_t stream_obj) { return val; } -static bool ws_is_thenable(ant_t *js, ant_value_t val) { - if (vtype(val) == T_PROMISE) return true; - if (!is_object_type(val)) return false; - ant_value_t then = js_get(js, val, "then"); - return is_callable(then); -} - -static void ws_chain_thenable(ant_t *js, ant_value_t val, ant_value_t res_fn, ant_value_t rej_fn) { -if (vtype(val) == T_PROMISE) { - ant_value_t then_fn = js_get(js, val, "then"); - if (is_callable(then_fn)) { - ant_value_t then_args[2] = { res_fn, rej_fn }; - ant_value_t then_result = sv_vm_call(js->vm, js, then_fn, val, then_args, 2, NULL, false); - promise_mark_handled(then_result); - }} else { - ant_value_t resolved = js_mkpromise(js); - js_resolve_promise(js, resolved, val); - ant_value_t then_fn = js_get(js, resolved, "then"); - if (is_callable(then_fn)) { - ant_value_t then_args[2] = { res_fn, rej_fn }; - ant_value_t then_result = sv_vm_call(js->vm, js, then_fn, resolved, then_args, 2, NULL, false); - promise_mark_handled(then_result); +static void ws_chain_promise(ant_t *js, ant_value_t val, ant_value_t res_fn, ant_value_t rej_fn) { + ant_value_t promise = val; + if (vtype(promise) != T_PROMISE) { + promise = js_mkpromise(js); + js_resolve_promise(js, promise, val); } -}} + + ant_value_t then_fn = js_get(js, promise, "then"); + if (!is_callable(then_fn)) return; + + ant_value_t then_args[2] = { res_fn, rej_fn }; + ant_value_t then_result = sv_vm_call(js->vm, js, then_fn, promise, then_args, 2, NULL, false); + promise_mark_handled(then_result); +} static void ws_default_controller_clear_algorithms(ant_value_t ctrl_obj) { js_set_slot(ctrl_obj, SLOT_WS_WRITE, js_mkundef()); @@ -268,12 +274,12 @@ static void writable_stream_start_erroring(ant_t *js, ant_value_t stream_obj, an ant_value_t signal_ac = ws_ctrl_signal(ctrl_obj); if (is_object_type(signal_ac)) { ant_value_t signal = js_get(js, signal_ac, "signal"); - if (is_object_type(signal)) + if (abort_signal_is_signal(signal)) signal_do_abort(js, signal, reason); } ant_value_t writer_obj = ws_stream_writer(stream_obj); - if (is_object_type(writer_obj)) + if (ws_is_writer(writer_obj)) ws_writer_reject_ready_promise(js, writer_obj, reason); if (!writable_stream_has_operation_in_flight(stream_obj) && ctrl && ctrl->started) @@ -288,7 +294,7 @@ static void ws_reject_close_and_closed(ant_t *js, ant_value_t stream_obj) { js_set_slot(stream_obj, SLOT_WS_CLOSE, js_mkundef()); } ant_value_t writer_obj = ws_stream_writer(stream_obj); - if (is_object_type(writer_obj)) + if (ws_is_writer(writer_obj)) ws_writer_reject_closed_promise(js, writer_obj, stored_error); } @@ -365,12 +371,7 @@ void writable_stream_finish_erroring(ant_t *js, ant_value_t stream_obj) { js_set_slot(wrapper, SLOT_ENTRIES, stream_obj); ant_value_t res_fn = js_heavy_mkfun(js, ws_finish_erroring_abort_resolve, wrapper); ant_value_t rej_fn = js_heavy_mkfun(js, ws_finish_erroring_abort_reject, wrapper); - if (ws_is_thenable(js, result)) ws_chain_thenable(js, result, res_fn, rej_fn); - else { - ant_value_t resolved = js_mkpromise(js); - js_resolve_promise(js, resolved, js_mkundef()); - ws_chain_thenable(js, resolved, res_fn, rej_fn); - } + ws_chain_promise(js, result, res_fn, rej_fn); } return; @@ -394,7 +395,7 @@ static void writable_stream_update_backpressure(ant_t *js, ant_value_t stream_ob if (!stream) return; ant_value_t writer_obj = ws_stream_writer(stream_obj); - if (is_object_type(writer_obj) && stream->backpressure != backpressure) { + if (ws_is_writer(writer_obj) && stream->backpressure != backpressure) { if (backpressure) { ant_value_t ready = js_mkpromise(js); promise_mark_handled(ready); @@ -440,7 +441,7 @@ static void writable_stream_finish_in_flight_close(ant_t *js, ant_value_t stream stream->state = WS_STATE_CLOSED; ant_value_t writer_obj = ws_stream_writer(stream_obj); - if (is_object_type(writer_obj)) { + if (ws_is_writer(writer_obj)) { ant_value_t closed = ws_writer_closed(writer_obj); if (!is_undefined(closed)) js_resolve_promise(js, closed, js_mkundef()); } @@ -529,12 +530,7 @@ static void ws_default_controller_process_write(ant_t *js, ant_value_t ctrl_obj, } else { ant_value_t res_fn = js_heavy_mkfun(js, ws_process_write_resolve, ctrl_obj); ant_value_t rej_fn = js_heavy_mkfun(js, ws_process_write_reject, ctrl_obj); - if (ws_is_thenable(js, result)) ws_chain_thenable(js, result, res_fn, rej_fn); - else { - ant_value_t resolved = js_mkpromise(js); - js_resolve_promise(js, resolved, js_mkundef()); - ws_chain_thenable(js, resolved, res_fn, rej_fn); - } + ws_chain_promise(js, result, res_fn, rej_fn); } } @@ -577,13 +573,7 @@ static void ws_default_controller_process_close(ant_t *js, ant_value_t ctrl_obj) } else { ant_value_t res_fn = js_heavy_mkfun(js, ws_process_close_resolve, stream_obj); ant_value_t rej_fn = js_heavy_mkfun(js, ws_process_close_reject, stream_obj); - if (ws_is_thenable(js, result)) { - ws_chain_thenable(js, result, res_fn, rej_fn); - } else { - ant_value_t resolved = js_mkpromise(js); - js_resolve_promise(js, resolved, js_mkundef()); - ws_chain_thenable(js, resolved, res_fn, rej_fn); - } + ws_chain_promise(js, result, res_fn, rej_fn); } } @@ -680,7 +670,7 @@ ant_value_t writable_stream_close(ant_t *js, ant_value_t stream_obj) { js_set_slot(stream_obj, SLOT_WS_CLOSE, promise); ant_value_t writer_obj = ws_stream_writer(stream_obj); - if (is_object_type(writer_obj) && stream->backpressure && stream->state == WS_STATE_WRITABLE) { + if (ws_is_writer(writer_obj) && stream->backpressure && stream->state == WS_STATE_WRITABLE) { ant_value_t ready = ws_writer_ready(writer_obj); if (!is_undefined(ready)) js_resolve_promise(js, ready, js_mkundef()); } @@ -710,7 +700,7 @@ static ant_value_t ws_abort_resolve(ant_t *js, ant_value_t *args, int nargs) { } ant_value_t writer_obj = ws_stream_writer(stream_obj); - if (is_object_type(writer_obj)) ws_writer_reject_closed_promise(js, writer_obj, stored_error); + if (ws_is_writer(writer_obj)) ws_writer_reject_closed_promise(js, writer_obj, stored_error); return js_mkundef(); } @@ -735,7 +725,7 @@ static ant_value_t ws_abort_reject(ant_t *js, ant_value_t *args, int nargs) { } ant_value_t writer_obj = ws_stream_writer(stream_obj); - if (is_object_type(writer_obj)) + if (ws_is_writer(writer_obj)) ws_writer_reject_closed_promise(js, writer_obj, stored_error); return js_mkundef(); @@ -770,7 +760,7 @@ ant_value_t writable_stream_abort(ant_t *js, ant_value_t stream_obj, ant_value_t ant_value_t ws_writer_write(ant_t *js, ant_value_t writer_obj, ant_value_t chunk) { ant_value_t stream_obj = ws_writer_stream(writer_obj); - if (!is_object_type(stream_obj)) { + if (!ws_is_stream(stream_obj)) { ant_value_t p = js_mkpromise(js); js_mkerr_typed(js, JS_ERR_TYPE, "Writer has no stream"); js_reject_promise(js, p, js->thrown_value); @@ -893,7 +883,7 @@ static ant_value_t js_ws_writer_get_ready(ant_t *js, ant_value_t *args, int narg static ant_value_t js_ws_writer_get_desired_size(ant_t *js, ant_value_t *args, int nargs) { ant_value_t stream_obj = ws_writer_stream(js->this_val); - if (!is_object_type(stream_obj)) + if (!ws_is_stream(stream_obj)) return js_mkerr_typed(js, JS_ERR_TYPE, "Writer has no stream"); ws_stream_t *stream = ws_get_stream(stream_obj); if (!stream) return js_mknull(); @@ -907,7 +897,7 @@ static ant_value_t js_ws_writer_get_desired_size(ant_t *js, ant_value_t *args, i static ant_value_t js_ws_writer_abort(ant_t *js, ant_value_t *args, int nargs) { ant_value_t stream_obj = ws_writer_stream(js->this_val); - if (!is_object_type(stream_obj)) { + if (!ws_is_stream(stream_obj)) { ant_value_t p = js_mkpromise(js); js_mkerr_typed(js, JS_ERR_TYPE, "Writer has no stream"); js_reject_promise(js, p, js->thrown_value); @@ -919,7 +909,7 @@ static ant_value_t js_ws_writer_abort(ant_t *js, ant_value_t *args, int nargs) { static ant_value_t js_ws_writer_close(ant_t *js, ant_value_t *args, int nargs) { ant_value_t stream_obj = ws_writer_stream(js->this_val); - if (!is_object_type(stream_obj)) { + if (!ws_is_stream(stream_obj)) { ant_value_t p = js_mkpromise(js); js_mkerr_typed(js, JS_ERR_TYPE, "Writer has no stream"); js_reject_promise(js, p, js->thrown_value); @@ -936,7 +926,7 @@ static ant_value_t js_ws_writer_close(ant_t *js, ant_value_t *args, int nargs) { static ant_value_t js_ws_writer_release_lock(ant_t *js, ant_value_t *args, int nargs) { ant_value_t stream_obj = ws_writer_stream(js->this_val); - if (!is_object_type(stream_obj)) return js_mkundef(); + if (!ws_is_stream(stream_obj)) return js_mkundef(); ant_value_t release_err = js_make_error_silent(js, JS_ERR_TYPE, "Writer was released"); ws_writer_reject_ready_promise(js, js->this_val, release_err); @@ -956,7 +946,7 @@ static ant_value_t js_ws_writer_release_lock(ant_t *js, ant_value_t *args, int n static ant_value_t js_ws_writer_write(ant_t *js, ant_value_t *args, int nargs) { ant_value_t stream_obj = ws_writer_stream(js->this_val); - if (!is_object_type(stream_obj)) { + if (!ws_is_stream(stream_obj)) { ant_value_t p = js_mkpromise(js); js_mkerr_typed(js, JS_ERR_TYPE, "Writer has no stream"); js_reject_promise(js, p, js->thrown_value); @@ -973,12 +963,10 @@ ant_value_t js_ws_writer_ctor(ant_t *js, ant_value_t *args, int nargs) { return js_mkerr_typed(js, JS_ERR_TYPE, "WritableStreamDefaultWriter requires a stream argument"); ant_value_t stream_obj = args[0]; - if (!is_object_type(stream_obj)) + if (!ws_is_stream(stream_obj)) return js_mkerr_typed(js, JS_ERR_TYPE, "WritableStreamDefaultWriter argument must be a WritableStream"); ws_stream_t *stream = ws_get_stream(stream_obj); - if (!stream) - return js_mkerr_typed(js, JS_ERR_TYPE, "WritableStreamDefaultWriter argument must be a WritableStream"); - if (is_object_type(ws_stream_writer(stream_obj))) + if (ws_is_writer(ws_stream_writer(stream_obj))) return js_mkerr_typed(js, JS_ERR_TYPE, "WritableStream is already locked to a writer"); ant_value_t obj = js_mkobj(js); @@ -1027,13 +1015,13 @@ ant_value_t ws_acquire_writer(ant_t *js, ant_value_t stream_obj) { static ant_value_t js_ws_get_locked(ant_t *js, ant_value_t *args, int nargs) { ws_stream_t *stream = ws_get_stream(js->this_val); if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WritableStream"); - return js_bool(is_object_type(ws_stream_writer(js->this_val))); + return js_bool(ws_is_writer(ws_stream_writer(js->this_val))); } static ant_value_t js_ws_abort(ant_t *js, ant_value_t *args, int nargs) { ws_stream_t *stream = ws_get_stream(js->this_val); if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WritableStream"); - if (is_object_type(ws_stream_writer(js->this_val))) { + if (ws_is_writer(ws_stream_writer(js->this_val))) { ant_value_t p = js_mkpromise(js); js_mkerr_typed(js, JS_ERR_TYPE, "Cannot abort a locked WritableStream"); js_reject_promise(js, p, js->thrown_value); @@ -1046,7 +1034,7 @@ static ant_value_t js_ws_abort(ant_t *js, ant_value_t *args, int nargs) { static ant_value_t js_ws_close(ant_t *js, ant_value_t *args, int nargs) { ws_stream_t *stream = ws_get_stream(js->this_val); if (!stream) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WritableStream"); - if (is_object_type(ws_stream_writer(js->this_val))) { + if (ws_is_writer(ws_stream_writer(js->this_val))) { ant_value_t p = js_mkpromise(js); js_mkerr_typed(js, JS_ERR_TYPE, "Cannot close a locked WritableStream"); js_reject_promise(js, p, js->thrown_value);