MIRROR: javascript for 🐜's, a tiny runtime with big ambitions
handle abort in stream
+82
-40
+82
-40
src/streams/writable.c
+82
-40
src/streams/writable.c
···
9
9
10
10
#include "silver/engine.h"
11
11
#include "modules/symbol.h"
12
+
#include "modules/abort.h"
12
13
#include "streams/writable.h"
13
14
14
15
static ant_value_t g_ws_proto;
···
184
185
static void writable_stream_update_backpressure(ant_t *js, ant_value_t stream_obj, bool backpressure);
185
186
static ant_value_t js_ws_writer_ctor(ant_t *js, ant_value_t *args, int nargs);
186
187
188
+
static bool ws_is_thenable(ant_t *js, ant_value_t val) {
189
+
if (vtype(val) == T_PROMISE) return true;
190
+
if (!is_object_type(val)) return false;
191
+
ant_value_t then = js_get(js, val, "then");
192
+
return is_callable(then);
193
+
}
194
+
195
+
static void ws_chain_thenable(ant_t *js, ant_value_t val, ant_value_t res_fn, ant_value_t rej_fn) {
196
+
if (vtype(val) == T_PROMISE) {
197
+
ant_value_t then_fn = js_get(js, val, "then");
198
+
if (is_callable(then_fn)) {
199
+
ant_value_t then_args[2] = { res_fn, rej_fn };
200
+
sv_vm_call(js->vm, js, then_fn, val, then_args, 2, NULL, false);
201
+
}} else {
202
+
ant_value_t resolved = js_mkpromise(js);
203
+
js_resolve_promise(js, resolved, val);
204
+
ant_value_t then_fn = js_get(js, resolved, "then");
205
+
if (is_callable(then_fn)) {
206
+
ant_value_t then_args[2] = { res_fn, rej_fn };
207
+
sv_vm_call(js->vm, js, then_fn, resolved, then_args, 2, NULL, false);
208
+
}
209
+
}}
210
+
187
211
static void ws_default_controller_clear_algorithms(ant_value_t ctrl_obj) {
188
212
js_set_slot(ctrl_obj, SLOT_WS_WRITE, js_mkundef());
189
213
js_set_slot(ctrl_obj, SLOT_WS_CLOSE, js_mkundef());
···
232
256
233
257
stream->state = WS_STATE_ERRORING;
234
258
js_set_slot(stream_obj, SLOT_BUFFER, reason);
259
+
260
+
ant_value_t signal_ac = ws_ctrl_signal(ctrl_obj);
261
+
if (is_object_type(signal_ac)) {
262
+
ant_value_t signal = js_get(js, signal_ac, "signal");
263
+
if (is_object_type(signal))
264
+
signal_do_abort(js, signal, reason);
265
+
}
235
266
236
267
ant_value_t writer_obj = ws_stream_writer(stream_obj);
237
268
if (is_object_type(writer_obj))
···
320
351
ant_value_t thrown = js->thrown_value;
321
352
js_reject_promise(js, abort_promise, is_object_type(thrown) ? thrown : result);
322
353
ws_reject_close_and_closed(js, stream_obj);
323
-
} else if (vtype(result) == T_PROMISE) {
354
+
} else {
324
355
ant_value_t wrapper = js_mkobj(js);
325
356
js_set_slot(wrapper, SLOT_DATA, abort_promise);
326
357
js_set_slot(wrapper, SLOT_ENTRIES, stream_obj);
327
358
ant_value_t res_fn = js_heavy_mkfun(js, ws_finish_erroring_abort_resolve, wrapper);
328
359
ant_value_t rej_fn = js_heavy_mkfun(js, ws_finish_erroring_abort_reject, wrapper);
329
-
ant_value_t then_fn = js_get(js, result, "then");
330
-
if (is_callable(then_fn)) {
331
-
ant_value_t then_args[2] = { res_fn, rej_fn };
332
-
sv_vm_call(js->vm, js, then_fn, result, then_args, 2, NULL, false);
360
+
if (ws_is_thenable(js, result)) ws_chain_thenable(js, result, res_fn, rej_fn);
361
+
else {
362
+
ant_value_t resolved = js_mkpromise(js);
363
+
js_resolve_promise(js, resolved, js_mkundef());
364
+
ws_chain_thenable(js, resolved, res_fn, rej_fn);
333
365
}
334
-
} else {
335
-
js_resolve_promise(js, abort_promise, js_mkundef());
336
-
ws_reject_close_and_closed(js, stream_obj);
337
366
}
367
+
338
368
return;
339
369
}
340
370
···
486
516
if (stream && stream->state == WS_STATE_WRITABLE)
487
517
ws_default_controller_clear_algorithms(ctrl_obj);
488
518
writable_stream_finish_in_flight_write_with_error(js, stream_obj, err);
489
-
} else if (vtype(result) == T_PROMISE) {
490
-
ant_value_t res_fn = js_heavy_mkfun(js, ws_process_write_resolve, ctrl_obj);
491
-
ant_value_t rej_fn = js_heavy_mkfun(js, ws_process_write_reject, ctrl_obj);
492
-
ant_value_t then_fn = js_get(js, result, "then");
493
-
if (is_callable(then_fn)) {
494
-
ant_value_t then_args[2] = { res_fn, rej_fn };
495
-
sv_vm_call(js->vm, js, then_fn, result, then_args, 2, NULL, false);
496
-
}
497
519
} else {
498
-
ant_value_t resolved = js_mkpromise(js);
499
-
js_resolve_promise(js, resolved, js_mkundef());
500
520
ant_value_t res_fn = js_heavy_mkfun(js, ws_process_write_resolve, ctrl_obj);
501
521
ant_value_t rej_fn = js_heavy_mkfun(js, ws_process_write_reject, ctrl_obj);
502
-
ant_value_t then_fn = js_get(js, resolved, "then");
503
-
if (is_callable(then_fn)) {
504
-
ant_value_t then_args[2] = { res_fn, rej_fn };
505
-
sv_vm_call(js->vm, js, then_fn, resolved, then_args, 2, NULL, false);
522
+
if (ws_is_thenable(js, result)) ws_chain_thenable(js, result, res_fn, rej_fn);
523
+
else {
524
+
ant_value_t resolved = js_mkpromise(js);
525
+
js_resolve_promise(js, resolved, js_mkundef());
526
+
ws_chain_thenable(js, resolved, res_fn, rej_fn);
506
527
}
507
528
}
508
529
}
···
543
564
ant_value_t thrown = js->thrown_value;
544
565
ant_value_t err = is_object_type(thrown) ? thrown : result;
545
566
writable_stream_finish_in_flight_close_with_error(js, stream_obj, err);
546
-
} else if (vtype(result) == T_PROMISE) {
547
-
ant_value_t res_fn = js_heavy_mkfun(js, ws_process_close_resolve, stream_obj);
548
-
ant_value_t rej_fn = js_heavy_mkfun(js, ws_process_close_reject, stream_obj);
549
-
ant_value_t then_fn = js_get(js, result, "then");
550
-
if (is_callable(then_fn)) {
551
-
ant_value_t then_args[2] = { res_fn, rej_fn };
552
-
sv_vm_call(js->vm, js, then_fn, result, then_args, 2, NULL, false);
553
-
}
554
567
} else {
555
-
ant_value_t resolved = js_mkpromise(js);
556
-
js_resolve_promise(js, resolved, js_mkundef());
557
568
ant_value_t res_fn = js_heavy_mkfun(js, ws_process_close_resolve, stream_obj);
558
569
ant_value_t rej_fn = js_heavy_mkfun(js, ws_process_close_reject, stream_obj);
559
-
ant_value_t then_fn = js_get(js, resolved, "then");
560
-
if (is_callable(then_fn)) {
561
-
ant_value_t then_args[2] = { res_fn, rej_fn };
562
-
sv_vm_call(js->vm, js, then_fn, resolved, then_args, 2, NULL, false);
570
+
if (ws_is_thenable(js, result)) {
571
+
ws_chain_thenable(js, result, res_fn, rej_fn);
572
+
} else {
573
+
ant_value_t resolved = js_mkpromise(js);
574
+
js_resolve_promise(js, resolved, js_mkundef());
575
+
ws_chain_thenable(js, resolved, res_fn, rej_fn);
563
576
}
564
577
}
565
578
}
···
756
769
ant_value_t p = js_mkpromise(js);
757
770
js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WritableStream");
758
771
js_reject_promise(js, p, js->thrown_value);
772
+
return p;
773
+
}
774
+
775
+
if (stream->state == WS_STATE_CLOSED) {
776
+
ant_value_t p = js_mkpromise(js);
777
+
js_mkerr_typed(js, JS_ERR_TYPE, "Cannot write to a closed WritableStream");
778
+
js_reject_promise(js, p, js->thrown_value);
779
+
return p;
780
+
}
781
+
782
+
if (stream->state == WS_STATE_ERRORED) {
783
+
ant_value_t p = js_mkpromise(js);
784
+
js_reject_promise(js, p, ws_stream_stored_error(stream_obj));
785
+
return p;
786
+
}
787
+
788
+
if (writable_stream_close_queued_or_in_flight(stream_obj)) {
789
+
ant_value_t p = js_mkpromise(js);
790
+
js_mkerr_typed(js, JS_ERR_TYPE, "Cannot write to a closing WritableStream");
791
+
js_reject_promise(js, p, js->thrown_value);
792
+
return p;
793
+
}
794
+
795
+
if (stream->state == WS_STATE_ERRORING) {
796
+
ant_value_t p = js_mkpromise(js);
797
+
js_reject_promise(js, p, ws_stream_stored_error(stream_obj));
759
798
return p;
760
799
}
761
800
···
1103
1142
ant_value_t start_fn = js_mkundef();
1104
1143
1105
1144
if (is_object_type(underlying_sink)) {
1106
-
ant_value_t wv = js_get(js, underlying_sink, "write");
1145
+
ant_value_t wv = js_getprop_fallback(js, underlying_sink, "write");
1107
1146
if (is_err(wv)) return wv;
1108
1147
if (!is_undefined(wv)) {
1109
1148
if (!is_callable(wv)) return js_mkerr_typed(js, JS_ERR_TYPE, "write must be a function");
1110
1149
write_fn = wv;
1111
1150
}
1112
-
ant_value_t cv = js_get(js, underlying_sink, "close");
1151
+
1152
+
ant_value_t cv = js_getprop_fallback(js, underlying_sink, "close");
1113
1153
if (is_err(cv)) return cv;
1114
1154
if (!is_undefined(cv)) {
1115
1155
if (!is_callable(cv)) return js_mkerr_typed(js, JS_ERR_TYPE, "close must be a function");
1116
1156
close_fn = cv;
1117
1157
}
1118
-
ant_value_t av = js_get(js, underlying_sink, "abort");
1158
+
1159
+
ant_value_t av = js_getprop_fallback(js, underlying_sink, "abort");
1119
1160
if (is_err(av)) return av;
1120
1161
if (!is_undefined(av)) {
1121
1162
if (!is_callable(av)) return js_mkerr_typed(js, JS_ERR_TYPE, "abort must be a function");
1122
1163
abort_fn = av;
1123
1164
}
1124
-
ant_value_t sv = js_get(js, underlying_sink, "start");
1165
+
1166
+
ant_value_t sv = js_getprop_fallback(js, underlying_sink, "start");
1125
1167
if (is_err(sv)) return sv;
1126
1168
if (!is_undefined(sv)) {
1127
1169
if (!is_callable(sv)) return js_mkerr_typed(js, JS_ERR_TYPE, "start must be a function");