char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somport webstreams extension to typescripte9b1b43

main
branchesmainall refs →
8.6 KiB292 linesraw
1const portPost = MessagePort.prototype.postMessage;
2const portClose = MessagePort.prototype.close;
3const readerClosed = descriptor(ReadableDefaultReader.prototype, "closed")!
4  .get as () => Promise<void>;
5
6function send(port: Port, value: RequestPacket | ReplyPacket): unknown {
7  return apply(portPost, port, [value]);
8}
9function disconnect(port: Port): void {
10  apply(portClose, port, []);
11}
12function prepareWireStream(_state: unknown, source: object): StreamRecord {
13  const record: StreamRecord = { source, readable: null, writable: null };
14  if (descriptor(source, "_readableStreamController")) {
15    record.kind = "readable";
16    record.readable = source as ReadableResource;
17  } else if (descriptor(source, "_writableStreamController")) {
18    record.kind = "writable";
19    record.writable = source as WritableResource;
20  } else if (descriptor(source, "_transformStreamController")) {
21    record.kind = "transform";
22    record.readable = ownValue(source, "_readable") as ReadableResource;
23    record.writable = ownValue(source, "_writable") as WritableResource;
24  } else throw dataCloneError("Invalid transferable stream");
25  validateRecord(record);
26  if (record.readable) record.readPair = op_port_pair();
27  if (record.writable) record.writePair = op_port_pair();
28  return record;
29}
30function serveStream(
31  port: Port,
32  resource: Reader | Writer,
33  readable: boolean,
34): void {
35  let finished = false, disconnected = false, active = 0;
36  function finish() {
37    finished = true;
38    if (active === 0 && !disconnected) {
39      disconnected = true;
40      disconnect(port);
41    }
42  }
43  const closed = invoke(readable ? readerClosed : writerClosed, resource);
44  then(closed, undefined, (error) => {
45    if (!disconnected) {
46      send(port, [null, false, cloneError(error)]);
47      disconnected = true;
48      disconnect(port);
49    }
50  });
51  setPortCloseHandler(port, () => {
52    if (!disconnected) {
53      disconnected = true;
54      const reason = dataCloneError("Transferred stream endpoint closed");
55      markPromiseHandled(
56        invoke(readable ? cancelReader : abortWriter, resource, [reason]),
57      );
58      disconnect(port);
59    }
60  });
61  addListener(port, "message", (event) => {
62    if (finished || disconnected) return;
63    const packet = messageData(event) as RequestPacket;
64    const id = packet[0], command = packet[1], value = packet[2];
65    active++;
66    let operation;
67    try {
68      if (readable) {
69        operation = command === "read"
70          ? invoke(read, resource)
71          : invoke(cancelReader, resource, [value]);
72      } else {
73        operation = command === "write"
74          ? invoke(write, resource, [value])
75          : command === "close"
76          ? invoke(closeWriter, resource)
77          : invoke(abortWriter, resource, [value]);
78      }
79    } catch (error) {
80      operation = safePromiseReject(error);
81    }
82    then(safePromiseResolve(operation), (result) => {
83      active--;
84      if (disconnected) return;
85      try {
86        send(port, [id, true, result]);
87      } catch (error) {
88        markPromiseHandled(
89          invoke(readable ? cancelReader : abortWriter, resource, [error]),
90        );
91        send(port, [id, false, cloneError(error)]);
92        finish();
93        return;
94      }
95      if (
96        finished || command === "cancel" || command === "close" ||
97        command === "abort" ||
98        (readable && (result as ReadResult | undefined)?.done)
99      ) finish();
100    }, (error) => {
101      active--;
102      if (!disconnected) {
103        send(port, [id, false, cloneError(error)]);
104        finish();
105      }
106    });
107  });
108  startPort(port);
109}
110function commitWireStream(
111  _state: unknown,
112  _source: object,
113  record: StreamRecord,
114): void {
115  if (record.readable) {
116    record.reader = acquireReader(record.readable);
117    op_port_move(record.readPair![1]);
118    serveStream(adoptPort(record.readPair![0]), record.reader, true);
119  }
120  if (record.writable) {
121    record.writer = acquireWriter(record.writable);
122    op_port_move(record.writePair![1]);
123    serveStream(adoptPort(record.writePair![0]), record.writer, false);
124  }
125}
126function streamClient(id: number) {
127  const port = adoptPort(id),
128    pending = new Map<
129      number,
130      { resolve(value: unknown): void; reject(reason: unknown): void }
131    >();
132  let next = 0, ended = false, onerror: ((reason: unknown) => void) | undefined;
133  function fail(error: unknown): void {
134    if (ended) return;
135    ended = true;
136    mapForEach(pending, (entry) => entry.reject(error));
137    mapClear(pending);
138    onerror?.(error);
139    disconnect(port);
140  }
141  setPortCloseHandler(
142    port,
143    () => fail(dataCloneError("Transferred stream endpoint closed")),
144  );
145  addListener(port, "message", (event) => {
146    const packet = messageData(event) as ReplyPacket;
147    const id = packet[0], ok = packet[1], value = packet[2];
148    if (id === null) {
149      fail(value);
150      return;
151    }
152    const entry = mapGet(pending, id);
153    if (!entry) return;
154    mapDelete(pending, id);
155    if (ok) entry.resolve(value);
156    else entry.reject(value);
157  });
158  startPort(port);
159  return {
160    set onerror(callback: (reason: unknown) => void) {
161      onerror = callback;
162    },
163    request<C extends RequestPacket[1]>(
164      command: C,
165      value?: unknown,
166    ): Promise<C extends "read" ? ReadResult : void> {
167      if (ended) {
168        return safePromiseReject(
169          dataCloneError("Transferred stream endpoint closed"),
170        );
171      }
172      const id = next++;
173      return safeNewPromise<C extends "read" ? ReadResult : void>(
174        (resolve, reject) => {
175          mapSet(pending, id, {
176            resolve: resolve as (value: unknown) => void,
177            reject,
178          });
179          try {
180            send(port, [id, command, value]);
181            const retain = () =>
182              then(referencePort(port), () => {
183                if (!ended && mapGet(pending, id)) retain();
184              });
185            retain();
186          } catch (error) {
187            fail(error);
188          }
189        },
190      );
191    },
192    close() {
193      ended = true;
194      disconnect(port);
195    },
196  };
197}
198function decodeWireStream(packet: StreamPacket): StreamResource {
199  const kind = packet[0],
200    readId = packet[1],
201    writeId = packet[2],
202    writableClosed = packet[3];
203  let readable, writable;
204  if (readId !== null) {
205    const client = streamClient(readId);
206    readable = new Readable({
207      start(controller) {
208        client.onerror = (error) => controller.error(error);
209      },
210      pull(controller) {
211        return then(client.request("read"), (result) => {
212          if (result.done) {
213            controller.close();
214            client.close();
215          } else controller.enqueue(result.value);
216        });
217      },
218      cancel(reason) {
219        return then(
220          client.request("cancel", reason),
221          () => client.close(),
222          (error) => {
223            client.close();
224            throw error;
225          },
226        );
227      },
228    }, { highWaterMark: 0 });
229  }
230  if (writeId !== null) {
231    const client = streamClient(writeId);
232    writable = new Writable(
233      writableClosed ? {} : {
234        start(controller) {
235          client.onerror = (error) => controller.error(error);
236        },
237        write(chunk) {
238          return client.request("write", chunk);
239        },
240        close() {
241          return then(client.request("close"), () => client.close());
242        },
243        abort(reason) {
244          return then(client.request("abort", reason), () => client.close());
245        },
246      },
247    );
248    if (writableClosed) {
249      markPromiseHandled(invoke(closeStream, writable));
250      client.close();
251    }
252  }
253  if (kind === "readable") return readable!;
254  if (kind === "writable") return writable!;
255  const transform = new Transform();
256  define(transform, "_readable", { value: readable });
257  define(transform, "_writable", { value: writable });
258  return transform;
259}
260registerWire<undefined, StreamPacket, StreamRecord>({
261  name: "Stream",
262  accepts: (value) =>
263    !!(
264      descriptor(value, "_readableStreamController") ||
265      descriptor(value, "_writableStreamController") ||
266      descriptor(value, "_transformStreamController")
267    ),
268  encode: (
269    _state,
270    _source,
271    record,
272  ) => [
273    record.kind!,
274    record.readPair?.[1] ?? null,
275    record.writePair?.[1] ?? null,
276    record.writableState === "closed",
277  ],
278  decode: decodeWireStream,
279  prepare: prepareWireStream,
280  validate: (_state, _source, record) => validateRecord(record),
281  commit: commitWireStream,
282  rollback: (_state, _source, record) => {
283    const pairs = [record.readPair, record.writePair];
284    for (let i = 0; i < pairs.length; i++) {
285      const pair = pairs[i];
286      if (pair) {
287        op_port_close(pair[0]);
288        op_port_close(pair[1]);
289      }
290    }
291  },
292});