Something went wrong. Try again.
javascript for 🐜's, a tiny runtime with big ambitions antjs.org
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977#include <stdlib.h>#include <string.h>
#include "ant.h"#include "cfunc.h"#include "ptr.h"#include "errors.h"#include "modules/timer.h"#include "internal.h"#include "silver/call.h"#include "esm/loader.h"#include "gc/roots.h"
#include "modules/assert.h"#include "modules/buffer.h"#include "modules/events.h"#include "modules/stream.h"#include "modules/iterator.h"#include "modules/string_decoder.h"#include "modules/symbol.h"#include "streams/readable.h"
static constexpr uint32_t STREAM_NATIVE_TAG = Ant_NativeTag("STRM");
static ant_value_t stream_noop(ant_params_t) { return js_mkundef();}
static bool stream_is_instance(ant_value_t value) { return is_object_type(value) && js_check_native_tag(value, STREAM_NATIVE_TAG);}
static inline void stream_set_end_callback(ant_t *js, ant_value_t stream_obj, ant_value_t callback) { js_set_slot_wb(js, stream_obj, SLOT_AUX, callback);}
static stream_private_state_t *stream_private_state(ant_value_t stream_obj) { if (!stream_is_instance(stream_obj)) return NULL; return (stream_private_state_t *)js_get_native(stream_obj, STREAM_NATIVE_TAG);}
static ant_value_t stream_require_this(ant_t *js, ant_value_t value, const char *label) { if (!stream_is_instance(value)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid %s", label); return value;}
static ant_value_t stream_truthy_or_object(ant_t *js, ant_value_t value) { return js_truthy(js, value) ? value : Ant_Object_CreateBare(js);}
static ant_value_t stream_readable_state(ant_t *js, ant_value_t stream_obj) { return js_get(js, stream_obj, "_readableState");}
static ant_value_t stream_writable_state(ant_t *js, ant_value_t stream_obj) { return js_get(js, stream_obj, "_writableState");}
static ant_value_t stream_pipes(ant_t *js, ant_value_t stream_obj) { return js_get(js, stream_obj, "_pipes");}
static bool stream_key_is_cstr(ant_t *js, ant_value_t value, const char *expected) { size_t len = 0; const char *s = NULL; if (vtype(value) != kTypeString) return false; s = js_getstr(js, value, &len); return s && len == strlen(expected) && memcmp(s, expected, len) == 0;}
static ant_value_t stream_event_key(ant_t *js, ant_value_t value) { uint8_t t = vtype(value); if (t == kTypeString || t == kTypeSymbol) return value; return js_mkerr(js, "event must be a string or Symbol");}
void *stream_get_attached_state(ant_value_t stream_obj) { stream_private_state_t *priv = stream_private_state(stream_obj); return priv ? priv->attached_state : NULL;}
void stream_set_attached_state( ant_value_t stream_obj, void *state, stream_finalize_fn finalize) { stream_private_state_t *priv = stream_private_state(stream_obj); if (!priv) return; priv->attached_state = state; priv->attached_state_finalize = finalize;}
void stream_clear_attached_state(ant_value_t stream_obj) { stream_private_state_t *priv = stream_private_state(stream_obj); if (!priv) return; priv->attached_state = NULL; priv->attached_state_finalize = NULL;}
static void stream_finalize(ant_t *js, ant_object_t *obj) { ant_value_t stream_obj = js_obj_from_ptr(obj); stream_private_state_t *priv = stream_private_state(stream_obj); if (!priv) return; js_clear_native(stream_obj, STREAM_NATIVE_TAG); if (priv->attached_state && priv->attached_state_finalize) priv->attached_state_finalize(js, stream_obj, priv->attached_state); free(priv);}
static ant_value_t stream_call( ant_t *js, ant_value_t fn, ant_value_t this_val, ant_value_t *args, int nargs) { if (!is_callable(fn)) return js_mkundef(); if (sv_check_c_stack_overflow(js)) return Ant_Error_CreateStackOverflow(js);
return sv_vm_call_mode( js->vm, js, fn, this_val, args, nargs, NULL, SV_CALL_MODE_NORMAL, js_mkundef() );}
static ant_value_t stream_call_prop( ant_t *js, ant_value_t target, const char *name, ant_value_t *args, int nargs) { ant_value_t fn = js_getprop_fallback(js, target, name); if (is_err(fn)) return fn; if (!is_callable(fn)) return js_mkundef(); return stream_call(js, fn, target, args, nargs);}
static void stream_call_callback(ant_t *js, ant_value_t fn, ant_value_t *args, int nargs) { if (!is_callable(fn)) return; stream_call(js, fn, js_mkundef(), args, nargs);}
static void stream_schedule_microtask(ant_t *js, ant_cfunc_t fn, ant_value_t data) { ant_value_t cb = js_heavy_mkfun(js, fn, data); if (!is_err(cb)) queue_microtask(js, cb);}
static ant_value_t stream_buffer_ctor(ant_t *js) { ant_value_t ns = js_esm_import_sync_cstr(js, "buffer", 6); if (is_err(ns)) return ns; return js_get(js, ns, "Buffer");}
static ant_value_t stream_readable_decoder(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj); if (!is_object_type(state)) return js_mkundef(); return js_get(js, state, "decoder");}
static ant_value_t stream_readable_decode_chunk( ant_t *js, ant_value_t stream_obj, ant_value_t chunk, bool flush) { ant_value_t decoder = stream_readable_decoder(js, stream_obj); if (!is_object_type(decoder)) return chunk; return string_decoder_decode_value(js, decoder, chunk, flush);}
static bool stream_value_is_empty_string(ant_t *js, ant_value_t value) { size_t len = 0; if (vtype(value) != kTypeString) return false; (void)js_getstr(js, value, &len); return len == 0;}
static ant_value_t stream_make_buffer(ant_t *js, ant_value_t value, ant_value_t encoding) { ant_value_t buffer_ctor = stream_buffer_ctor(js); ant_value_t from_fn = 0; ant_value_t args[2];
if (is_err(buffer_ctor)) return buffer_ctor; from_fn = js_get(js, buffer_ctor, "from"); if (is_err(from_fn) || !is_callable(from_fn)) return js_mkerr(js, "Buffer.from is not available"); args[0] = value; args[1] = encoding; return stream_call(js, from_fn, buffer_ctor, args, 2);}
static ant_value_t stream_normalize_chunk( ant_t *js, ant_value_t chunk, bool object_mode, ant_value_t encoding) { ant_value_t str_val = 0;
if ( object_mode || is_null(chunk) || is_undefined(chunk) || vtype(chunk) == kTypeTypedArray || buffer_is_binary_source(chunk) ) return chunk;
if (vtype(chunk) == kTypeString) return stream_make_buffer(js, chunk, encoding);
str_val = js_tostring_val(js, chunk); if (is_err(str_val)) return str_val; return stream_make_buffer(js, str_val, encoding);}
static ant_value_t stream_readable_buffer(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj); if (!is_object_type(state)) return js_mkundef(); return js_get(js, state, "buffer");}
static ant_offset_t stream_readable_buffer_head(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj); ant_value_t head = is_object_type(state) ? js_get(js, state, "bufferHead") : js_mkundef(); return vtype(head) == kTypeNumber ? (ant_offset_t)js_getnum(head) : 0;}
static void stream_set_readable_buffer_head(ant_t *js, ant_value_t stream_obj, ant_offset_t head) { ant_value_t state = stream_readable_state(js, stream_obj); if (is_object_type(state)) js_set(js, state, "bufferHead", js_mknum((double)head));}
static ant_offset_t stream_readable_buffer_len(ant_t *js, ant_value_t stream_obj) { ant_value_t buffer = stream_readable_buffer(js, stream_obj); ant_offset_t head = stream_readable_buffer_head(js, stream_obj); ant_offset_t len = vtype(buffer) == kTypeArray ? js_arr_len(js, buffer) : 0; return len > head ? len - head : 0;}
static double stream_chunk_size(ant_t *js, ant_value_t chunk, bool object_mode) { const uint8_t *bytes = NULL; size_t byte_len = 0; if (object_mode) return 1; if (buffer_source_get_bytes(js, chunk, &bytes, &byte_len)) return (double)byte_len; return 1;}
static void stream_compact_readable_buffer(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj); ant_value_t buffer = stream_readable_buffer(js, stream_obj); ant_offset_t head = stream_readable_buffer_head(js, stream_obj); ant_offset_t len = vtype(buffer) == kTypeArray ? js_arr_len(js, buffer) : 0; ant_value_t compact = 0;
if (!is_object_type(state) || vtype(buffer) != kTypeArray) return; if (head == 0) return; if (head >= len) { compact = js_mkarr(js); js_set(js, state, "buffer", compact); js_set(js, state, "bufferHead", js_mknum(0)); return; }
if (head <= 32 && head * 2 < len) return; compact = js_mkarr(js); for (ant_offset_t i = head; i < len; i++) js_arr_push(js, compact, js_arr_get(js, buffer, i)); js_set(js, state, "buffer", compact); js_set(js, state, "bufferHead", js_mknum(0));}
static void stream_buffer_push(ant_t *js, ant_value_t stream_obj, ant_value_t value) { ant_value_t state = stream_readable_state(js, stream_obj); ant_value_t buffer = stream_readable_buffer(js, stream_obj); ant_value_t length = 0; bool object_mode = false;
if (!is_object_type(state) || vtype(buffer) != kTypeArray) return; js_arr_push(js, buffer, value); object_mode = js_truthy(js, js_get(js, state, "objectMode")); length = js_get(js, state, "length"); js_set(js, state, "length", js_mknum((vtype(length) == kTypeNumber ? tod(length) : 0) + stream_chunk_size(js, value, object_mode)));}
static ant_value_t stream_buffer_shift(ant_t *js, ant_value_t stream_obj) { ant_value_t buffer = stream_readable_buffer(js, stream_obj); ant_offset_t head = stream_readable_buffer_head(js, stream_obj); ant_offset_t len = vtype(buffer) == kTypeArray ? js_arr_len(js, buffer) : 0; ant_value_t value = js_mkundef();
if (vtype(buffer) != kTypeArray || head >= len) return js_mkundef(); value = js_arr_get(js, buffer, head); ant_value_t state = stream_readable_state(js, stream_obj); if (is_object_type(state)) { ant_value_t length = js_get(js, state, "length"); bool object_mode = js_truthy(js, js_get(js, state, "objectMode")); double next_length = (vtype(length) == kTypeNumber ? tod(length) : 0) - stream_chunk_size(js, value, object_mode); js_set(js, state, "length", js_mknum(next_length > 0 ? next_length : 0)); js_set(js, state, "dataEmitted", js_true); } stream_set_readable_buffer_head(js, stream_obj, head + 1); stream_compact_readable_buffer(js, stream_obj); return value;}
static bool stream_listener_count_positive(ant_t *js, ant_value_t target, const char *event_name) { ant_value_t args[1]; ant_value_t result = 0;
args[0] = js_mkstr(js, event_name, strlen(event_name)); result = stream_call_prop(js, target, "listenerCount", args, 1); return vtype(result) == kTypeNumber && js_getnum(result) > 0;}
static void stream_remove_listener( ant_t *js, ant_value_t target, const char *event_name, ant_value_t listener) { ant_value_t args[2]; args[0] = js_mkstr(js, event_name, strlen(event_name)); args[1] = listener; stream_call_prop(js, target, "removeListener", args, 2);}
static ant_value_t stream_add_listener( ant_t *js, ant_value_t target, const char *event_name, ant_value_t listener, bool once) { const char *method = once ? "once" : "on"; ant_value_t add = js_getprop_fallback(js, target, method); if (is_err(add)) return add;
if (is_callable(add)) { ant_value_t args[2]; args[0] = js_mkstr(js, event_name, strlen(event_name)); args[1] = listener; return stream_call(js, add, target, args, 2); }
eventemitter_add_listener(js, target, event_name, listener, once); return js_mkundef();}
static ant_value_t stream_get_option(ant_t *js, ant_value_t options, const char *name) { if (!is_object_type(options)) return js_mkundef(); return js_get(js, options, name);}
static double stream_default_high_water_mark(ant_t *js, bool object_mode) { return object_mode ? js->defaults.stream_object_high_water_mark : js->defaults.stream_high_water_mark;}
static double stream_high_water_mark_from_options(ant_t *js, ant_value_t options, bool object_mode) { ant_value_t hwm = stream_get_option(js, options, "highWaterMark"); return (vtype(hwm) == kTypeNumber && js_getnum(hwm) > 0) ? js_getnum(hwm) : stream_default_high_water_mark(js, object_mode);}
static ant_value_t stream_make_base_object(ant_t *js, ant_value_t proto) { ant_value_t obj = Ant_Object_CreateBare(js); stream_private_state_t *priv = calloc(1, sizeof(*priv)); if (is_object_type(proto)) js_set_proto_init(obj, proto); js_set_native(obj, priv, STREAM_NATIVE_TAG); js_set_slot(obj, SLOT_AUX, js_mkundef()); js_set_finalizer(obj, stream_finalize); return obj;}
static void stream_init_base(ant_t *js, ant_value_t obj, ant_value_t raw_options) { ant_value_t pipes = js_mkarr(js); ant_value_t construct_fn = stream_get_option(js, raw_options, "construct"); ant_value_t destroy_fn = stream_get_option(js, raw_options, "destroy"); if (is_callable(construct_fn)) { js_set(js, obj, "_construct", construct_fn); stream_begin_construct(js, obj); } if (is_callable(destroy_fn)) js_set(js, obj, "_destroy", destroy_fn); js_set(js, obj, "readable", js_true); js_set(js, obj, "destroyed", js_false); js_set(js, obj, "_paused", js_false); js_set(js, obj, "_pipes", pipes); js_set(js, obj, "_streamOptions", stream_truthy_or_object(js, raw_options));}
static void stream_init_readable(ant_t *js, ant_value_t obj, ant_value_t raw_options) { ant_value_t options = is_object_type(raw_options) ? raw_options : Ant_Object_CreateBare(js); ant_value_t state = Ant_Object_CreateBare(js); ant_value_t read_fn = stream_get_option(js, options, "read");
bool object_mode = js_truthy(js, stream_get_option(js, options, "objectMode")); double high_water_mark = stream_high_water_mark_from_options(js, options, object_mode);
stream_init_base(js, obj, raw_options); js_set(js, obj, "readable", js_true); js_set(js, obj, "readableEnded", js_false);
js_set(js, state, "objectMode", js_bool(object_mode)); js_set(js, state, "ended", js_false); js_set(js, state, "endEmitted", js_false); js_set(js, state, "destroyed", js_false); js_set(js, state, "endScheduled", js_false); js_set(js, state, "endRequested", js_false); js_set(js, state, "readableListening", js_false); js_set(js, state, "readableScheduled", js_false); js_set(js, state, "dataEmitted", js_false); js_set(js, state, "errored", js_mkundef()); js_set(js, state, "flowing", js_false); js_set(js, state, "flowingReadScheduled", js_false); js_set(js, state, "reading", js_false); js_set(js, state, "highWaterMark", js_mknum(high_water_mark)); js_set(js, state, "length", js_mknum(0)); js_set(js, state, "buffer", js_mkarr(js)); js_set(js, state, "bufferHead", js_mknum(0)); js_set(js, obj, "_readableState", state);
if (is_callable(read_fn)) js_set(js, obj, "_read", read_fn);}
static ant_value_t js_writable_length_getter(ant_params_t) { ant_value_t state = stream_writable_state(js, js_getthis(js)); ant_value_t length = is_object_type(state) ? js_get(js, state, "length") : js_mknum(0); return vtype(length) == kTypeNumber ? length : js_mknum(0);}
static ant_value_t js_writable_need_drain_getter(ant_params_t) { ant_value_t state = stream_writable_state(js, js_getthis(js)); if (!is_object_type(state)) return js_false; return js_bool(js_truthy(js, js_get(js, state, "needDrain")));}
static ant_value_t js_writable_writable_getter(ant_params_t) { ant_value_t stream_obj = js_getthis(js); ant_value_t state = stream_writable_state(js, stream_obj);
if (!is_object_type(state)) return js_false; if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return js_false;
return js_bool(!js_truthy(js, js_get(js, state, "ended")));}
static ant_value_t js_writable_ended_getter(ant_params_t) { ant_value_t state = stream_writable_state(js, js_getthis(js)); return js_bool(is_object_type(state) && js_truthy(js, js_get(js, state, "ended")));}
static ant_value_t js_writable_finished_getter(ant_params_t) { ant_value_t state = stream_writable_state(js, js_getthis(js)); return js_bool(is_object_type(state) && js_truthy(js, js_get(js, state, "finished")));}
static ant_value_t js_writable_hwm_getter(ant_params_t) { ant_value_t state = stream_writable_state(js, js_getthis(js)); ant_value_t hwm = is_object_type(state) ? js_get(js, state, "highWaterMark") : js_mkundef(); return vtype(hwm) == kTypeNumber ? hwm : js_mknum(0);}
static void stream_define_writable_accessors(ant_t *js, ant_value_t proto) { js_set_getter_desc(js, proto, "writable", 8, Ant_Function_Create(js_writable_writable_getter), JS_DESC_C); js_set_getter_desc(js, proto, "writableEnded", 13, Ant_Function_Create(js_writable_ended_getter), JS_DESC_C); js_set_getter_desc(js, proto, "writableFinished", 16, Ant_Function_Create(js_writable_finished_getter), JS_DESC_C); js_set_getter_desc(js, proto, "writableHighWaterMark", 21, Ant_Function_Create(js_writable_hwm_getter), JS_DESC_C); js_set_getter_desc(js, proto, "writableLength", 14, Ant_Function_Create(js_writable_length_getter), JS_DESC_C); js_set_getter_desc(js, proto, "writableNeedDrain", 17, Ant_Function_Create(js_writable_need_drain_getter), JS_DESC_C);}
static void stream_init_writable(ant_t *js, ant_value_t obj, ant_value_t raw_options) { ant_value_t options = is_object_type(raw_options) ? raw_options : Ant_Object_CreateBare(js); ant_value_t state = Ant_Object_CreateBare(js); bool object_mode = js_truthy(js, stream_get_option(js, options, "objectMode")) || js_truthy(js, stream_get_option(js, options, "writableObjectMode")); double high_water_mark = stream_high_water_mark_from_options(js, options, object_mode); ant_value_t write_fn = stream_get_option(js, options, "write"); ant_value_t final_fn = stream_get_option(js, options, "final");
stream_init_base(js, obj, raw_options); js_set(js, obj, "readable", js_false);
js_set(js, state, "objectMode", js_bool(object_mode)); js_set(js, state, "finished", js_false); js_set(js, state, "ended", js_false); js_set(js, state, "errored", js_mkundef()); js_set(js, state, "highWaterMark", js_mknum(high_water_mark)); js_set(js, state, "length", js_mknum(0)); js_set(js, state, "needDrain", js_false); js_set(js, state, "buffer", js_mkarr(js)); js_set(js, state, "bufferHead", js_mknum(0)); js_set(js, obj, "_writableState", state);
if (is_callable(write_fn)) js_set(js, obj, "_write", write_fn); if (is_callable(final_fn)) js_set(js, obj, "_final", final_fn);}
static ant_value_t stream_emit_named(ant_t *js, ant_value_t stream_obj, const char *event_name) { return js_bool(eventemitter_emit_args(js, stream_obj, event_name, NULL, 0));}
static void stream_set_errored(ant_t *js, ant_value_t stream_obj, ant_value_t error) { ant_value_t state = stream_readable_state(js, stream_obj); if (is_object_type(state)) js_set(js, state, "errored", error);
state = stream_writable_state(js, stream_obj); if (is_object_type(state)) js_set(js, state, "errored", error);}
static void stream_emit_error(ant_t *js, ant_value_t stream_obj, ant_value_t error) { ant_value_t args[1]; stream_set_errored(js, stream_obj, error); args[0] = error; eventemitter_emit_args(js, stream_obj, "error", args, 1);}
static void stream_readable_schedule_flowing_tick(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj);
if (!is_object_type(state)) return; if (!js_truthy(js, js_get(js, state, "flowing"))) return; if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return; if (js_truthy(js, js_get(js, state, "flowingReadScheduled"))) return; js_set(js, state, "flowingReadScheduled", js_true); stream_schedule_microtask(js, stream_readable_continue_flowing, stream_obj);}
static void stream_readable_schedule_continue_flowing(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj);
if (!is_object_type(state)) return; if (!js_truthy(js, js_get(js, state, "flowing"))) return; if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return; if (js_truthy(js, js_get(js, state, "ended"))) return; if (stream_readable_buffer_len(js, stream_obj) > 0) return; stream_readable_schedule_flowing_tick(js, stream_obj);}
static ant_value_t stream_set_paused(ant_t *js, ant_value_t stream_obj, bool paused) { js_set(js, stream_obj, "_paused", js_bool(paused)); stream_emit_named(js, stream_obj, paused ? "pause" : "resume"); return stream_obj;}
static ant_value_t js_stream_pause(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "stream"); if (is_err(stream_obj)) return stream_obj; return stream_set_paused(js, stream_obj, true);}
static ant_value_t js_stream_resume(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "stream"); if (is_err(stream_obj)) return stream_obj; return stream_set_paused(js, stream_obj, false);}
static ant_value_t js_stream_is_paused(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "stream"); ant_value_t paused = 0; if (is_err(stream_obj)) return stream_obj; paused = js_get(js, stream_obj, "_paused"); return js_bool(js_truthy(js, paused));}
static void stream_pipe_remove_state(ant_t *js, ant_value_t source, ant_value_t state_obj) { ant_value_t pipes = stream_pipes(js, source); ant_offset_t len = vtype(pipes) == kTypeArray ? js_arr_len(js, pipes) : 0; ant_value_t next = js_mkarr(js);
for (ant_offset_t i = 0; i < len; i++) { ant_value_t item = js_arr_get(js, pipes, i); if (item != state_obj) js_arr_push(js, next, item); }
js_set(js, source, "_pipes", next);}
static void stream_pipe_cleanup(ant_t *js, ant_value_t state_obj) { ant_value_t cleaned = js_get(js, state_obj, "cleaned"); ant_value_t source = js_get(js, state_obj, "source"); ant_value_t dest = js_get(js, state_obj, "dest"); ant_value_t on_data = js_get(js, state_obj, "onData"); ant_value_t on_drain = js_get(js, state_obj, "onDrain"); ant_value_t on_end = js_get(js, state_obj, "onEnd"); ant_value_t on_close = js_get(js, state_obj, "onClose"); ant_value_t on_error = js_get(js, state_obj, "onError");
if (js_truthy(js, cleaned)) return; js_set(js, state_obj, "cleaned", js_true);
if (stream_is_instance(source)) { stream_remove_listener(js, source, "data", on_data); stream_remove_listener(js, source, "end", on_end); stream_remove_listener(js, source, "close", on_close); stream_remove_listener(js, source, "error", on_error); stream_pipe_remove_state(js, source, state_obj); }
if (is_object_type(dest)) stream_remove_listener(js, dest, "drain", on_drain);}
static ant_value_t stream_pipe_on_data(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t source = js_get(js, state_obj, "source"); ant_value_t dest = js_get(js, state_obj, "dest"); ant_value_t result = js_mkundef(); if (!is_object_type(dest)) return js_mkundef(); result = stream_call_prop(js, dest, "write", nargs > 0 ? &args[0] : NULL, nargs > 0 ? 1 : 0); if (is_err(result)) return result;
if (result == js_false) stream_call_prop(js, source, "pause", NULL, 0); return js_mkundef();}
static ant_value_t stream_pipe_on_drain(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t source = js_get(js, state_obj, "source"); stream_call_prop(js, source, "resume", NULL, 0); return js_mkundef();}
static ant_value_t stream_pipe_on_end(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t dest = js_get(js, state_obj, "dest"); bool end_dest = js_truthy(js, js_get(js, state_obj, "end")); stream_pipe_cleanup(js, state_obj); if (end_dest) stream_call_prop(js, dest, "end", NULL, 0); return js_mkundef();}
static ant_value_t stream_pipe_on_close(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); stream_pipe_cleanup(js, state_obj); return js_mkundef();}
static ant_value_t stream_pipe_on_error(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t dest = js_get(js, state_obj, "dest"); stream_pipe_cleanup(js, state_obj); if (is_object_type(dest) && stream_listener_count_positive(js, dest, "error") && nargs > 0) eventemitter_emit_args(js, dest, "error", &args[0], 1); return js_mkundef();}
static ant_value_t js_stream_pipe(ant_params_t) { ant_value_t source = stream_require_this(js, js_getthis(js), "stream"); ant_value_t options = nargs > 1 ? args[1] : js_mkundef(); ant_value_t state_obj = 0; ant_value_t readable_state = 0; bool end_dest = true;
if (is_err(source)) return source; if (nargs < 1 || !is_object_type(args[0])) return js_mkerr(js, "pipe requires a destination stream"); if (is_object_type(options)) { ant_value_t end_val = js_get(js, options, "end"); if (!is_undefined(end_val)) end_dest = end_val != js_false; }
state_obj = Ant_Object_CreateBare(js); js_set(js, state_obj, "source", source); js_set(js, state_obj, "dest", args[0]); js_set(js, state_obj, "end", js_bool(end_dest)); js_set(js, state_obj, "cleaned", js_false); js_set(js, state_obj, "onData", js_heavy_mkfun(js, stream_pipe_on_data, state_obj)); js_set(js, state_obj, "onDrain", js_heavy_mkfun(js, stream_pipe_on_drain, state_obj)); js_set(js, state_obj, "onEnd", js_heavy_mkfun(js, stream_pipe_on_end, state_obj)); js_set(js, state_obj, "onClose", js_heavy_mkfun(js, stream_pipe_on_close, state_obj)); js_set(js, state_obj, "onError", js_heavy_mkfun(js, stream_pipe_on_error, state_obj));
js_arr_push(js, stream_pipes(js, source), state_obj); stream_add_listener(js, source, "data", js_get(js, state_obj, "onData"), false); stream_add_listener(js, source, "end", js_get(js, state_obj, "onEnd"), true); stream_add_listener(js, source, "close", js_get(js, state_obj, "onClose"), true); stream_add_listener(js, source, "error", js_get(js, state_obj, "onError"), false); stream_add_listener(js, args[0], "drain", js_get(js, state_obj, "onDrain"), false); eventemitter_emit_args(js, args[0], "pipe", &source, 1); readable_state = stream_readable_state(js, source); if (is_object_type(readable_state)) js_set(js, readable_state, "flowing", js_true); stream_call_prop(js, source, "resume", NULL, 0);
return args[0];}
static ant_value_t js_stream_unpipe(ant_params_t) { ant_value_t source = stream_require_this(js, js_getthis(js), "stream"); ant_value_t pipes = 0; ant_value_t matches = 0; ant_offset_t len = 0; ant_value_t dest = nargs > 0 ? args[0] : js_mkundef();
if (is_err(source)) return source; pipes = stream_pipes(js, source); if (vtype(pipes) != kTypeArray) return source;
matches = js_mkarr(js); len = js_arr_len(js, pipes); for (ant_offset_t i = 0; i < len; i++) { ant_value_t state_obj = js_arr_get(js, pipes, i); ant_value_t entry_dest = js_get(js, state_obj, "dest"); if (!is_object_type(dest) || entry_dest == dest) js_arr_push(js, matches, state_obj); }
len = js_arr_len(js, matches); for (ant_offset_t i = 0; i < len; i++) stream_pipe_cleanup(js, js_arr_get(js, matches, i)); return source;}
static ant_value_t stream_destroy_emit_tick(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t stream_obj = js_get(js, state_obj, "stream"); ant_value_t destroyed_err = js_get(js, state_obj, "error"); if (!is_null(destroyed_err) && !is_undefined(destroyed_err)) stream_emit_error(js, stream_obj, destroyed_err); stream_emit_named(js, stream_obj, "close"); return js_mkundef();}
static ant_value_t stream_destroy_done(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t destroyed_err = (nargs > 0) ? args[0] : js_mkundef(); js_set(js, state_obj, "error", destroyed_err); stream_schedule_microtask(js, stream_destroy_emit_tick, state_obj); return js_mkundef();}
static ant_value_t stream_once_call(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t fn = js_get(js, state_obj, "fn"); ant_value_t this_val = js_get(js, state_obj, "thisVal"); ant_value_t called = js_get(js, state_obj, "called");
if (js_truthy(js, called)) return js_mkundef(); js_set(js, state_obj, "called", js_true); return stream_call(js, fn, this_val, args, nargs);}
static ant_value_t stream_make_once(ant_t *js, ant_value_t fn, ant_value_t this_val) { ant_value_t state_obj = Ant_Object_CreateBare(js); js_set(js, state_obj, "fn", fn); js_set(js, state_obj, "thisVal", this_val); js_set(js, state_obj, "called", js_false); return js_heavy_mkfun(js, stream_once_call, state_obj);}
static ant_value_t stream_run_destroy(ant_t *js, ant_value_t stream_obj, ant_value_t error) { ant_value_t destroy_fn = 0; ant_value_t done_state = 0; ant_value_t done = 0; ant_value_t destroy_args[2]; ant_value_t result = 0;
done_state = Ant_Object_CreateBare(js); js_set(js, done_state, "stream", stream_obj); done = stream_make_once(js, js_heavy_mkfun(js, stream_destroy_done, done_state), js_mkundef()); destroy_fn = js_getprop_fallback(js, stream_obj, "_destroy");
if (is_callable(destroy_fn)) { destroy_args[0] = is_undefined(error) ? js_mknull() : error; destroy_args[1] = done; result = stream_call(js, destroy_fn, stream_obj, destroy_args, 2); return is_err(result) ? result : stream_obj; }
destroy_args[0] = is_undefined(error) ? js_mknull() : error; stream_call_callback(js, done, destroy_args, 1); return stream_obj;}
static ant_value_t js_stream_destroy(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "stream"); ant_value_t error = nargs > 0 ? args[0] : js_mkundef();
if (is_err(stream_obj)) return stream_obj; if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return stream_obj;
js_set(js, stream_obj, "destroyed", js_true); ant_value_t readable_state = stream_readable_state(js, stream_obj); if (is_object_type(readable_state)) { js_set(js, readable_state, "destroyed", js_true); js_set(js, stream_obj, "readable", js_false); } if (!is_undefined(error) && !is_null(error)) stream_set_errored(js, stream_obj, error); stream_private_state_t *priv = stream_private_state(stream_obj); if (priv && priv->constructing) { priv->destroy_pending = true; js_set_slot_wb(js, stream_obj, SLOT_STREAM_DESTROY_ERROR, error); return stream_obj; }
return stream_run_destroy(js, stream_obj, error);}
static ant_value_t js_readable__read(ant_params_t) { return js_mkundef();}
static ant_value_t stream_readable_start_flowing(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); return stream_readable_begin_flowing(js, stream_obj);}
static ant_value_t stream_readable_read_tick(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); return stream_readable_maybe_read(js, stream_obj);}
static ant_value_t stream_readable_emit_readable(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t state = stream_readable_state(js, stream_obj);
if (!is_object_type(state)) return js_mkundef(); js_set(js, state, "readableScheduled", js_false);
if (js_truthy(js, js_get(js, state, "flowing"))) return js_mkundef(); if (js_truthy(js, js_get(js, state, "endEmitted"))) return js_mkundef(); if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return js_mkundef(); if (stream_readable_buffer_len(js, stream_obj) == 0 && !js_truthy(js, js_get(js, state, "ended"))) return js_mkundef();
stream_emit_named(js, stream_obj, "readable"); return js_mkundef();}
static void stream_readable_schedule_readable(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj);
if (!is_object_type(state)) return; if (!js_truthy(js, js_get(js, state, "readableListening"))) return; if (js_truthy(js, js_get(js, state, "flowing"))) return; if (js_truthy(js, js_get(js, state, "readableScheduled"))) return; js_set(js, state, "readableScheduled", js_true); stream_schedule_microtask(js, stream_readable_emit_readable, stream_obj);}
ant_value_t stream_readable_continue_flowing(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t state = stream_readable_state(js, stream_obj);
if (!is_object_type(state)) return js_mkundef(); js_set(js, state, "flowingReadScheduled", js_false);
if (!js_truthy(js, js_get(js, state, "flowing"))) return js_mkundef(); if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return js_mkundef();
stream_readable_maybe_read(js, stream_obj); stream_readable_flush(js, stream_obj); return js_mkundef();}
ant_value_t stream_readable_begin_flowing(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj);
if (!is_object_type(state)) return js_mkundef(); if (!js_truthy(js, js_get(js, state, "flowing"))) return js_mkundef();
sv_call_native_this(js, js_stream_resume, stream_obj, NULL, 0); stream_readable_maybe_read(js, stream_obj); stream_readable_flush(js, stream_obj); return js_mkundef();}
static bool stream_auto_destroy(ant_t *js, ant_value_t stream_obj) { ant_value_t options = js_get(js, stream_obj, "_streamOptions"); ant_value_t auto_destroy = is_object_type(options) ? js_get(js, options, "autoDestroy") : js_mkundef(); return is_undefined(auto_destroy) || js_truthy(js, auto_destroy);}
static ant_value_t stream_half_open_end_tick(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); if (js_truthy(js, js_get(js, stream_obj, "writableEnded")) || js_truthy(js, js_get(js, stream_obj, "destroyed"))) return js_mkundef(); return stream_call_prop(js, stream_obj, "end", NULL, 0);}
static ant_value_t stream_readable_emit_end(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t state = stream_readable_state(js, stream_obj); ant_value_t tail = 0;
if (!is_object_type(state)) return js_mkundef(); js_set(js, state, "endScheduled", js_false);
if (!js_truthy(js, js_get(js, state, "flowing")) && !js_truthy(js, js_get(js, state, "endRequested"))) return js_mkundef(); if (!js_truthy(js, js_get(js, state, "ended"))) return js_mkundef(); if (stream_readable_buffer_len(js, stream_obj) != 0) return js_mkundef(); if (js_truthy(js, js_get(js, state, "endEmitted"))) return js_mkundef(); if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return js_mkundef();
tail = stream_readable_decode_chunk(js, stream_obj, js_mkundef(), true); if (is_err(tail)) return tail; if (!is_undefined(tail) && !stream_value_is_empty_string(js, tail)) { js_set(js, state, "dataEmitted", js_true); eventemitter_emit_args(js, stream_obj, "data", &tail, 1); } js_set(js, state, "endEmitted", js_true); js_set(js, stream_obj, "readable", js_false); js_set(js, stream_obj, "readableEnded", js_true); stream_emit_named(js, stream_obj, "end");
ant_value_t writable_state = stream_writable_state(js, stream_obj); bool writable_done = !is_object_type(writable_state) || js_truthy(js, js_get(js, writable_state, "finished")); if (writable_done && stream_auto_destroy(js, stream_obj)) sv_call_native_this(js, js_stream_destroy, stream_obj, NULL, 0); else if (!writable_done && js_get(js, stream_obj, "allowHalfOpen") == js_false) stream_schedule_microtask(js, stream_half_open_end_tick, stream_obj);
return js_mkundef();}
static void stream_readable_schedule_end(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj);
if (!is_object_type(state)) return; if (js_truthy(js, js_get(js, state, "endScheduled"))) return; js_set(js, state, "endScheduled", js_true); stream_schedule_microtask(js, stream_readable_emit_end, stream_obj);}
static void stream_readable_end_if_drained(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj);
if (!is_object_type(state)) return; if (!js_truthy(js, js_get(js, state, "ended"))) return; if (stream_readable_buffer_len(js, stream_obj) != 0) return; js_set(js, state, "endRequested", js_true); stream_readable_schedule_end(js, stream_obj);}
ant_value_t stream_readable_flush(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj); bool emitted_data = false;
if (!is_object_type(state)) return js_mkundef();
while (js_truthy(js, js_get(js, state, "flowing")) && stream_readable_buffer_len(js, stream_obj) > 0) { ant_value_t chunk = stream_buffer_shift(js, stream_obj); chunk = stream_readable_decode_chunk(js, stream_obj, chunk, false); if (is_err(chunk)) return chunk; emitted_data = true; eventemitter_emit_args(js, stream_obj, "data", &chunk, 1); }
if ( js_truthy(js, js_get(js, state, "flowing")) && js_truthy(js, js_get(js, state, "ended")) && stream_readable_buffer_len(js, stream_obj) == 0 && !js_truthy(js, js_get(js, state, "endEmitted")) ) { stream_readable_schedule_end(js, stream_obj); } else if (emitted_data) stream_readable_schedule_continue_flowing(js, stream_obj);
return js_mkundef();}
ant_value_t stream_readable_maybe_read(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj); ant_value_t read_fn = 0; ant_value_t args[1]; ant_value_t hwm = 0; ant_value_t length = 0;
stream_private_state_t *priv = stream_private_state(stream_obj);
if (!is_object_type(state)) return js_mkundef(); if (priv && priv->constructing) return js_mkundef(); if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return js_mkundef(); if (js_truthy(js, js_get(js, state, "reading"))) return js_mkundef(); if (js_truthy(js, js_get(js, state, "ended"))) return js_mkundef();
hwm = js_get(js, state, "highWaterMark"); length = js_get(js, state, "length"); if (vtype(hwm) == kTypeNumber && vtype(length) == kTypeNumber && tod(length) >= tod(hwm)) return js_mkundef();
read_fn = js_getprop_fallback(js, stream_obj, "_read"); js_set(js, state, "reading", js_true); args[0] = hwm; if (is_callable(read_fn)) stream_call(js, read_fn, stream_obj, args, 1); js_set(js, state, "reading", js_false); return js_mkundef();}
ant_value_t stream_readable_push_value( ant_t *js, ant_value_t stream_obj, ant_value_t chunk, ant_value_t encoding) { ant_value_t state = stream_readable_state(js, stream_obj); ant_value_t normalized = 0;
if (!is_object_type(state)) return js_false; if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return js_false;
if (is_null(chunk)) { js_set(js, state, "ended", js_true); stream_readable_flush(js, stream_obj); stream_readable_schedule_readable(js, stream_obj); return js_false; }
normalized = stream_normalize_chunk( js, chunk, js_truthy(js, js_get(js, state, "objectMode")), encoding ); if (is_err(normalized)) return normalized; stream_buffer_push(js, stream_obj, normalized); if (js_truthy(js, js_get(js, state, "flowing"))) stream_readable_flush(js, stream_obj); else stream_readable_schedule_readable(js, stream_obj);
ant_value_t length = js_get(js, state, "length"); ant_value_t high_water_mark = js_get(js, state, "highWaterMark"); return js_bool( vtype(length) != kTypeNumber || vtype(high_water_mark) != kTypeNumber || tod(length) < tod(high_water_mark) );}
static ant_value_t js_readable_push(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Readable"); ant_value_t chunk = nargs > 0 ? args[0] : js_mkundef(); ant_value_t encoding = nargs > 1 ? args[1] : js_mkundef(); if (is_err(stream_obj)) return stream_obj; return stream_readable_push_value(js, stream_obj, chunk, encoding);}
static ant_value_t js_readable_read(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Readable"); ant_value_t state = 0; ant_value_t chunk = 0;
if (is_err(stream_obj)) return stream_obj; state = stream_readable_state(js, stream_obj); if (!is_object_type(state)) return js_mknull();
if (nargs > 0 && vtype(args[0]) == kTypeNumber && tod(args[0]) == 0.0) { stream_readable_maybe_read(js, stream_obj); return js_mknull(); }
if (stream_readable_buffer_len(js, stream_obj) == 0) stream_readable_maybe_read(js, stream_obj); if (stream_readable_buffer_len(js, stream_obj) == 0) { stream_readable_end_if_drained(js, stream_obj); return js_mknull(); }
chunk = stream_buffer_shift(js, stream_obj); chunk = stream_readable_decode_chunk(js, stream_obj, chunk, false); if (is_err(chunk)) return chunk; if (js_truthy(js, js_get(js, state, "flowing"))) stream_readable_flush(js, stream_obj); else stream_readable_end_if_drained(js, stream_obj);
return chunk;}
static ant_value_t js_readable_set_encoding(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Readable"); ant_value_t state = 0; ant_value_t decoder = 0;
ant_value_t encoding = nargs > 0 && !is_undefined(args[0]) ? args[0] : js->builtins.stream_utf8; ant_value_t encoding_str = 0;
if (is_err(stream_obj)) return stream_obj; state = stream_readable_state(js, stream_obj); if (!is_object_type(state)) return stream_obj;
decoder = string_decoder_create(js, encoding, js_mkundef()); if (is_err(decoder)) return decoder; encoding_str = js_tostring_val(js, encoding); if (is_err(encoding_str)) return encoding_str;
js_set(js, state, "decoder", decoder); js_set(js, stream_obj, "encoding", encoding_str); js_set(js, stream_obj, "readableEncoding", encoding_str); return stream_obj;}
static ant_value_t js_readable_add(ant_native_params_t, bool once) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Readable"); ant_value_t key = 0; ant_value_t state = 0;
if (is_err(stream_obj)) return stream_obj; if (nargs < 2) return js_mkerr(js, "on requires 2 arguments (event, listener)"); key = stream_event_key(js, args[0]);
if (is_err(key)) return key; if (!eventemitter_add_listener_val(js, stream_obj, key, args[1], once)) return js_mkerr(js, "listener must be a function");
state = stream_readable_state(js, stream_obj); if (!is_object_type(state)) return stream_obj;
if (stream_key_is_cstr(js, key, "data") && !js_truthy(js, js_get(js, state, "readableListening"))) { js_set(js, state, "flowing", js_true); stream_schedule_microtask(js, stream_readable_start_flowing, stream_obj); } else if (stream_key_is_cstr(js, key, "readable")) { js_set(js, state, "readableListening", js_true); js_set(js, state, "flowing", js_false); if (stream_readable_buffer_len(js, stream_obj) > 0 || js_truthy(js, js_get(js, state, "ended"))) stream_readable_schedule_readable(js, stream_obj); else stream_schedule_microtask(js, stream_readable_read_tick, stream_obj); }
return stream_obj;}
static ant_value_t js_readable_on(ant_params_t) { return js_readable_add(js, args, nargs, false);}
static ant_value_t js_readable_once(ant_params_t) { return js_readable_add(js, args, nargs, true);}
ant_value_t stream_readable_resume(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj); if (is_object_type(state)) js_set(js, state, "flowing", js_true);
stream_set_paused(js, stream_obj, false); stream_readable_schedule_flowing_tick(js, stream_obj); return stream_obj;}
ant_value_t stream_readable_pause(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_readable_state(js, stream_obj); if (is_object_type(state)) js_set(js, state, "flowing", js_false); return stream_set_paused(js, stream_obj, true);}
static ant_value_t js_readable_resume(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Readable"); if (is_err(stream_obj)) return stream_obj; return stream_readable_resume(js, stream_obj);}
static ant_value_t js_readable_pause(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Readable"); if (is_err(stream_obj)) return stream_obj; return stream_readable_pause(js, stream_obj);}
static ant_value_t js_writable__write(ant_params_t) { ant_value_t callback = nargs > 2 ? args[2] : js_mkundef(); stream_call_callback(js, callback, NULL, 0); return js_mkundef();}
static ant_value_t js_writable__final(ant_params_t) { ant_value_t callback = nargs > 0 ? args[0] : js_mkundef(); stream_call_callback(js, callback, NULL, 0); return js_mkundef();}
typedef enum { STREAM_WRITE_THROWN = -1, STREAM_WRITE_PENDING = 0, STREAM_WRITE_COMPLETE = 1,} stream_write_status_t;
static stream_write_status_t stream_writable_start( ant_t *js, ant_value_t stream_obj, ant_value_t state, ant_value_t chunk, ant_value_t encoding, ant_value_t callback, double chunk_size);
static void stream_writable_set_length( ant_t *js, ant_value_t stream_obj, ant_value_t state, double length) { if (length < 0) length = 0; js_set(js, state, "length", js_mknum(length));}
static ant_offset_t stream_writable_buffer_head(ant_t *js, ant_value_t state) { ant_value_t head = js_get(js, state, "bufferHead"); return vtype(head) == kTypeNumber ? (ant_offset_t)js_getnum(head) : 0;}
static bool stream_writable_has_buffered(ant_t *js, ant_value_t state) { ant_value_t buffer = js_get(js, state, "buffer"); ant_offset_t head = stream_writable_buffer_head(js, state); return vtype(buffer) == kTypeArray && head < js_arr_len(js, buffer);}
static void stream_writable_compact_buffer(ant_t *js, ant_value_t state) { ant_value_t buffer = js_get(js, state, "buffer"); ant_offset_t head = stream_writable_buffer_head(js, state); ant_offset_t len = vtype(buffer) == kTypeArray ? js_arr_len(js, buffer) : 0; ant_value_t compact = 0;
if (vtype(buffer) != kTypeArray || head == 0) return; if (head < len && (head < 32 || head * 2 < len)) return;
compact = js_mkarr(js); for (ant_offset_t i = head; i < len; i++) js_arr_push(js, compact, js_arr_get(js, buffer, i));
js_set(js, state, "buffer", compact); js_set(js, state, "bufferHead", js_mknum(0));}
static void stream_writable_enqueue( ant_t *js, ant_value_t state, ant_value_t chunk, ant_value_t encoding, ant_value_t callback, double chunk_size) { ant_value_t buffer = js_get(js, state, "buffer"); ant_value_t entry = Ant_Object_CreateBare(js);
if (vtype(buffer) != kTypeArray) { buffer = js_mkarr(js); js_set(js, state, "buffer", buffer); js_set(js, state, "bufferHead", js_mknum(0)); }
js_set(js, entry, "chunk", chunk); js_set(js, entry, "encoding", encoding); js_set(js, entry, "callback", callback); js_set(js, entry, "size", js_mknum(chunk_size)); js_arr_push(js, buffer, entry);}
static ant_value_t stream_writable_dequeue(ant_t *js, ant_value_t state) { ant_value_t buffer = js_get(js, state, "buffer"); ant_offset_t head = stream_writable_buffer_head(js, state); ant_offset_t len = vtype(buffer) == kTypeArray ? js_arr_len(js, buffer) : 0; ant_value_t entry = js_mkundef();
if (head >= len) return entry; entry = js_arr_get(js, buffer, head); js_set(js, state, "bufferHead", js_mknum((double)(head + 1))); stream_writable_compact_buffer(js, state); return entry;}
static void stream_writable_fail_buffered(ant_t *js, ant_value_t stream_obj, ant_value_t err) { ant_value_t state = stream_writable_state(js, stream_obj); if (!is_object_type(state)) return;
while (stream_writable_has_buffered(js, state)) { ant_value_t entry = stream_writable_dequeue(js, state); ant_value_t entry_cb = 0;
if (!is_object_type(entry)) break; entry_cb = js_get(js, entry, "callback"); if (is_callable(entry_cb)) stream_call_callback(js, entry_cb, &err, 1); }
stream_writable_set_length(js, stream_obj, state, 0); js_set(js, state, "needDrain", js_false);}
static ant_value_t stream_writable_emit_drain(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); stream_emit_named(js, stream_obj, "drain"); return js_mkundef();}
static ant_value_t stream_writable_drain(ant_t *js, ant_value_t stream_obj) { ant_value_t state = stream_writable_state(js, stream_obj); stream_private_state_t *priv = stream_private_state(stream_obj); ant_value_t next = js_mkundef();
if (priv && priv->draining) return js_mkundef(); if (priv) priv->draining = true; while ( is_object_type(state) && stream_writable_has_buffered(js, state) && !js_truthy(js, js_get(js, stream_obj, "destroyed")) ) { ant_value_t next_size = 0;
next = stream_writable_dequeue(js, state); if (!is_object_type(next)) break; next_size = js_get(js, next, "size");
stream_write_status_t status = stream_writable_start( js, stream_obj, state, js_get(js, next, "chunk"), js_get(js, next, "encoding"), js_get(js, next, "callback"), vtype(next_size) == kTypeNumber ? js_getnum(next_size) : 0 );
if (status == STREAM_WRITE_COMPLETE) continue;
if (status == STREAM_WRITE_THROWN) { if (priv) priv->draining = false; return Ant_Exception_Current(js); }
break; } if (priv) priv->draining = false;
if ( is_object_type(state) && (!priv || !priv->writing) && !stream_writable_has_buffered(js, state) && js_getnum(js_get(js, state, "length")) <= 0 && js_truthy(js, js_get(js, state, "needDrain")) ) { js_set(js, state, "needDrain", js_false);
if (priv && priv->sync) stream_schedule_microtask(js, stream_writable_emit_drain, stream_obj); else stream_emit_named(js, stream_obj, "drain"); } if (priv && !priv->writing && priv->pending_final && !priv->final_started) { ant_value_t end_callback = js_get_slot(stream_obj, SLOT_AUX); priv->pending_final = false; stream_set_end_callback(js, stream_obj, js_mkundef()); return stream_writable_begin_end(js, stream_obj, end_callback); } return js_mkundef();}
static ant_value_t stream_writable_write_done(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t stream_obj = 0;
if (js_truthy(js, js_get(js, state_obj, "done"))) return js_mkundef(); js_set(js, state_obj, "done", js_true); stream_obj = js_get(js, state_obj, "stream"); ant_value_t callback = js_get(js, state_obj, "callback"); ant_value_t state = stream_writable_state(js, stream_obj); stream_private_state_t *priv = stream_private_state(stream_obj); ant_value_t err = nargs > 0 ? args[0] : js_mkundef(); ant_value_t size = js_get(js, state_obj, "size"); double length = 0; bool has_buffered = false;
if (is_object_type(state)) { ant_value_t length_value = js_get(js, state, "length"); length = vtype(length_value) == kTypeNumber ? js_getnum(length_value) : 0; if (vtype(size) == kTypeNumber) length -= js_getnum(size); stream_writable_set_length(js, stream_obj, state, length); has_buffered = stream_writable_has_buffered(js, state); }
if (priv && !has_buffered) priv->writing = false;
if (!is_undefined(err) && !is_null(err)) { ant_value_t destroy_args[1] = { err };
if (priv) { priv->pending_final = false; priv->writing = false; } stream_set_errored(js, stream_obj, err); if (is_callable(callback)) stream_call_callback(js, callback, &err, 1); stream_writable_fail_buffered(js, stream_obj, err); sv_call_native_this(js, js_stream_destroy, stream_obj, destroy_args, 1); return js_mkundef(); }
if (is_callable(callback)) stream_call_callback(js, callback, NULL, 0);
if ( js_truthy(js, js_get(js, stream_obj, "destroyed")) && is_object_type(state) && stream_writable_has_buffered(js, state) ) { ant_value_t destroyed_err = Ant_Error_Create( js, JS_ERR_GENERIC, "Cannot call write after a stream was destroyed" );
js_set(js, destroyed_err, "code", js_mkstr(js, "ERR_STREAM_DESTROYED", 20)); stream_writable_fail_buffered(js, stream_obj, destroyed_err); return js_mkundef(); }
return stream_writable_drain(js, stream_obj);}
static stream_write_status_t stream_writable_start( ant_t *js, ant_value_t stream_obj, ant_value_t state, ant_value_t chunk, ant_value_t encoding, ant_value_t callback, double chunk_size) { ant_value_t done_state = Ant_Object_CreateBare(js); ant_value_t done = 0; ant_value_t write_fn = 0; ant_value_t write_args[3]; stream_private_state_t *priv = stream_private_state(stream_obj);
js_set(js, done_state, "stream", stream_obj); js_set(js, done_state, "callback", callback); js_set(js, done_state, "size", js_mknum(chunk_size)); js_set(js, done_state, "done", js_false);
done = js_heavy_mkfun(js, stream_writable_write_done, done_state); if (priv) priv->writing = true;
write_fn = js_getprop_fallback(js, stream_obj, "_write"); write_args[0] = chunk; write_args[1] = encoding; write_args[2] = done;
bool prev_sync = priv ? priv->sync : false; if (priv) priv->sync = true;
if (is_callable(write_fn)) { ant_value_t result = stream_call(js, write_fn, stream_obj, write_args, 3); if (is_err(result)) { if (priv) priv->sync = prev_sync; return STREAM_WRITE_THROWN; } } else stream_call_callback(js, done, NULL, 0); if (priv) priv->sync = prev_sync; return js_truthy(js, js_get(js, done_state, "done")) ? STREAM_WRITE_COMPLETE : STREAM_WRITE_PENDING;}
static ant_value_t stream_errored_write_cb_tick(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t callback = js_get(js, state_obj, "callback"); ant_value_t error = js_get(js, state_obj, "error"); if (is_callable(callback)) stream_call_callback(js, callback, &error, 1); return js_mkundef();}
static void stream_write_callback_later(ant_t *js, ant_value_t callback, ant_value_t error) { GC_ROOT_SAVE(mark, js); GC_ROOT_PIN(js, callback); GC_ROOT_PIN(js, error); ant_value_t tick_state = Ant_Object_CreateBare(js); GC_ROOT_PIN(js, tick_state); js_set(js, tick_state, "callback", callback); js_set(js, tick_state, "error", error); stream_schedule_microtask(js, stream_errored_write_cb_tick, tick_state); GC_ROOT_RESTORE(js, mark);}
static ant_value_t stream_write_state_error(ant_t *js, bool ending) { const char *code = ending ? "ERR_STREAM_WRITE_AFTER_END" : "ERR_STREAM_DESTROYED"; ant_value_t err = Ant_Error_Create( js, JS_ERR_GENERIC, ending ? "write after end" : "Cannot call write after a stream was destroyed" ); if (is_object_type(err)) js_set(js, err, "code", js_mkstr(js, code, strlen(code))); return err;}
static ant_value_t stream_writable_write_impl( ant_t *js, ant_value_t stream_obj, ant_value_t chunk, ant_value_t encoding, ant_value_t callback, bool allow_after_end) { ant_value_t state = 0; ant_value_t normalized = 0; ant_value_t length_value = 0; ant_value_t high_water_mark = 0; stream_private_state_t *priv = NULL; bool object_mode = false; double chunk_size = 0; double length = 0; double hwm = 0; bool accepted = true;
state = stream_writable_state(js, stream_obj); if (!is_object_type(state)) return js_false;
ant_value_t prior_error = js_get(js, state, "errored"); if (!is_undefined(prior_error) && !is_null(prior_error)) { if (is_callable(callback)) { ant_value_t tick_state = Ant_Object_CreateBare(js); js_set(js, tick_state, "callback", callback); js_set(js, tick_state, "error", prior_error); stream_schedule_microtask(js, stream_errored_write_cb_tick, tick_state); } return js_false; }
bool ending = !allow_after_end && js_truthy(js, js_get(js, stream_obj, "writableEnded")); if (ending || js_truthy(js, js_get(js, stream_obj, "destroyed"))) { ant_value_t err = stream_write_state_error(js, ending); stream_set_errored(js, stream_obj, err); if (is_callable(callback)) stream_write_callback_later(js, callback, err); else stream_emit_error(js, stream_obj, err); return js_false; }
object_mode = js_truthy(js, js_get(js, state, "objectMode")); normalized = stream_normalize_chunk(js, chunk, object_mode, encoding);
if (is_err(normalized)) return normalized; chunk_size = stream_chunk_size(js, normalized, object_mode); length_value = js_get(js, state, "length"); high_water_mark = js_get(js, state, "highWaterMark"); length = (vtype(length_value) == kTypeNumber ? js_getnum(length_value) : 0) + chunk_size; hwm = vtype(high_water_mark) == kTypeNumber ? js_getnum(high_water_mark) : stream_default_high_water_mark(js, object_mode); accepted = length < hwm;
stream_writable_set_length(js, stream_obj, state, length); if (!accepted) js_set(js, state, "needDrain", js_true);
priv = stream_private_state(stream_obj); if (priv && (priv->writing || priv->constructing)) stream_writable_enqueue(js, state, normalized, encoding, callback, chunk_size); else if (stream_writable_start(js, stream_obj, state, normalized, encoding, callback, chunk_size) == STREAM_WRITE_THROWN) return Ant_Exception_Current(js);
return js_bool( accepted && !js_truthy(js, js_get(js, stream_obj, "destroyed")) );}
static ant_value_t js_writable_write(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Writable"); ant_value_t callback = js_mkundef(); ant_value_t encoding = js->builtins.stream_utf8;
if (is_err(stream_obj)) return stream_obj;
if (nargs > 1 && is_callable(args[1])) callback = args[1]; else if (nargs > 1 && !is_undefined(args[1])) encoding = args[1]; if (nargs > 2 && is_callable(args[2])) callback = args[2];
return stream_writable_write_impl( js, stream_obj, nargs > 0 ? args[0] : js_mkundef(), encoding, callback, false );}
static ant_value_t stream_writable_finish_tick(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t stream_obj = js_get(js, state_obj, "stream"); ant_value_t callback = js_get(js, state_obj, "callback"); ant_value_t state = stream_writable_state(js, stream_obj);
if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) { if (is_callable(callback)) stream_call_callback(js, callback, NULL, 0); return js_mkundef(); } if (is_object_type(state)) js_set(js, state, "finished", js_true); if (is_callable(callback)) stream_call_callback(js, callback, NULL, 0); stream_emit_named(js, stream_obj, "finish");
if (!js_truthy(js, js_get(js, stream_obj, "readable")) && stream_auto_destroy(js, stream_obj)) sv_call_native_this(js, js_stream_destroy, stream_obj, NULL, 0);
return js_mkundef();}
static ant_value_t stream_writable_end_done(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t stream_obj = js_get(js, state_obj, "stream"); ant_value_t callback = js_get(js, state_obj, "callback"); ant_value_t err = nargs > 0 ? args[0] : js_mkundef();
if (!is_undefined(err) && !is_null(err)) { ant_value_t destroy_args[1] = { err }; sv_call_native_this(js, js_stream_destroy, stream_obj, destroy_args, 1); if (is_callable(callback)) stream_call_callback(js, callback, &err, 1); return js_mkundef(); }
stream_schedule_microtask(js, stream_writable_finish_tick, state_obj); return js_mkundef();}
ant_value_t stream_writable_begin_end(ant_t *js, ant_value_t stream_obj, ant_value_t callback) { ant_value_t final_fn = 0; ant_value_t final_args[1]; ant_value_t done_state = 0; ant_value_t done = 0; stream_private_state_t *priv = stream_private_state(stream_obj);
done_state = Ant_Object_CreateBare(js); js_set(js, done_state, "stream", stream_obj); js_set(js, done_state, "callback", callback); done = stream_make_once(js, js_heavy_mkfun(js, stream_writable_end_done, done_state), js_mkundef()); if (priv) { priv->final_started = true; priv->pending_final = false; } stream_set_end_callback(js, stream_obj, js_mkundef()); final_fn = js_getprop_fallback(js, stream_obj, "_final"); final_args[0] = done; if (is_callable(final_fn)) stream_call(js, final_fn, stream_obj, final_args, 1); else stream_call_callback(js, done, NULL, 0);
return stream_obj;}
static ant_value_t stream_writable_end_after_write(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t stream_obj = js_get(js, state_obj, "stream"); ant_value_t callback = js_get(js, state_obj, "callback"); stream_private_state_t *priv = stream_private_state(stream_obj); ant_value_t err = nargs > 0 ? args[0] : js_mkundef();
if (!is_undefined(err) && !is_null(err)) { ant_value_t destroy_args[1] = { err }; sv_call_native_this(js, js_stream_destroy, stream_obj, destroy_args, 1); if (is_callable(callback)) stream_call_callback(js, callback, &err, 1); return js_mkundef(); }
if (priv) priv->pending_final = false; stream_set_end_callback(js, stream_obj, js_mkundef());
return stream_writable_begin_end(js, stream_obj, callback);}
static ant_value_t js_writable_end(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Writable"); ant_value_t callback = js_mkundef(); ant_value_t chunk = js_mkundef(); ant_value_t encoding = js_mkundef(); ant_value_t after_write_state = 0; ant_value_t after_write = 0; stream_private_state_t *priv = NULL;
if (is_err(stream_obj)) return stream_obj; if (nargs > 0 && is_callable(args[0])) callback = args[0]; else { if (nargs > 0) chunk = args[0]; if (nargs > 1 && is_callable(args[1])) callback = args[1]; else if (nargs > 1) encoding = args[1]; if (nargs > 2 && is_callable(args[2])) callback = args[2]; }
if (js_truthy(js, js_get(js, stream_obj, "writableEnded"))) { if (is_callable(callback)) stream_call_callback(js, callback, NULL, 0); return stream_obj; }
js_set(js, stream_writable_state(js, stream_obj), "ended", js_true); priv = stream_private_state(stream_obj);
if (!is_undefined(chunk) && !is_null(chunk)) { after_write_state = Ant_Object_CreateBare(js); js_set(js, after_write_state, "stream", stream_obj); js_set(js, after_write_state, "callback", callback); after_write = stream_make_once( js, js_heavy_mkfun(js, stream_writable_end_after_write, after_write_state), js_mkundef() ); ant_value_t result = stream_writable_write_impl(js, stream_obj, chunk, encoding, after_write, true); return is_err(result) ? result : stream_obj; }
if (priv && (priv->writing || priv->constructing) && !priv->final_started) { priv->pending_final = true; stream_set_end_callback(js, stream_obj, callback); return stream_obj; }
return stream_writable_begin_end(js, stream_obj, callback);}
static ant_value_t js_transform__transform(ant_params_t) { ant_value_t callback = nargs > 2 ? args[2] : js_mkundef(); ant_value_t cb_args[2]; cb_args[0] = js_mknull(); cb_args[1] = nargs > 0 ? args[0] : js_mkundef(); stream_call_callback(js, callback, cb_args, 2); return js_mkundef();}
static ant_value_t stream_transform_write_callback(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t stream_obj = js_get(js, state_obj, "stream"); ant_value_t outer_callback = js_get(js, state_obj, "callback");
if (nargs > 0 && !is_null(args[0]) && !is_undefined(args[0])) { if (is_callable(outer_callback)) stream_call(js, outer_callback, stream_obj, &args[0], 1); return js_mkundef(); }
if (nargs > 1 && !is_null(args[1]) && !is_undefined(args[1])) stream_readable_push_value(js, stream_obj, args[1], js_mkundef());
if (is_callable(outer_callback)) stream_call(js, outer_callback, stream_obj, NULL, 0); return js_mkundef();}
static ant_value_t js_transform__write(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Transform"); ant_value_t transform_fn = 0; ant_value_t cb_state = 0; ant_value_t cb = 0; ant_value_t call_args[3];
if (is_err(stream_obj)) return stream_obj; transform_fn = js_getprop_fallback(js, stream_obj, "_transform");
cb_state = Ant_Object_CreateBare(js); js_set(js, cb_state, "stream", stream_obj); js_set(js, cb_state, "callback", nargs > 2 ? args[2] : js_mkundef()); cb = js_heavy_mkfun(js, stream_transform_write_callback, cb_state);
call_args[0] = nargs > 0 ? args[0] : js_mkundef(); call_args[1] = nargs > 1 ? args[1] : js->builtins.stream_utf8; call_args[2] = cb;
return stream_call(js, transform_fn, stream_obj, call_args, 3);}
static ant_value_t stream_transform_final_callback(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t stream_obj = js_get(js, state_obj, "stream"); ant_value_t callback = js_get(js, state_obj, "callback");
if (nargs > 0 && !is_null(args[0]) && !is_undefined(args[0])) { if (is_callable(callback)) stream_call(js, callback, stream_obj, &args[0], 1); return js_mkundef(); }
if (nargs > 1 && !is_null(args[1]) && !is_undefined(args[1])) stream_readable_push_value(js, stream_obj, args[1], js_mkundef()); stream_readable_push_value(js, stream_obj, js_mknull(), js_mkundef()); if (is_callable(callback)) stream_call(js, callback, stream_obj, NULL, 0); return js_mkundef();}
static ant_value_t js_transform__final(ant_params_t) { ant_value_t stream_obj = stream_require_this(js, js_getthis(js), "Transform"); ant_value_t flush_fn = 0; ant_value_t cb_state = 0; ant_value_t cb = 0; ant_value_t call_args[1];
if (is_err(stream_obj)) return stream_obj; flush_fn = js_getprop_fallback(js, stream_obj, "_flush"); if (!is_callable(flush_fn)) { stream_readable_push_value(js, stream_obj, js_mknull(), js_mkundef()); stream_call_callback(js, nargs > 0 ? args[0] : js_mkundef(), NULL, 0); return js_mkundef(); }
cb_state = Ant_Object_CreateBare(js); js_set(js, cb_state, "stream", stream_obj); js_set(js, cb_state, "callback", nargs > 0 ? args[0] : js_mkundef()); cb = js_heavy_mkfun(js, stream_transform_final_callback, cb_state); call_args[0] = cb; return stream_call(js, flush_fn, stream_obj, call_args, 1);}
static ant_value_t js_passthrough__transform(ant_params_t) { return js_transform__transform(js, args, nargs, js_mkundef());}
static ant_value_t stream_finished_cleanup(ant_t *js, ant_value_t state_obj) { ant_value_t stream_obj = js_get(js, state_obj, "stream"); if (stream_is_instance(stream_obj) || is_object_type(stream_obj)) { stream_remove_listener(js, stream_obj, "end", js_get(js, state_obj, "onFinish")); stream_remove_listener(js, stream_obj, "finish", js_get(js, state_obj, "onFinish")); stream_remove_listener(js, stream_obj, "close", js_get(js, state_obj, "onFinish")); stream_remove_listener(js, stream_obj, "error", js_get(js, state_obj, "onError")); } return js_mkundef();}
static ant_value_t stream_finished_fire(ant_t *js, ant_value_t state_obj, ant_value_t error) { ant_value_t called = js_get(js, state_obj, "called"); ant_value_t callback = js_get(js, state_obj, "callback"); ant_value_t cb_args[1];
if (js_truthy(js, called)) return js_mkundef(); js_set(js, state_obj, "called", js_true); stream_finished_cleanup(js, state_obj);
if (is_undefined(error)) stream_call_callback(js, callback, NULL, 0); else { cb_args[0] = error; stream_call_callback(js, callback, cb_args, 1); } return js_mkundef();}
static ant_value_t stream_finished_on_finish(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); return stream_finished_fire(js, state_obj, js_mkundef());}
static ant_value_t stream_finished_on_error(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t error = nargs > 0 ? args[0] : js_mkundef(); return stream_finished_fire(js, state_obj, error);}
static ant_value_t stream_finished_register(ant_t *js, ant_value_t stream_obj, ant_value_t callback) { ant_value_t state_obj = Ant_Object_CreateBare(js); ant_value_t on_finish = 0; ant_value_t on_error = 0;
if (!is_callable(callback)) callback = Ant_Function_Create(stream_noop);
js_set(js, state_obj, "stream", stream_obj); js_set(js, state_obj, "callback", callback); js_set(js, state_obj, "called", js_false); on_finish = js_heavy_mkfun(js, stream_finished_on_finish, state_obj); on_error = js_heavy_mkfun(js, stream_finished_on_error, state_obj); js_set(js, state_obj, "onFinish", on_finish); js_set(js, state_obj, "onError", on_error);
ant_value_t added = stream_add_listener(js, stream_obj, "end", on_finish, false); if (is_err(added)) return added; added = stream_add_listener(js, stream_obj, "finish", on_finish, false); if (is_err(added)) return added; added = stream_add_listener(js, stream_obj, "close", on_finish, false); if (is_err(added)) return added; added = stream_add_listener(js, stream_obj, "error", on_error, false); if (is_err(added)) return added;
return stream_obj;}
static ant_value_t js_stream_finished(ant_params_t) { ant_value_t callback = nargs > 1 ? args[1] : js_mkundef(); if (nargs < 1 || !is_object_type(args[0])) return js_mkerr(js, "finished requires a stream"); return stream_finished_register(js, args[0], callback);}
static ant_value_t stream_pipeline_done(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t callback = js_get(js, state_obj, "callback"); ant_value_t called = js_get(js, state_obj, "called"); ant_value_t cb_args[1];
if (js_truthy(js, called)) return js_mkundef(); js_set(js, state_obj, "called", js_true);
if (nargs > 0 && !is_undefined(args[0])) { cb_args[0] = args[0]; stream_call_callback(js, callback, cb_args, 1); } else stream_call_callback(js, callback, NULL, 0); return js_mkundef();}
static ant_value_t stream_pipeline_error(ant_params_t) { ant_value_t done = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); if (nargs > 0 && !is_undefined(args[0])) stream_call_callback(js, done, &args[0], 1); return js_mkundef();}
static ant_value_t stream_pipeline_schedule_done(ant_params_t) { ant_value_t done = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); stream_call_callback(js, done, NULL, 0); return js_mkundef();}
static ant_value_t js_stream_pipeline(ant_params_t) { int stream_count = nargs; ant_value_t callback = js_mkundef(); ant_value_t done_state = 0; ant_value_t done = 0;
if (nargs > 0 && is_callable(args[nargs - 1])) { callback = args[nargs - 1]; stream_count--; } if (!is_callable(callback)) callback = Ant_Function_Create(stream_noop); if (stream_count <= 0) return js_mkundef();
done_state = Ant_Object_CreateBare(js); js_set(js, done_state, "callback", callback); js_set(js, done_state, "called", js_false); done = js_heavy_mkfun(js, stream_pipeline_done, done_state);
if (stream_count < 2) { stream_schedule_microtask(js, stream_pipeline_schedule_done, done); return args[0]; }
for (int i = 0; i < stream_count - 1; i++) { ant_value_t error_cb = js_heavy_mkfun(js, stream_pipeline_error, done); ant_value_t finished_args[2]; finished_args[0] = args[i]; finished_args[1] = error_cb; ant_value_t registered = js_stream_finished(js, finished_args, 2, js_mkundef()); if (is_err(registered)) return registered; ant_value_t pipe_args[2]; pipe_args[0] = args[i + 1]; pipe_args[1] = js_mkundef(); ant_value_t piped = stream_call_prop(js, args[i], "pipe", pipe_args, 2); if (is_err(piped)) return piped; }
{ ant_value_t finished_args[2]; finished_args[0] = args[stream_count - 1]; finished_args[1] = done; ant_value_t registered = js_stream_finished(js, finished_args, 2, js_mkundef()); if (is_err(registered)) return registered; }
return args[stream_count - 1];}
static ant_value_t stream_promise_callback(ant_params_t) { ant_value_t promise = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); if (nargs > 0 && !is_undefined(args[0])) js_reject_promise(js, promise, args[0]); else js_resolve_promise(js, promise, js_mkundef()); return js_mkundef();}
static ant_value_t js_stream_promises_finished(ant_params_t) { ant_value_t promise = js_mkpromise(js); ant_value_t finished_args[2]; if (nargs < 1 || !is_object_type(args[0])) { js_reject_promise(js, promise, js_mkerr(js, "finished requires a stream")); return promise; } finished_args[0] = args[0]; finished_args[1] = js_heavy_mkfun(js, stream_promise_callback, promise); ant_value_t result = js_stream_finished(js, finished_args, 2, js_mkundef()); if (is_err(result)) js_reject_promise(js, promise, result); return promise;}
static ant_value_t js_stream_promises_pipeline(ant_params_t) { ant_value_t promise = js_mkpromise(js); ant_value_t *call_args = NULL;
if (nargs <= 0) { js_resolve_promise(js, promise, js_mkundef()); return promise; }
call_args = malloc((size_t)(nargs + 1) * sizeof(*call_args)); if (!call_args) { js_reject_promise(js, promise, Ant_Error_Create(js, JS_ERR_GENERIC, "out of memory")); return promise; }
for (int i = 0; i < nargs; i++) call_args[i] = args[i]; GC_ROOT_SAVE(root_mark, js); call_args[nargs] = js_heavy_mkfun(js, stream_promise_callback, promise); GC_ROOT_PIN(js, call_args[nargs]); ant_value_t result = js_stream_pipeline(js, call_args, nargs + 1, js_mkundef()); GC_ROOT_RESTORE(js, root_mark);
free(call_args); if (is_err(result)) js_reject_promise(js, promise, result);
return promise;}
static void stream_release_reader(ant_t *js, ant_value_t state_obj) { ant_value_t reader = js_get(js, state_obj, "reader"); if (!is_object_type(reader)) return; stream_call_prop(js, reader, "releaseLock", NULL, 0);}
static ant_value_t stream_readable_from_step(ant_params_t);
static void stream_readable_from_schedule(ant_t *js, ant_value_t state_obj) { stream_schedule_microtask(js, stream_readable_from_step, state_obj);}
static ant_value_t stream_readable_from_fail(ant_t *js, ant_value_t state_obj, ant_value_t error) { GC_ROOT_SAVE(root_mark, js); if (is_err(error)) error = js_take_thrown(js, error);
GC_ROOT_PIN(js, error); ant_value_t readable = js_get(js, state_obj, "readable");
stream_release_reader(js, state_obj); if (stream_is_instance(readable)) { ant_value_t destroy_args[1] = { error }; sv_call_native_this(js, js_stream_destroy, readable, destroy_args, 1); } GC_ROOT_RESTORE(js, root_mark); return js_mkundef();}
static ant_value_t stream_readable_from_handle_result(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t readable = js_get(js, state_obj, "readable"); ant_value_t result = nargs > 0 ? args[0] : js_mkundef(); ant_value_t done = 0; ant_value_t value = 0;
if (js_truthy(js, js_get(js, readable, "destroyed"))) { stream_release_reader(js, state_obj); return js_mkundef(); }
if (!is_object_type(result)) return stream_readable_from_fail( js, state_obj, Ant_Error_Create(js, JS_ERR_TYPE, "iterator step must be an object")); done = js_get(js, result, "done"); if (is_err(done)) return stream_readable_from_fail(js, state_obj, done); value = js_get(js, result, "value"); if (is_err(value)) return stream_readable_from_fail(js, state_obj, value); if (js_truthy(js, done)) { stream_release_reader(js, state_obj); stream_readable_push_value(js, readable, js_mknull(), js_mkundef()); return js_mkundef(); }
stream_readable_push_value(js, readable, value, js_mkundef()); if (!js_truthy(js, js_get(js, readable, "destroyed"))) stream_readable_from_schedule(js, state_obj); return js_mkundef();}
static ant_value_t stream_readable_from_reject(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t error = nargs > 0 ? args[0] : Ant_Error_Create(js, JS_ERR_GENERIC, "stream iteration failed"); return stream_readable_from_fail(js, state_obj, error);}
static ant_value_t stream_readable_from_step(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t readable = js_get(js, state_obj, "readable"); ant_value_t mode = js_get(js, state_obj, "mode"); ant_value_t iterator = js_get(js, state_obj, "iterator"); ant_value_t next_result = js_mkundef(); ant_value_t on_resolve = js_heavy_mkfun(js, stream_readable_from_handle_result, state_obj); ant_value_t on_reject = js_heavy_mkfun(js, stream_readable_from_reject, state_obj); ant_value_t then_result = 0;
if (js_truthy(js, js_get(js, readable, "destroyed"))) { stream_release_reader(js, state_obj); return js_mkundef(); }
if (stream_key_is_cstr(js, mode, "reader")) next_result = stream_call_prop(js, iterator, "read", NULL, 0); else next_result = stream_call_prop(js, iterator, "next", NULL, 0);
if (is_err(next_result)) return stream_readable_from_fail(js, state_obj, next_result); if (vtype(next_result) == kTypePromise) { then_result = js_promise_then(js, next_result, on_resolve, on_reject); if (is_err(then_result)) return stream_readable_from_fail(js, state_obj, then_result); promise_mark_handled(then_result); return js_mkundef(); }
ant_value_t one_arg[1] = { next_result }; return stream_readable_from_handle_result(js, one_arg, 1, js_mkundef());}
static ant_value_t stream_readable_from_start(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t readable = js_get(js, state_obj, "readable"); ant_value_t source = js_get(js, state_obj, "source"); ant_value_t async_iter_fn = 0; ant_value_t reader_fn = 0; iterator_t it;
if (js_truthy(js, js_get(js, readable, "destroyed"))) return js_mkundef();
if (vtype(source) == kTypeString || (is_object_type(source) && js_instance_of_proto(js, source, js->builtins.buffer_proto))) { stream_readable_push_value(js, readable, source, js_mkundef()); stream_readable_push_value(js, readable, js_mknull(), js_mkundef()); return js_mkundef(); }
async_iter_fn = is_object_type(source) ? js_get_sym(js, source, js->sym.asyncIterator_sym) : js_mkundef(); if (is_err(async_iter_fn)) return stream_readable_from_fail(js, state_obj, async_iter_fn); if (is_callable(async_iter_fn)) { ant_value_t iterator = stream_call(js, async_iter_fn, source, NULL, 0); if (is_err(iterator)) return stream_readable_from_fail(js, state_obj, iterator); js_set(js, state_obj, "iterator", iterator); js_set(js, state_obj, "mode", js_mkstr(js, "async", 5)); stream_readable_from_schedule(js, state_obj); return js_mkundef(); }
if (js_iter_open(js, source, &it)) { ant_value_t value = 0; while (js_iter_next(js, &it, &value)) { if (js_truthy(js, js_get(js, readable, "destroyed"))) break; stream_readable_push_value(js, readable, value, js_mkundef()); } js_iter_close(js, &it); if (Ant_Exception_Pending(js)) return stream_readable_from_fail(js, state_obj, Ant_Exception_Current(js)); if (!js_truthy(js, js_get(js, readable, "destroyed"))) stream_readable_push_value(js, readable, js_mknull(), js_mkundef()); return js_mkundef(); }
if (Ant_Exception_Pending(js)) return stream_readable_from_fail(js, state_obj, Ant_Exception_Current(js)); reader_fn = is_object_type(source) ? js_get(js, source, "getReader") : js_mkundef(); if (is_err(reader_fn)) return stream_readable_from_fail(js, state_obj, reader_fn); if (is_callable(reader_fn)) { ant_value_t reader = stream_call(js, reader_fn, source, NULL, 0); if (is_err(reader)) return stream_readable_from_fail(js, state_obj, reader); js_set(js, state_obj, "reader", reader); js_set(js, state_obj, "iterator", reader); js_set(js, state_obj, "mode", js_mkstr(js, "reader", 6)); stream_readable_from_schedule(js, state_obj); return js_mkundef(); }
if (!is_undefined(source)) stream_readable_push_value(js, readable, source, js_mkundef()); stream_readable_push_value(js, readable, js_mknull(), js_mkundef()); return js_mkundef();}
static ant_value_t stream_readable_from_options(ant_t *js, ant_value_t opts) { ant_value_t options = Ant_Object_CreateBare(js); if (is_object_type(opts)) js_set_proto_init(options, opts); if (vtype(stream_get_option(js, opts, "objectMode")) == kTypeUndefined) js_set(js, options, "objectMode", js_true); if (vtype(stream_get_option(js, opts, "highWaterMark")) == kTypeUndefined) js_set(js, options, "highWaterMark", js_mknum(1)); return options;}
static ant_value_t stream_readable_from_source(ant_t *js, ant_value_t source, ant_value_t options) { ant_value_t readable = stream_construct(js, js->builtins.readable_proto, options, stream_init_readable, js_mkundef()); if (is_err(readable)) return readable;
ant_value_t state_obj = Ant_Object_CreateBare(js); js_set(js, state_obj, "readable", readable); js_set(js, state_obj, "source", source); js_set(js, state_obj, "iterator", js_mkundef()); js_set(js, state_obj, "reader", js_mkundef()); js_set(js, state_obj, "mode", js_mkundef()); stream_schedule_microtask(js, stream_readable_from_start, state_obj);
return readable;}
static ant_value_t js_readable_from(ant_params_t) { ant_value_t options = stream_readable_from_options(js, nargs > 1 ? args[1] : js_mkundef()); return stream_readable_from_source(js, nargs > 0 ? args[0] : js_mkundef(), options);}
static ant_value_t js_readable_from_web(ant_params_t) { return stream_readable_from_source(js, nargs > 0 ? args[0] : js_mkundef(), nargs > 1 ? args[1] : js_mkundef());}
static bool stream_to_web_closed(ant_t *js, ant_value_t state_obj) { return js_truthy(js, js_get(js, state_obj, "closed"));}
static void stream_to_web_cleanup(ant_t *js, ant_value_t state_obj) { ant_value_t source = js_get(js, state_obj, "source"); ant_value_t on_data = js_get(js, state_obj, "onData"); ant_value_t on_end = js_get(js, state_obj, "onEnd"); ant_value_t on_error = js_get(js, state_obj, "onError");
if (!is_object_type(source)) return; if (is_callable(on_data)) stream_remove_listener(js, source, "data", on_data); if (is_callable(on_end)) stream_remove_listener(js, source, "end", on_end); if (is_callable(on_error)) stream_remove_listener(js, source, "error", on_error);}
static ant_value_t stream_to_web_on_data(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t controller = js_get(js, state_obj, "controller"); ant_value_t chunk = nargs > 0 ? args[0] : js_mkundef();
if (stream_to_web_closed(js, state_obj)) return js_mkundef(); return rs_controller_enqueue(js, controller, chunk);}
static ant_value_t stream_to_web_on_end(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t controller = js_get(js, state_obj, "controller");
if (stream_to_web_closed(js, state_obj)) return js_mkundef(); js_set(js, state_obj, "closed", js_true); stream_to_web_cleanup(js, state_obj); rs_controller_close(js, controller); return js_mkundef();}
static ant_value_t stream_to_web_on_error(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t web_stream = js_get(js, state_obj, "webStream"); ant_value_t error = nargs > 0 ? args[0] : Ant_Error_Create(js, JS_ERR_GENERIC, "stream error");
if (stream_to_web_closed(js, state_obj)) return js_mkundef(); js_set(js, state_obj, "closed", js_true); stream_to_web_cleanup(js, state_obj); readable_stream_error(js, web_stream, error); return js_mkundef();}
static ant_value_t stream_to_web_cancel(ant_params_t) { ant_value_t state_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t source = js_get(js, state_obj, "source"); ant_value_t reason = nargs > 0 ? args[0] : js_mkundef(); ant_value_t destroy_fn = 0; ant_value_t cancel_result = 0;
if (stream_to_web_closed(js, state_obj)) return js_mkundef(); js_set(js, state_obj, "closed", js_true); stream_to_web_cleanup(js, state_obj);
if (!is_object_type(source)) return js_mkundef(); destroy_fn = js_getprop_fallback(js, source, "destroy"); if (!is_callable(destroy_fn)) return js_mkundef();
cancel_result = stream_call(js, destroy_fn, source, &reason, 1); return is_err(cancel_result) ? cancel_result : js_mkundef();}
static ant_value_t js_readable_to_web(ant_params_t) { ant_value_t source = nargs > 0 ? args[0] : js_mkundef(); ant_value_t state_obj = 0; ant_value_t cancel_fn = 0; ant_value_t web_stream = 0; ant_value_t controller = 0; ant_value_t on_data = 0; ant_value_t on_end = 0; ant_value_t on_error = 0;
if (!is_object_type(source)) return js_mkerr_typed(js, JS_ERR_TYPE, "Readable.toWeb requires a stream");
state_obj = Ant_Object_CreateBare(js); js_set(js, state_obj, "source", source); js_set(js, state_obj, "closed", js_false);
cancel_fn = js_heavy_mkfun(js, stream_to_web_cancel, state_obj); web_stream = rs_create_stream(js, js_mkundef(), cancel_fn, 1.0); if (is_err(web_stream)) return web_stream;
controller = rs_stream_controller(js, web_stream); js_set(js, state_obj, "webStream", web_stream); js_set(js, state_obj, "controller", controller);
on_data = js_heavy_mkfun(js, stream_to_web_on_data, state_obj); on_end = js_heavy_mkfun(js, stream_to_web_on_end, state_obj); on_error = js_heavy_mkfun(js, stream_to_web_on_error, state_obj);
js_set(js, state_obj, "onData", on_data); js_set(js, state_obj, "onEnd", on_end); js_set(js, state_obj, "onError", on_error);
stream_add_listener(js, source, "data", on_data, false); stream_add_listener(js, source, "end", on_end, true); stream_add_listener(js, source, "error", on_error, true);
stream_call_prop(js, source, "resume", NULL, 0); return web_stream;}
static constexpr double STREAM_ITER_PAUSE_AT = 16;static constexpr double STREAM_ITER_RESUME_AT = 4;
static ant_value_t stream_iter_resolved(ant_t *js, ant_value_t value) { ant_value_t promise = js_mkpromise(js); js_resolve_promise(js, promise, value); return promise;}
static ant_value_t stream_iter_rejected(ant_t *js, ant_value_t error) { ant_value_t promise = js_mkpromise(js); js_reject_promise(js, promise, error); return promise;}
static double stream_iter_queued(ant_t *js, ant_value_t state) { ant_value_t queue = js_get(js, state, "queue"); ant_value_t head = js_get(js, state, "head"); double len = vtype(queue) == kTypeArray ? (double)js_arr_len(js, queue) : 0; return len - (vtype(head) == kTypeNumber ? tod(head) : 0);}
static void stream_iter_detach(ant_t *js, ant_value_t state) { if (js_truthy(js, js_get(js, state, "detached"))) return; js_set(js, state, "detached", js_true);
ant_value_t stream_obj = js_get(js, state, "stream"); static const char *const events[] = { "data", "end", "error", "close" }; static const char *const keys[] = { "onData", "onEnd", "onError", "onClose" }; for (size_t i = 0; i < 4; i++) stream_remove_listener(js, stream_obj, events[i], js_get(js, state, keys[i]));}
static void stream_iter_settle(ant_t *js, ant_value_t state, ant_value_t result, bool reject) { ant_value_t pending = js_get(js, state, "pending"); if (vtype(pending) != kTypePromise) return; js_set(js, state, "pending", js_mkundef()); if (reject) js_reject_promise(js, pending, result); else js_resolve_promise(js, pending, result);}
static ant_value_t stream_iter_on_data(ant_params_t) { ant_value_t state = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t chunk = nargs > 0 ? args[0] : js_mkundef();
if (vtype(js_get(js, state, "pending")) == kTypePromise) { stream_iter_settle(js, state, js_iter_result(js, true, chunk), false); return js_mkundef(); }
js_arr_push(js, js_get(js, state, "queue"), chunk); if (stream_iter_queued(js, state) >= STREAM_ITER_PAUSE_AT && !js_truthy(js, js_get(js, state, "paused"))) { js_set(js, state, "paused", js_true); stream_call_prop(js, js_get(js, state, "stream"), "pause", NULL, 0); } return js_mkundef();}
static ant_value_t stream_iter_on_end(ant_params_t) { ant_value_t state = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); js_set(js, state, "ended", js_true); stream_iter_detach(js, state); stream_iter_settle(js, state, js_iter_result(js, false, js_mkundef()), false); return js_mkundef();}
static ant_value_t stream_iter_on_error(ant_params_t) { ant_value_t state = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t error = nargs > 0 ? args[0] : js_mkundef(); stream_iter_detach(js, state);
if (vtype(js_get(js, state, "pending")) == kTypePromise) { js_set(js, state, "ended", js_true); stream_iter_settle(js, state, error, true); } else { js_set(js, state, "error", error); js_set(js, state, "hasError", js_true); } return js_mkundef();}
static ant_value_t stream_iter_next(ant_params_t) { ant_value_t state = js_get_slot(js_getcurrentfunc(js), SLOT_DATA);
if (stream_iter_queued(js, state) > 0) { ant_value_t queue = js_get(js, state, "queue"); ant_offset_t head = (ant_offset_t)tod(js_get(js, state, "head")); ant_value_t chunk = js_arr_get(js, queue, head); head++;
if (head >= js_arr_len(js, queue)) { js_set(js, state, "queue", js_mkarr(js)); head = 0; } js_set(js, state, "head", js_mknum((double)head));
if ( js_truthy(js, js_get(js, state, "paused")) && stream_iter_queued(js, state) <= STREAM_ITER_RESUME_AT && !js_truthy(js, js_get(js, state, "detached")) ) { js_set(js, state, "paused", js_false); stream_call_prop(js, js_get(js, state, "stream"), "resume", NULL, 0); } return stream_iter_resolved(js, js_iter_result(js, true, chunk)); }
if (js_truthy(js, js_get(js, state, "hasError"))) { ant_value_t error = js_get(js, state, "error"); js_set(js, state, "hasError", js_false); js_set(js, state, "error", js_mkundef()); js_set(js, state, "ended", js_true); return stream_iter_rejected(js, error); }
if (js_truthy(js, js_get(js, state, "ended"))) return stream_iter_resolved(js, js_iter_result(js, false, js_mkundef()));
ant_value_t pending = js_get(js, state, "pending"); if (vtype(pending) == kTypePromise) return pending;
pending = js_mkpromise(js); js_set(js, state, "pending", pending); return pending;}
static ant_value_t stream_iter_return(ant_params_t) { ant_value_t state = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t done = js_iter_result(js, false, nargs > 0 ? args[0] : js_mkundef()); ant_value_t stream_obj = js_get(js, state, "stream");
js_set(js, state, "ended", js_true); js_set(js, state, "queue", js_mkarr(js)); js_set(js, state, "head", js_mknum(0));
stream_iter_detach(js, state); stream_iter_settle(js, state, js_iter_result(js, false, js_mkundef()), false);
if (!js_truthy(js, js_get(js, stream_obj, "destroyed"))) { ant_value_t result = stream_call_prop(js, stream_obj, "destroy", NULL, 0); if (is_err(result)) return stream_iter_rejected(js, js_take_thrown(js, result)); }
return stream_iter_resolved(js, done);}
static ant_value_t js_readable_async_iterator(ant_params_t) { ant_value_t stream_obj = js_getthis(js); if (!is_object_type(stream_obj)) return js_mkerr_typed(js, JS_ERR_TYPE, "Readable[Symbol.asyncIterator] called on incompatible receiver");
ant_value_t state = Ant_Object_CreateBare(js); ant_value_t iterator = Ant_Object_CreateBare(js); GC_ROOT_SAVE(root_mark, js); GC_ROOT_PIN(js, state); GC_ROOT_PIN(js, iterator);
js_set(js, state, "stream", stream_obj); js_set(js, state, "queue", js_mkarr(js)); js_set(js, state, "head", js_mknum(0)); js_set(js, state, "pending", js_mkundef()); js_set(js, state, "error", js_mkundef()); js_set(js, state, "hasError", js_false); js_set(js, state, "paused", js_false); js_set(js, state, "detached", js_false); js_set(js, state, "onData", js_heavy_mkfun(js, stream_iter_on_data, state)); js_set(js, state, "onEnd", js_heavy_mkfun(js, stream_iter_on_end, state)); js_set(js, state, "onError", js_heavy_mkfun(js, stream_iter_on_error, state)); js_set(js, state, "onClose", js_get(js, state, "onEnd"));
bool ended = js_truthy(js, js_get(js, stream_obj, "readableEnded")) || js_truthy(js, js_get(js, stream_obj, "destroyed")); js_set(js, state, "ended", js_bool(ended));
js_set(js, iterator, "next", js_heavy_mkfun(js, stream_iter_next, state)); js_set(js, iterator, "return", js_heavy_mkfun(js, stream_iter_return, state)); js_set_sym(js, iterator, js->sym.asyncIterator_sym, Ant_Function_Create(sym_this_cb));
if (!ended) { static const char *const events[] = { "error", "end", "close", "data" }; static const char *const keys[] = { "onError", "onEnd", "onClose", "onData" }; for (size_t i = 0; i < 4; i++) { ant_value_t added = stream_add_listener(js, stream_obj, events[i], js_get(js, state, keys[i]), false); if (is_err(added)) { GC_ROOT_RESTORE(js, root_mark); return added; } } } else js_set(js, state, "detached", js_true);
GC_ROOT_RESTORE(js, root_mark); return iterator;}
static ant_value_t js_stream_ctor(ant_params_t) { return stream_construct(js, js->builtins.stream_proto, nargs > 0 ? args[0] : js_mkundef(), stream_init_base, call_new_target);}
static ant_value_t js_readable_ctor(ant_params_t) { return stream_construct(js, js->builtins.readable_proto, nargs > 0 ? args[0] : js_mkundef(), stream_init_readable, call_new_target);}
static ant_value_t js_writable_ctor(ant_params_t) { return stream_construct(js, js->builtins.writable_proto, nargs > 0 ? args[0] : js_mkundef(), stream_init_writable, call_new_target);}
static ant_value_t js_writable_has_instance(ant_params_t) { ant_value_t obj = nargs > 0 ? args[0] : js_mkundef(); ant_value_t ctor = js_getthis(js); if (!is_object_type(obj)) return js_false;
ant_value_t proto = js_get(js, ctor, "prototype"); if (is_err(proto)) return proto; if (is_object_type(proto) && js_instance_of_proto(js, obj, proto)) return js_true; if (ctor != js->builtins.writable_ctor) return js_false; return js_bool(js_instance_of_proto(js, obj, js->builtins.duplex_proto));}
static void stream_init_duplex(ant_t *js, ant_value_t obj, ant_value_t options) { stream_init_readable(js, obj, options); stream_init_writable(js, obj, options); js_set(js, obj, "readable", js_true);}
static ant_value_t stream_construct_duplex_like(ant_t *js, ant_value_t base_proto, ant_value_t options, ant_value_t new_target) { ant_value_t obj = stream_construct(js, base_proto, options, stream_init_duplex, new_target); js_set(js, obj, "allowHalfOpen", js_bool(stream_get_option(js, options, "allowHalfOpen") != js_false)); return obj;}
static ant_value_t js_duplex_ctor(ant_params_t) { ant_value_t options = nargs > 0 ? args[0] : js_mkundef(); return stream_construct_duplex_like(js, js->builtins.duplex_proto, options, call_new_target);}
static ant_value_t js_transform_ctor(ant_params_t) { ant_value_t options = nargs > 0 ? args[0] : js_mkundef(); ant_value_t obj = stream_construct_duplex_like(js, js->builtins.stream_transform_proto, options, call_new_target); ant_value_t transform_fn = stream_get_option(js, options, "transform"); ant_value_t flush_fn = stream_get_option(js, options, "flush"); if (is_callable(transform_fn)) js_set(js, obj, "_transform", transform_fn); if (is_callable(flush_fn)) js_set(js, obj, "_flush", flush_fn); return obj;}
static ant_value_t js_passthrough_ctor(ant_params_t) { ant_value_t options = nargs > 0 ? args[0] : js_mkundef(); return stream_construct_duplex_like(js, js->builtins.passthrough_proto, options, call_new_target);}
void stream_init_constructors(ant_t *js) { ant_value_t events = 0; ant_value_t ee_ctor = 0; ant_value_t ee_proto = 0;
if (js->builtins.stream_ctor) return;
events = events_library(js); ee_ctor = js_get(js, events, "EventEmitter"); ee_proto = js_get(js, ee_ctor, "prototype");
js->builtins.stream_proto = Ant_Object_CreateBare(js); js_set_proto_init(js->builtins.stream_proto, ee_proto); js_set(js, js->builtins.stream_proto, "pipe", Ant_Function_CreateWithArity(js_stream_pipe, 2)); js_set(js, js->builtins.stream_proto, "unpipe", Ant_Function_Create(js_stream_unpipe)); js_set(js, js->builtins.stream_proto, "pause", Ant_Function_Create(js_stream_pause)); js_set(js, js->builtins.stream_proto, "resume", Ant_Function_Create(js_stream_resume)); js_set(js, js->builtins.stream_proto, "isPaused", Ant_Function_Create(js_stream_is_paused)); js_set(js, js->builtins.stream_proto, "destroy", Ant_Function_Create(js_stream_destroy)); js_set_tostringtag(js, js->builtins.stream_proto, js_mkstr(js, "Stream", 6)); js->builtins.stream_ctor = Ant_Function_CreateConstructor(js, js_stream_ctor, js->builtins.stream_proto, "Stream", 1);
js->builtins.readable_proto = Ant_Object_CreateBare(js); js_set_proto_init(js->builtins.readable_proto, js->builtins.stream_proto); js_set(js, js->builtins.readable_proto, "_read", Ant_Function_CreateWithArity(js_readable__read, 1)); js_set(js, js->builtins.readable_proto, "push", Ant_Function_CreateWithArity(js_readable_push, 2)); js_set(js, js->builtins.readable_proto, "read", Ant_Function_CreateWithArity(js_readable_read, 1)); js_set(js, js->builtins.readable_proto, "setEncoding", Ant_Function_CreateWithArity(js_readable_set_encoding, 1)); js_set(js, js->builtins.readable_proto, "on", Ant_Function_CreateWithArity(js_readable_on, 2)); js_set_exact(js, js->builtins.readable_proto, "addListener", js_get(js, js->builtins.readable_proto, "on")); js_set(js, js->builtins.readable_proto, "once", Ant_Function_Create(js_readable_once)); js_set(js, js->builtins.readable_proto, "resume", Ant_Function_Create(js_readable_resume)); js_set(js, js->builtins.readable_proto, "pause", Ant_Function_Create(js_readable_pause)); js_set_sym(js, js->builtins.readable_proto, js->sym.asyncIterator_sym, Ant_Function_Create(js_readable_async_iterator)); js_set_tostringtag(js, js->builtins.readable_proto, js_mkstr(js, "Readable", 8)); js->builtins.readable_ctor = Ant_Function_CreateConstructor(js, js_readable_ctor, js->builtins.readable_proto, "Readable", 1); js_set(js, js->builtins.readable_ctor, "from", Ant_Function_Create(js_readable_from)); js_set(js, js->builtins.readable_ctor, "fromWeb", Ant_Function_Create(js_readable_from_web)); js_set(js, js->builtins.readable_ctor, "toWeb", Ant_Function_Create(js_readable_to_web));
js->builtins.writable_proto = Ant_Object_CreateBare(js); js_set_proto_init(js->builtins.writable_proto, js->builtins.stream_proto); js_set(js, js->builtins.writable_proto, "_write", Ant_Function_CreateWithArity(js_writable__write, 3)); js_set(js, js->builtins.writable_proto, "_final", Ant_Function_Create(js_writable__final)); js_set(js, js->builtins.writable_proto, "write", Ant_Function_CreateWithArity(js_writable_write, 3)); stream_define_writable_accessors(js, js->builtins.writable_proto); js_set(js, js->builtins.writable_proto, "end", Ant_Function_CreateWithArity(js_writable_end, 3)); js_set(js, js->builtins.writable_proto, "cork", Ant_Function_Create(stream_noop)); js_set(js, js->builtins.writable_proto, "uncork", Ant_Function_Create(stream_noop)); js_set_tostringtag(js, js->builtins.writable_proto, js_mkstr(js, "Writable", 8)); js->builtins.writable_ctor = Ant_Function_CreateConstructor(js, js_writable_ctor, js->builtins.writable_proto, "Writable", 1); js_set_sym(js, js->builtins.writable_ctor, js->sym.hasInstance_sym, Ant_Function_CreateWithArity(js_writable_has_instance, 1));
js->builtins.duplex_proto = Ant_Object_CreateBare(js); js_set_proto_init(js->builtins.duplex_proto, js->builtins.readable_proto); js_set(js, js->builtins.duplex_proto, "_write", Ant_Function_CreateWithArity(js_writable__write, 3)); js_set(js, js->builtins.duplex_proto, "_final", Ant_Function_Create(js_writable__final)); js_set(js, js->builtins.duplex_proto, "write", Ant_Function_CreateWithArity(js_writable_write, 3)); stream_define_writable_accessors(js, js->builtins.duplex_proto); js_set(js, js->builtins.duplex_proto, "end", Ant_Function_CreateWithArity(js_writable_end, 3)); js_set(js, js->builtins.duplex_proto, "cork", Ant_Function_Create(stream_noop)); js_set(js, js->builtins.duplex_proto, "uncork", Ant_Function_Create(stream_noop)); js_set_tostringtag(js, js->builtins.duplex_proto, js_mkstr(js, "Duplex", 6)); js->builtins.duplex_ctor = Ant_Function_CreateConstructor(js, js_duplex_ctor, js->builtins.duplex_proto, "Duplex", 1);
js->builtins.stream_transform_proto = Ant_Object_CreateBare(js); js_set_proto_init(js->builtins.stream_transform_proto, js->builtins.duplex_proto); js_set(js, js->builtins.stream_transform_proto, "_transform", Ant_Function_CreateWithArity(js_transform__transform, 3)); js_set(js, js->builtins.stream_transform_proto, "_write", Ant_Function_CreateWithArity(js_transform__write, 3)); js_set(js, js->builtins.stream_transform_proto, "_final", Ant_Function_CreateWithArity(js_transform__final, 1)); js_set_tostringtag(js, js->builtins.stream_transform_proto, js_mkstr(js, "Transform", 9)); js->builtins.stream_transform_ctor = Ant_Function_CreateConstructor(js, js_transform_ctor, js->builtins.stream_transform_proto, "Transform", 1);
js->builtins.passthrough_proto = Ant_Object_CreateBare(js); js_set_proto_init(js->builtins.passthrough_proto, js->builtins.stream_transform_proto); js_set(js, js->builtins.passthrough_proto, "_transform", Ant_Function_CreateWithArity(js_passthrough__transform, 3)); js_set_tostringtag(js, js->builtins.passthrough_proto, js_mkstr(js, "PassThrough", 11)); js->builtins.passthrough_ctor = Ant_Function_CreateConstructor(js, js_passthrough_ctor, js->builtins.passthrough_proto, "PassThrough", 1); js->builtins.stream_utf8 = js_mkstr(js, "utf8", 4);}
ant_value_t stream_readable_constructor(ant_t *js) { stream_init_constructors(js); return js->builtins.readable_ctor;}
ant_value_t stream_writable_constructor(ant_t *js) { stream_init_constructors(js); return js->builtins.writable_ctor;}
ant_value_t stream_readable_prototype(ant_t *js) { stream_init_constructors(js); return js->builtins.readable_proto;}
ant_value_t stream_writable_prototype(ant_t *js) { stream_init_constructors(js); return js->builtins.writable_proto;}
ant_value_t stream_duplex_prototype(ant_t *js) { stream_init_constructors(js); return js->builtins.duplex_proto;}
ant_value_t stream_transform_prototype(ant_t *js) { stream_init_constructors(js); return js->builtins.stream_transform_proto;}
ant_value_t stream_construct_readable(ant_t *js, ant_value_t base_proto, ant_value_t options, ant_value_t new_target) { stream_init_constructors(js); return stream_construct(js, base_proto, options, stream_init_readable, new_target);}
ant_value_t stream_construct_writable(ant_t *js, ant_value_t base_proto, ant_value_t options, ant_value_t new_target) { stream_init_constructors(js); return stream_construct(js, base_proto, options, stream_init_writable, new_target);}
ant_value_t stream_construct_duplex(ant_t *js, ant_value_t base_proto, ant_value_t options, ant_value_t new_target) { stream_init_constructors(js); return stream_construct(js, base_proto, options, stream_init_duplex, new_target);}
void stream_init_readable_object(ant_t *js, ant_value_t obj, ant_value_t options) { stream_init_constructors(js); if (!is_object_type(obj)) return; js_set_native(obj, NULL, STREAM_NATIVE_TAG); stream_init_readable(js, obj, options);}
void stream_init_writable_object(ant_t *js, ant_value_t obj, ant_value_t options) { stream_init_constructors(js); if (!is_object_type(obj)) return; js_set_native(obj, NULL, STREAM_NATIVE_TAG); stream_init_writable(js, obj, options);}
void stream_init_duplex_object(ant_t *js, ant_value_t obj, ant_value_t options) { stream_init_constructors(js); if (!is_object_type(obj)) return; js_set_native(obj, NULL, STREAM_NATIVE_TAG); stream_init_readable(js, obj, options); stream_init_writable(js, obj, options);}
ant_value_t stream_readable_push(ant_t *js, ant_value_t stream_obj, ant_value_t chunk, ant_value_t encoding) { stream_init_constructors(js); return stream_readable_push_value(js, stream_obj, chunk, encoding);}
static ant_value_t stream_construct_done(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); stream_private_state_t *priv = stream_private_state(stream_obj); ant_value_t err = nargs > 0 ? args[0] : js_mkundef();
bool failed = !is_undefined(err) && !is_null(err); if (priv) priv->constructing = false;
if (priv && priv->destroy_pending) { ant_value_t pending_err = js_get_slot(stream_obj, SLOT_STREAM_DESTROY_ERROR); priv->destroy_pending = false; js_set_slot(stream_obj, SLOT_STREAM_DESTROY_ERROR, js_mkundef()); if (failed) stream_set_errored(js, stream_obj, err); return stream_run_destroy(js, stream_obj, failed ? err : pending_err); }
if (failed) { sv_call_native_this(js, js_stream_destroy, stream_obj, &err, 1); return js_mkundef(); }
ant_value_t readable_state = stream_readable_state(js, stream_obj); if (is_object_type(readable_state) && ( js_truthy(js, js_get(js, readable_state, "flowing")) || js_truthy(js, js_get(js, readable_state, "readableListening")) )) stream_readable_maybe_read(js, stream_obj);
if (is_object_type(stream_writable_state(js, stream_obj))) return stream_writable_drain(js, stream_obj); return js_mkundef();}
static ant_value_t stream_construct_tick(ant_params_t) { ant_value_t stream_obj = js_get_slot(js_getcurrentfunc(js), SLOT_DATA); ant_value_t done = stream_make_once(js, js_heavy_mkfun(js, stream_construct_done, stream_obj), js_mkundef()); GC_ROOT_SAVE(mark, js); GC_ROOT_PIN(js, done);
ant_value_t result = stream_call_prop(js, stream_obj, "_construct", &done, 1); if (is_err(result)) { ant_value_t err = js_take_thrown(js, result); stream_call_callback(js, done, &err, 1); }
GC_ROOT_RESTORE(js, mark); return js_mkundef();}
void stream_begin_construct(ant_t *js, ant_value_t stream_obj) { stream_private_state_t *priv = stream_private_state(stream_obj); if (!priv || priv->constructing) return; priv->constructing = true; stream_schedule_microtask(js, stream_construct_tick, stream_obj);}
ant_value_t stream_construct( ant_t *js, ant_value_t base_proto, ant_value_t raw_options, void (*init_fn)(ant_t *, ant_value_t, ant_value_t), ant_value_t new_target) { ant_value_t proto = js_instance_proto_from_new_target(js, base_proto, new_target); bool subclassed = is_object_type(proto) && proto != base_proto; ant_value_t obj = stream_make_base_object(js, subclassed ? proto : base_proto);
init_fn(js, obj, raw_options); if (subclassed && is_callable(js_getprop_fallback(js, obj, "_construct"))) stream_begin_construct(js, obj); return obj;}
ant_value_t stream_destroy(ant_t *js, ant_value_t stream_obj, ant_value_t error) { GC_ROOT_SAVE(mark, js); GC_ROOT_PIN(js, error); ant_value_t result = stream_call_prop(js, stream_obj, "destroy", &error, 1); GC_ROOT_RESTORE(js, mark); return result;}
static ant_value_t js_stream_get_default_high_water_mark(ant_params_t) { bool object_mode = nargs > 0 && js_truthy(js, args[0]); return js_mknum(stream_default_high_water_mark(js, object_mode));}
static ant_value_t js_stream_set_default_high_water_mark(ant_params_t) { if (nargs < 2 || vtype(args[1]) != kTypeNumber || js_getnum(args[1]) < 0) return js_mkerr_typed(js, JS_ERR_RANGE, "setDefaultHighWaterMark requires a non-negative number");
bool object_mode = js_truthy(js, args[0]); if (object_mode) js->defaults.stream_object_high_water_mark = js_getnum(args[1]); else js->defaults.stream_high_water_mark = js_getnum(args[1]);
return js_mkundef();}
static ant_value_t js_stream_is_destroyed(ant_params_t) { ant_value_t stream_obj = nargs > 0 ? args[0] : js_mkundef(); if (!is_object_type(stream_obj)) return js_false; return js_bool(js_truthy(js, js_get(js, stream_obj, "destroyed")));}
static ant_value_t js_stream_is_disturbed(ant_params_t) { ant_value_t stream_obj = nargs > 0 ? args[0] : js_mkundef(); ant_value_t state = 0;
if (!is_object_type(stream_obj)) return js_false; state = stream_readable_state(js, stream_obj); if (is_object_type(state)) { if (js_truthy(js, js_get(js, state, "dataEmitted"))) return js_true; if (js_truthy(js, js_get(js, state, "endEmitted"))) return js_false; } return js_bool(js_truthy(js, js_get(js, stream_obj, "destroyed")));}
static ant_value_t js_stream_is_errored(ant_params_t) { ant_value_t stream_obj = nargs > 0 ? args[0] : js_mkundef(); ant_value_t state = 0;
if (!is_object_type(stream_obj)) return js_false; state = stream_readable_state(js, stream_obj); if (is_object_type(state) && js_truthy(js, js_get(js, state, "errored"))) return js_true; state = stream_writable_state(js, stream_obj); if (is_object_type(state) && js_truthy(js, js_get(js, state, "errored"))) return js_true; return js_false;}
static ant_value_t js_stream_is_readable(ant_params_t) { ant_value_t stream_obj = nargs > 0 ? args[0] : js_mkundef(); ant_value_t state = 0;
if (!is_object_type(stream_obj)) return js_false; if (js_truthy(js, js_get(js, stream_obj, "destroyed"))) return js_false; if (js_truthy(js, js_get(js, stream_obj, "readableEnded"))) return js_false; state = stream_readable_state(js, stream_obj); if (is_object_type(state)) { if (js_truthy(js, js_get(js, state, "ended"))) return js_false; if (js_truthy(js, js_get(js, state, "errored"))) return js_false; } state = stream_writable_state(js, stream_obj); if (is_object_type(state) && js_truthy(js, js_get(js, state, "errored"))) return js_false; return js_bool(js_truthy(js, js_get(js, stream_obj, "readable")));}
static void stream_web_copy_global(ant_t *js, ant_value_t obj, const char *name) { ant_value_t value = js_get(js, js->global, name); if (is_err(value)) return; js_set(js, obj, name, value);}
// TODO: remove copy-on-startstatic void stream_web_define_common(ant_t *js, ant_value_t obj) { stream_web_copy_global(js, obj, "ReadableStream"); stream_web_copy_global(js, obj, "ReadableStreamDefaultReader"); stream_web_copy_global(js, obj, "ReadableStreamDefaultController"); stream_web_copy_global(js, obj, "WritableStream"); stream_web_copy_global(js, obj, "WritableStreamDefaultWriter"); stream_web_copy_global(js, obj, "WritableStreamDefaultController"); stream_web_copy_global(js, obj, "TransformStream"); stream_web_copy_global(js, obj, "TransformStreamDefaultController"); stream_web_copy_global(js, obj, "ByteLengthQueuingStrategy"); stream_web_copy_global(js, obj, "CountQueuingStrategy"); stream_web_copy_global(js, obj, "TextEncoderStream"); stream_web_copy_global(js, obj, "TextDecoderStream"); stream_web_copy_global(js, obj, "CompressionStream"); stream_web_copy_global(js, obj, "DecompressionStream");}
ant_value_t stream_library(ant_t *js) { ant_value_t lib = Ant_Object_CreateBare(js); ant_value_t promises = Ant_Object_CreateBare(js); stream_init_constructors(js);
js_set(js, promises, "pipeline", Ant_Function_Create(js_stream_promises_pipeline)); js_set(js, promises, "finished", Ant_Function_Create(js_stream_promises_finished));
js_set_module_default(js, lib, js->builtins.stream_ctor, "Stream"); js_set(js, lib, "Readable", js->builtins.readable_ctor); js_set(js, lib, "Writable", js->builtins.writable_ctor); js_set(js, lib, "Duplex", js->builtins.duplex_ctor); js_set(js, lib, "Transform", js->builtins.stream_transform_ctor); js_set(js, lib, "PassThrough", js->builtins.passthrough_ctor); js_set(js, lib, "pipeline", Ant_Function_Create(js_stream_pipeline)); js_set(js, lib, "finished", Ant_Function_CreateWithArity(js_stream_finished, 3)); js_set(js, lib, "getDefaultHighWaterMark", Ant_Function_CreateWithArity(js_stream_get_default_high_water_mark, 1)); js_set(js, lib, "setDefaultHighWaterMark", Ant_Function_CreateWithArity(js_stream_set_default_high_water_mark, 2)); js_set(js, lib, "isDestroyed", Ant_Function_CreateWithArity(js_stream_is_destroyed, 1)); js_set(js, lib, "isDisturbed", Ant_Function_CreateWithArity(js_stream_is_disturbed, 1)); js_set(js, lib, "isErrored", Ant_Function_CreateWithArity(js_stream_is_errored, 1)); js_set(js, lib, "isReadable", Ant_Function_CreateWithArity(js_stream_is_readable, 1)); js_set(js, lib, "promises", promises);
js_set(js, js->builtins.stream_ctor, "Readable", js->builtins.readable_ctor); js_set(js, js->builtins.stream_ctor, "Writable", js->builtins.writable_ctor); js_set(js, js->builtins.stream_ctor, "Duplex", js->builtins.duplex_ctor); js_set(js, js->builtins.stream_ctor, "Transform", js->builtins.stream_transform_ctor); js_set(js, js->builtins.stream_ctor, "PassThrough", js->builtins.passthrough_ctor); js_set(js, js->builtins.stream_ctor, "pipeline", js_get(js, lib, "pipeline")); js_set(js, js->builtins.stream_ctor, "finished", js_get(js, lib, "finished")); js_set(js, js->builtins.stream_ctor, "getDefaultHighWaterMark", js_get(js, lib, "getDefaultHighWaterMark")); js_set(js, js->builtins.stream_ctor, "setDefaultHighWaterMark", js_get(js, lib, "setDefaultHighWaterMark")); js_set(js, js->builtins.stream_ctor, "isDestroyed", js_get(js, lib, "isDestroyed")); js_set(js, js->builtins.stream_ctor, "isDisturbed", js_get(js, lib, "isDisturbed")); js_set(js, js->builtins.stream_ctor, "isErrored", js_get(js, lib, "isErrored")); js_set(js, js->builtins.stream_ctor, "isReadable", js_get(js, lib, "isReadable")); js_set(js, js->builtins.stream_ctor, "promises", promises);
js_set(js, promises, "default", promises); js_set_slot_wb(js, promises, SLOT_DEFAULT, promises); js_set_tostringtag(js, lib, js_mkstr(js, "stream", 6)); return lib;}
ant_value_t stream_promises_library(ant_t *js) { ant_value_t stream_ns = js_esm_import_sync_cstr(js, "stream", 6); ant_value_t promises = 0; if (is_err(stream_ns)) return stream_ns; promises = js_get(js, stream_ns, "promises"); if (!is_object_type(promises)) return js_mkerr(js, "stream.promises is not available"); js_set(js, promises, "default", promises); js_set_slot_wb(js, promises, SLOT_DEFAULT, promises); return promises;}
ant_value_t stream_web_library(ant_t *js) { ant_value_t lib = Ant_Object_CreateBare(js);
stream_web_define_common(js, lib); js_set(js, lib, "default", lib); js_set_slot_wb(js, lib, SLOT_DEFAULT, lib); js_set_tostringtag(js, lib, js_mkstr(js, "stream/web", 10));
return lib;}