io-events.js (4292B)
1 function bytes(value, limit) { 2 return Array.isArray(value) && value.length <= limit && value.every(x => Number.isInteger(x) && x >= 0 && x <= 255); 3 } 4 export class Events { 5 run; 6 limit; 7 next = 1; 8 pending = new Map(); 9 queue = []; 10 constructor(run, limit = Number.MAX_SAFE_INTEGER) { 11 this.run = run; 12 this.limit = limit; 13 if (!/^[1-9][0-9]*$/.test(run) || run.length > 20 || BigInt(run) > 0xffffffffffffffffn || 14 !Number.isSafeInteger(limit) || limit < 1) 15 throw new Error('invalid event identity'); 16 } 17 get counts() { return { pending: this.pending.size, queued: this.queue.length }; } 18 read(limit, signal) { 19 if (!Number.isInteger(limit) || limit < 1 || limit > 4096) 20 throw new Error('invalid read bound'); 21 return this.start({ run: this.run, id: 0, kind: 'read', limit }, signal); 22 } 23 write(stream, value, signal) { 24 if (!['output', 'error'].includes(stream) || !bytes(value, 4096) || !value.length) 25 throw new Error('invalid write chunk'); 26 return this.start({ run: this.run, id: 0, kind: 'write', stream, bytes: [...value] }, signal); 27 } 28 start(event, signal) { 29 if (signal.aborted) 30 return Promise.resolve({ kind: 'cancelled' }); 31 if (this.next > this.limit) 32 throw new Error('event identity exhausted'); 33 event.id = this.next++; 34 return new Promise(resolve => { 35 const cancel = () => { 36 const entry = this.pending.get(event.id); 37 if (!entry || entry.cancelling) 38 return; 39 entry.cancelling = true; 40 if (!entry.delivered) 41 this.finish(event.id, { kind: 'cancelled' }); 42 else 43 this.queue.push({ run: this.run, id: event.id, kind: 'cancel', operation: event.kind }); 44 }; 45 this.pending.set(event.id, { event, delivered: false, cancelling: false, resolve, 46 remove: () => signal.removeEventListener('abort', cancel) }); 47 signal.addEventListener('abort', cancel, { once: true }); 48 this.queue.push(event); 49 }); 50 } 51 take() { 52 const out = this.queue; 53 this.queue = []; 54 for (const event of out) 55 this.pending.get(event.id).delivered = true; 56 // Never expose the mutable validation record or its byte array to the peer. 57 return out.map(event => event.kind === 'write' ? { ...event, bytes: [...event.bytes] } : { ...event }); 58 } 59 finish(id, value) { 60 const entry = this.pending.get(id); 61 this.pending.delete(id); 62 this.queue = this.queue.filter(event => event.id !== id); 63 entry.remove(); 64 entry.resolve(entry.cancelling ? { kind: 'cancelled' } : value); 65 } 66 // Wrong run/kind/bounds and stale/duplicate replies never consume a live wait. 67 // A valid normal completion can race with cancellation: it acknowledges work, 68 // but its data is discarded. No extra cancellation acknowledgement is needed. 69 reply(value) { 70 if (!value || typeof value !== 'object') 71 return false; 72 const v = value; 73 if (v.run !== this.run || typeof v.id !== 'number' || !Number.isSafeInteger(v.id)) 74 return false; 75 const entry = this.pending.get(v.id); 76 if (!entry || !entry.delivered) 77 return false; 78 let completion; 79 if (v.kind === 'cancelled' && entry.cancelling) 80 completion = { kind: 'cancelled' }; 81 else if (v.kind === 'error') 82 completion = { kind: 'error' }; 83 else if (v.kind === 'read' && entry.event.kind === 'read' && bytes(v.bytes, entry.event.limit) && 84 typeof v.eof === 'boolean' && (v.eof ? v.bytes.length === 0 : v.bytes.length > 0)) 85 completion = { kind: 'read', bytes: [...v.bytes], eof: v.eof }; 86 else if (v.kind === 'write' && entry.event.kind === 'write' && typeof v.count === 'number' && 87 Number.isInteger(v.count) && v.count > 0 && v.count <= entry.event.bytes.length) 88 completion = { kind: 'write', count: v.count }; 89 else 90 return false; 91 this.finish(v.id, completion); 92 return true; 93 } 94 }