char/avril
(wip) why'd you have to go and make things so complicaaaaated?
git clone https://git.t4t.associates/char/avril
38d7f6c
main
1import { 2ArrayIsArray , 3getter , 4NumberIsFinite , 5ObjectCreate , 6ObjectDefineProperty as define , 7ObjectFreeze as freeze , 8RangeError , 9TypedArrayPrototypeGetBuffer as bufferOf , 10TypedArrayPrototypeGetByteLength as length , 11TypedArrayPrototypeGetByteOffset as offsetOf , 12TypedArrayPrototypeGetTag as tag , 13TypeError , 14Uint8Array , 15Uint8ArrayPrototypeSet as setBytes , 16uncurryThis , 17WeakMap , 18WeakMapPrototypeSet as set , 19} from "ext:core/primordials" ; 20import { 21promiseResolve , 22promiseThen as then , 23safePromise , 24} from "ext:core/promises" ; 25import { 26op_net_accept , 27op_net_close , 28op_net_connect_start , 29op_net_connected , 30op_net_datagram_broadcast , 31op_net_datagram_listen , 32op_net_datagram_receive , 33op_net_datagram_send , 34op_net_listen , 35op_net_read , 36op_net_shutdown , 37op_net_write , 38op_net_writev , 39} from "ext:core/ops" ; 40import type { 41NetAddress , 42NetConnection , 43NetListener , 44NetListenerId , 45NetUdpAddress , 46} from "ext:core/ops" ; 47import { array , dictionary , state , usv } from "ext:web/idl.js" ; 48import { addListener , removeListener , signals } from "ext:web/events.js" ; 49import { registerUncloneable } from "ext:web/serialization.js" ; 50import { 51byobRequest , 52enqueue , 53errorStream , 54finish , 55ReadableStream , 56ReadableStreamBYOBRequest , 57WritableStream , 58} from "ext:webstreams/streams.js" ; 59import type { 60ReadableByteStreamController , 61WritableStreamDefaultController , 62} from "ext:webstreams/streams.js" ; 63import { failure , result } from "ext:file/errors.js" ; 64 65interface SocketData { 66id :number | null ; 67owner :object ; 68reader ?:ReadableByteStreamController | null ; 69writer ?:WritableStreamDefaultController | null ; 70} 71interface ConnectionData < A extends NetAddress = NetAddress > extends SocketData { 72writing :Promise < void >; 73local :A ; 74remote :A ; 75readable :ReadableStream < Uint8Array < ArrayBuffer >> | null ; 76writable :WritableStream < Uint8Array < ArrayBufferLike >> | null ; 77reader :ReadableByteStreamController | null ; 78writer :WritableStreamDefaultController | null ; 79} 80interface ListenerData < A extends NetAddress = NetAddress > extends SocketData { 81id :NetListenerId < A > | null ; 82local :A ; 83} 84interface DatagramData extends SocketData { 85writing :Promise < void >; 86local :Readonly < NetUdpAddress >; 87} 88type Datagram = [ 89bytes :Uint8Array < ArrayBufferLike >, 90remote :Readonly < NetUdpAddress >, 91]; 92const connections = new WeakMap < object , ConnectionData >(), 93listeners = new WeakMap < object , ListenerData >(), 94datagrams = new WeakMap < object , DatagramData >(); 95registerUncloneable ( connections ); 96registerUncloneable ( listeners ); 97registerUncloneable ( datagrams ); 98const FinalizationRegistry = globalThis . FinalizationRegistry ; 99const register = uncurryThis ( 100( FinalizationRegistry . prototype as FinalizationRegistry < number >). register , 101); 102const unregister = uncurryThis ( FinalizationRegistry . prototype . unregister ); 103const registry = new FinalizationRegistry (( id :number ) => op_net_close ( id )); 104const getView = getter ( ReadableStreamBYOBRequest . prototype , "view" ); 105const respond = uncurryThis ( ReadableStreamBYOBRequest . prototype . respond ); 106const writableError = uncurryThis ( 107WritableStreamDefaultController . prototype . error , 108); 109const dispose :typeof Symbol . dispose = Symbol . dispose , 110asyncIterator :typeof Symbol . asyncIterator = Symbol . asyncIterator , 111empty = new Uint8Array (); 112const safeInteger = Number . isSafeInteger ; 113 114function args < T extends unknown [] > ( ...items :T ) :T { 115const value = array (); 116for ( let i = 0 ; i < items . length ; i ++ ) value [ i ] = items [ i ]; 117return value as T ; 118} 119function address < A extends NetAddress | NetUdpAddress > ( value :A ) :A { 120const copy :( NetAddress | NetUdpAddress ) & { __proto__ ?:null } = 121value . transport === "unix" 122 ?{ __proto__ :null , transport :"unix" , path :value . path } 123 :{ 124__proto__ :null , 125transport :value . transport , 126hostname :value . hostname , 127port :value . port , 128}; 129return freeze ( copy ) as A ; 130} 131function options ( 132value :unknown , 133connecting = false , 134) :NetAddress & { __proto__ ?:null } { 135const options = dictionary ( value ); 136const transport = options . transport , 137host = options . hostname , 138inputPort = options . port , 139path = options . path ; 140if ( transport === "unix" ) { 141if ( host !== undefined || inputPort !== undefined ) { 142throw new TypeError ( "Unix sockets cannot specify hostname or port" ); 143} 144if ( path === undefined ) throw new TypeError ( "Unix sockets require a path" ); 145const label = usv ( path ); 146if ( ! label ) throw new TypeError ( "Unix sockets require a path" ); 147return { __proto__ :null , transport :"unix" , path :label }; 148} 149if ( path !== undefined ) { 150throw new TypeError ( "Socket paths require Unix transport" ); 151} 152const port = + ( inputPort as number ); 153if ( transport !== undefined && transport !== "tcp" ) { 154throw new TypeError ( "Expected TCP transport" ); 155} 156if ( 157! NumberIsFinite ( port ) || port < ( connecting ?1 :0 ) || port > 65535 || 158port % 1 !== 0 159) throw new RangeError ( "Invalid socket port" ); 160return { 161__proto__ :null , 162transport :"tcp" as const , 163hostname :host === undefined ?"127.0.0.1" :usv ( host ), 164 port, 165}; 166} 167function id < I extends number > ( data :{ id :I | null }) :I { 168if ( data . id === null ) throw failure ( "BadResource" , "Socket is closed" ); 169return data . id ; 170} 171function bytes ( value :unknown ) :Uint8Array < ArrayBufferLike > { 172if ( tag ( value ) !== "Uint8Array" ) { 173throw new TypeError ( "Socket I/O requires a Uint8Array" ); 174} 175setBytes ( value as Uint8Array < ArrayBufferLike >, empty ); 176return value as Uint8Array < ArrayBufferLike >; 177} 178function segment < B extends ArrayBufferLike > ( 179value :Uint8Array < B >, 180offset = 0 , 181) :Uint8Array < B > { 182const count = length ( value ) - offset ; 183return new Uint8Array ( 184bufferOf ( value ), 185offsetOf ( value ) + offset , 186count < 65536 ?count :65536 , 187); 188} 189function vectors ( buffers :unknown ) :Uint8Array < ArrayBuffer >[] { 190if ( ! ArrayIsArray ( buffers )) { 191throw new TypeError ( "Socket I/O requires an array of Uint8Array" ); 192} 193// Snapshot before queueing, without rereading caller-controlled elements. 194const count = buffers . length , pieces :Uint8Array < ArrayBuffer >[] = args (); 195if ( ! safeInteger ( count ) || count < 0 || count > 4294967295 ) { 196throw new RangeError ( "Invalid vector length" ); 197} 198let total = 0 ; 199for ( let i = 0 ; i < count ; i ++ ) { 200const view :unknown = buffers [ i ]; 201if ( tag ( view ) !== "Uint8Array" ) { 202throw new TypeError ( "Socket I/O requires an array of Uint8Array" ); 203} 204setBytes ( view as Uint8Array < ArrayBufferLike >, empty ); 205const size = length ( view ); 206if ( size === 0 ) continue ; 207if ( ! safeInteger ( total + size )) { 208throw new RangeError ( "Vector exceeds safe integer range" ); 209} 210total += size ; 211// Fixed snapshot for the native call, which needs unshared storage. 212pieces [ pieces . length ] = new Uint8Array ( 213new Uint8Array ( 214bufferOf ( view ), 215offsetOf ( view ), 216size , 217), 218); 219} 220return pieces ; 221} 222function close ( 223data :SocketData , 224quiet = false , 225source :"read" | "write" | null = null , 226) :void { 227if ( data . id === null ) { 228if ( ! quiet ) id ( data ); 229return ; 230} 231const current = data . id ; 232data . id = null ; 233unregister ( registry , data . owner ); 234const error = failure ( "BadResource" , "Socket is closed" ); 235if ( data . reader && source !== "read" ) errorStream ( data . reader , error ); 236if ( data . writer && source !== "write" ) writableError ( data . writer , error ); 237const reply = op_net_close ( current ); 238if ( ! quiet ) result ( reply ); 239} 240function queue < T > ( 241data :{ writing :Promise < void > }, 242operation :() => T | PromiseLike < T >, 243) :Promise < T > { 244const promise = then ( data . writing , operation , operation ); 245data . writing = then ( promise , () => {}, () => {}); 246return promise ; 247} 248const socketKey = Symbol (); 249type TcpAddr = Extract < NetAddress , { transport :"tcp" }>; 250type UnixAddr = Extract < NetAddress , { transport :"unix" }>; 251type TcpOptions = { 252transport ?:"tcp" ; 253hostname ?:string ; 254port :number ; 255signal ?:AbortSignal ; 256}; 257type UnixOptions = { 258transport :"unix" ; 259path :string ; 260hostname ?:never ; 261port ?:never ; 262signal ?:AbortSignal ; 263}; 264 265class Conn < A extends NetAddress = NetAddress > { 266declare static readonly [ Symbol . hasInstance ] :( value :unknown ) => value isConn ; 267 #data:ConnectionData < A >; 268 269constructor ( ...args :[ typeof socketKey , NetConnection < A >]) { 270const key = args [ 0 ], row = args [ 1 ]; 271if ( key !== socketKey ) throw new TypeError ( "Illegal constructor" ); 272this . #data= { 273id :row [ 0 ], 274owner :this , 275local :address ( row [ 1 ]), 276remote :address ( row [ 2 ]), 277readable :null , 278writable :null , 279reader :null , 280writer :null , 281writing :promiseResolve ( undefined ), 282}; 283set ( connections , this , this . #data); 284register ( registry , this , row [ 0 ], this ); 285} 286get localAddr () :A { 287state ( connections , this ); 288return this . #data. local ; 289} 290get remoteAddr () :A { 291state ( connections , this ); 292return this . #data. remote ; 293} 294async read ( buffer :Uint8Array < ArrayBufferLike >) :Promise < number | null > { 295const data = state ( connections , this ), current = id ( data ); 296bytes ( buffer ); 297const chunk = result ( 298await safePromise ( op_net_read ( args ( current , length ( buffer )))), 299); 300setBytes ( buffer , empty ); 301if ( chunk === null ) return null ; 302setBytes ( buffer , chunk ); 303return length ( chunk ); 304} 305write ( buffer :Uint8Array < ArrayBufferLike >) :Promise < number > { 306const data = state ( connections , this ); 307id ( data ); 308const snapshot = new Uint8Array ( segment ( bytes ( buffer ))); 309return queue ( 310data , 311() => then ( op_net_write ( args ( id ( data ), snapshot )), result ), 312); 313} 314writev ( buffers :unknown ) :Promise < number > { 315const data = state ( connections , this ); 316id ( data ); 317const snapshots = vectors ( buffers ); 318return queue ( data , async () => { 319if ( snapshots . length === 0 ) { 320return result ( 321await safePromise ( op_net_writev ( args ( id ( data ), snapshots ))), 322); 323} 324let total = 0 , elem = 0 , inner = 0 ; 325while ( elem < snapshots . length ) { 326const batch :Uint8Array < ArrayBuffer >[] = args (); 327let budget = 65536 , scan = elem , scanInner = inner ; 328while ( scan < snapshots . length && batch . length < 16 && budget > 0 ) { 329const piece = snapshots [ scan ], size = length ( piece ) - scanInner ; 330if ( size === 0 ) { 331scan ++ ; 332scanInner = 0 ; 333continue ; 334} 335const take = size < budget ?size :budget ; 336batch [ batch . length ] = new Uint8Array ( 337bufferOf ( piece ), 338offsetOf ( piece ) + scanInner , 339take , 340); 341budget -= take ; 342scanInner += take ; 343if ( scanInner >= length ( piece )) { 344scan ++ ; 345scanInner = 0 ; 346} 347} 348const count = result ( 349await safePromise ( op_net_writev ( args ( id ( data ), batch ))), 350); 351if ( ! safeInteger ( count ) || count <= 0 || count > 65536 - budget ) { 352throw failure ( "IoError" , "Invalid partial socket write" ); 353} 354total += count ; 355let skip = count ; 356while ( skip > 0 ) { 357const remaining = length ( snapshots [ elem ]) - inner ; 358if ( remaining === 0 ) { 359elem ++ ; 360inner = 0 ; 361continue ; 362} 363if ( skip < remaining ) { 364inner += skip ; 365break ; 366} 367skip -= remaining ; 368elem ++ ; 369inner = 0 ; 370} 371} 372return total ; 373}); 374} 375closeWrite () :Promise < void > { 376const data = state ( connections , this ); 377id ( data ); 378return queue ( data , () => 379then ( op_net_shutdown ( id ( data )), ( reply ) => { 380result ( reply ); 381})); 382} 383close () { 384close ( state ( connections , this )); 385} 386[ dispose ]() { 387close ( state ( connections , this ), true ); 388} 389get readable () { 390const data = state ( connections , this ); 391if ( data . readable ) return data . readable ; 392id ( data ); 393data . readable = new ReadableStream ({ 394type :"bytes" , 395start ( controller ) { 396data . reader = controller ; 397}, 398async pull ( controller ) { 399try { 400const request = byobRequest ( controller ), 401view = request && getView ( request ); 402const chunk = result ( 403await safePromise ( 404op_net_read ( args ( id ( data ), view ?length ( view ) :65536 )), 405), 406); 407if ( data . id === null ) return ; 408if ( chunk === null ) finish ( controller ); 409else if ( request ) { 410setBytes ( view ! , chunk ); 411respond ( request , length ( chunk )); 412} else enqueue ( controller , chunk ); 413} catch ( error ) { 414errorStream ( controller , error ); 415close ( data , true , "read" ); 416} 417}, 418cancel () { 419close ( data , true , "read" ); 420}, 421}, { highWaterMark :0 }); 422return data . readable ; 423} 424get writable () { 425const data = state ( connections , this ); 426if ( data . writable ) return data . writable ; 427id ( data ); 428data . writable = new WritableStream ({ 429start ( controller ) { 430data . writer = controller ; 431}, 432async write ( value ) { 433try { 434const snapshot = new Uint8Array ( bytes ( value )); 435await safePromise ( queue ( data , async () => { 436let offset = 0 ; 437while ( offset < length ( snapshot )) { 438const count = result ( 439await safePromise ( 440op_net_write ( 441args ( 442id ( data ), 443segment ( snapshot , offset ), 444), 445), 446), 447); 448if ( count <= 0 || count > length ( snapshot ) - offset ) { 449throw failure ( "IoError" , "Invalid partial socket write" ); 450} 451offset += count ; 452} 453})); 454} catch ( error ) { 455close ( data , true , "write" ); 456throw error ; 457} 458}, 459close () { 460return queue ( data , () => then ( op_net_shutdown ( id ( data )), result )); 461}, 462abort () { 463close ( data , true , "write" ); 464}, 465}, { 466highWaterMark :65536 , 467size ( value ) { 468return tag ( value ) === "Uint8Array" ?length ( value ) :1 ; 469}, 470}); 471return data . writable ; 472} 473} 474function connection < A extends NetAddress > ( row :NetConnection < A >) :Conn < A > { 475return new Conn ( socketKey , row ); 476} 477class Listener < A extends NetAddress = NetAddress > { 478declare static readonly [ Symbol . hasInstance ] :( value :unknown ) => value isListener ; 479 #data:ListenerData < A >; 480 481constructor ( ...args :[ typeof socketKey , NetListener < A >]) { 482const key = args [ 0 ], row = args [ 1 ]; 483if ( key !== socketKey ) throw new TypeError ( "Illegal constructor" ); 484this . #data= { id :row [ 0 ], owner :this , local :address ( row [ 1 ]) }; 485set ( listeners , this , this . #data); 486register ( registry , this , row [ 0 ], this ); 487} 488get addr () :A { 489state ( listeners , this ); 490return this . #data. local ; 491} 492async accept () :Promise < Conn < A >> { 493state ( listeners , this ); 494return connection ( 495result ( await safePromise ( op_net_accept ( id ( this . #data)))), 496); 497} 498close () { 499close ( state ( listeners , this )); 500} 501[ dispose ]() { 502close ( state ( listeners , this ), true ); 503} 504[ asyncIterator ]() :AsyncIterableIterator < Conn < A >, undefined > { 505state ( listeners , this ); 506const data = this . #data; 507return { 508async next () { 509if ( data . id === null ) return { done :true , value :undefined }; 510try { 511return { 512done :false , 513value :connection ( 514result ( await safePromise ( op_net_accept ( id ( data )))), 515), 516}; 517} catch ( error ) { 518if ( data . id === null ) return { done :true , value :undefined }; 519throw error ; 520} 521}, 522return () { 523close ( data , true ); 524return promiseResolve ({ done :true , value :undefined }); 525}, 526[ asyncIterator ]() { 527return this ; 528}, 529}; 530} 531} 532function listen ( value :TcpOptions ) :Listener < TcpAddr >; 533function listen ( value :UnixOptions ) :Listener < UnixAddr >; 534function listen ( value :unknown ) :Listener { 535const row = result ( op_net_listen ( options ( value ))); 536return new Listener ( socketKey , row ); 537} 538function connect ( value :TcpOptions ) :Promise < Conn < TcpAddr >>; 539function connect ( value :UnixOptions ) :Promise < Conn < UnixAddr >>; 540async function connect ( value :unknown ) :Promise < Conn > { 541const settings = dictionary ( value ); 542const target = options ( settings , true ), signal = settings . signal ; 543if ( signal !== undefined && state ( signals , signal ). aborted ) { 544throw state ( signals , signal ). reason ; 545} 546const current = result ( op_net_connect_start ( target )); 547const cancel = () => op_net_close ( current ); 548if ( signal !== undefined ) { 549addListener ( signal , "abort" , cancel , { once :true }); 550} 551try { 552const row = result ( await safePromise ( op_net_connected ( current ))); 553if ( signal !== undefined && state ( signals , signal ). aborted ) { 554cancel (); 555throw state ( signals , signal ). reason ; 556} 557return connection ( row ); 558} catch ( error ) { 559cancel (); 560if ( signal !== undefined && state ( signals , signal ). aborted ) { 561throw state ( signals , signal ). reason ; 562} 563throw error ; 564} finally { 565if ( signal !== undefined ) removeListener ( signal , "abort" , cancel ); 566} 567} 568function datagramOptions ( 569value :unknown , 570sending = false , 571) :NetUdpAddress & { __proto__ ?:null } { 572const options = dictionary ( value ); 573const transport = options . transport , 574inputHost = options . hostname , 575inputPort = options . port , 576path = options . path ; 577if ( transport !== undefined && transport !== "udp" ) { 578throw new TypeError ( "Expected UDP transport" ); 579} 580if ( path !== undefined ) { 581throw new TypeError ( "UDP sockets cannot specify a path" ); 582} 583const port = + ( inputPort as number ); 584if ( 585! NumberIsFinite ( port ) || port < ( sending ?1 :0 ) || port > 65535 || 586port % 1 !== 0 587) throw new RangeError ( "Invalid UDP port" ); 588return { 589__proto__ :null , 590transport :"udp" as const , 591hostname :inputHost === undefined ?"127.0.0.1" :usv ( inputHost ), 592 port, 593}; 594} 595async function receiveDatagram ( 596data :DatagramData , 597buffer :Uint8Array < ArrayBufferLike > | undefined , 598) :Promise < Datagram > { 599const current = id ( data ); 600if ( buffer !== undefined ) bytes ( buffer ); 601const row = result ( 602await safePromise ( 603op_net_datagram_receive ( 604args ( current , buffer === undefined ?65535 :length ( buffer )), 605), 606), 607); 608let message :Uint8Array < ArrayBufferLike > = row [ 0 ]; 609if ( buffer !== undefined ) { 610setBytes ( buffer , empty ); 611setBytes ( buffer , message ); 612message = new Uint8Array ( 613bufferOf ( buffer ), 614offsetOf ( buffer ), 615length ( message ), 616); 617} 618return [ message , address ( row [ 1 ])]; 619} 620class DatagramConn { 621declare static readonly [ Symbol . hasInstance ] :( value :unknown ) => value isDatagramConn ; 622private constructor () { 623throw new TypeError ( "Illegal constructor" ); 624} 625get addr () { 626return state ( datagrams , this ). local ; 627} 628receive ( 629buffer :Uint8Array < ArrayBufferLike > | undefined = undefined , 630) :Promise < Datagram > { 631return receiveDatagram ( state ( datagrams , this ), buffer ); 632} 633send ( buffer :Uint8Array < ArrayBufferLike >, target :unknown ) :Promise < number > { 634const data = state ( datagrams , this ); 635id ( data ); 636bytes ( buffer ); 637if ( length ( buffer ) > 65507 ) { 638throw new RangeError ( "Datagram exceeds 65507 bytes" ); 639} 640const snapshot :Uint8Array < ArrayBuffer > = new Uint8Array ( buffer ), 641destination = datagramOptions ( target , true ); 642return queue ( 643data , 644() => 645then ( 646op_net_datagram_send ( args ( id ( data ), snapshot , destination )), 647result , 648), 649); 650} 651setBroadcast ( enabled = true ) :void { 652result ( 653op_net_datagram_broadcast ( args ( id ( state ( datagrams , this )), !! enabled )), 654); 655} 656close () { 657close ( state ( datagrams , this )); 658} 659[ dispose ]() { 660close ( state ( datagrams , this ), true ); 661} 662[ asyncIterator ]() :AsyncIterableIterator < Datagram , undefined > { 663const data = state ( datagrams , this ); 664return { 665async next () { 666if ( data . id === null ) return { done :true , value :undefined }; 667try { 668return { 669done :false , 670value :await safePromise ( receiveDatagram ( data , undefined )), 671}; 672} catch ( error ) { 673if ( data . id === null ) return { done :true , value :undefined }; 674throw error ; 675} 676}, 677return () { 678close ( data , true ); 679return promiseResolve ({ done :true , value :undefined }); 680}, 681[ asyncIterator ]() { 682return this ; 683}, 684}; 685} 686} 687function listenDatagram ( value :unknown ) :DatagramConn { 688const row = result ( op_net_datagram_listen ( datagramOptions ( value ))); 689const socket = ObjectCreate ( DatagramConn . prototype ) as DatagramConn ; 690set ( datagrams , socket , { 691id :row [ 0 ], 692owner :socket , 693local :address ( row [ 1 ]), 694writing :promiseResolve ( undefined ), 695}); 696register ( registry , socket , row [ 0 ], socket ); 697return socket ; 698} 699for ( 700const pair of [[ Conn , "Conn" ], [ Listener , "Listener" ], [ 701DatagramConn , 702"DatagramConn" , 703]] as const 704) { 705define ( pair [ 0 ]. prototype , Symbol . toStringTag , { 706value :pair [ 1 ], 707configurable :true , 708}); 709} 710export { Conn , connect , DatagramConn , listen , listenDatagram , Listener };