Open History to see the last commit
1/**
2 * Browsers paint a multipart/x-mixed-replace part only once the next part begins, and the hub
3 * writes a part only when the screen changes. Left alone, an `<img>` viewer never shows a still
4 * screen (clients such as the iOS app then give up waiting for the first frame) and always hides
5 * the last frame of a change. Re-sending the last part after a short quiet period makes it paint.
6 */
7import * as Arr from "effect/Array";
8import * as Duration from "effect/Duration";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as Option from "effect/Option";
12import * as Stream from "effect/Stream";
13
14const STILL_FRAME_DELAY = Duration.millis(100);
15
16const HEADER_END = Buffer.from("\r\n\r\n");
17const MAX_HEADER_BYTES = 16 * 1024;
18// Simulator frames are far smaller; a larger claim is unknown framing, not something to retain.
19const MAX_PART_BYTES = 16 * 1024 * 1024;
20
21type Part = Arr.NonEmptyReadonlyArray<Uint8Array>;
22
23const partLength = (headers: string) => {
24 if (!headers.startsWith("--")) return undefined;
25 const length = Number(/^content-length:\s*(\d+)\s*$/im.exec(headers)?.[1]);
26 return length <= MAX_PART_BYTES ? length : undefined;
27};
28
29const endsWithCrlf = (part: Part) => {
30 const last = Arr.lastNonEmpty(part);
31 const cr = last.length > 1 ? last.at(-2) : part.at(-2)?.at(-1);
32 return cr === 0x0d && last.at(-1) === 0x0a;
33};
34
35/**
36 * Follows the hub's part framing (`--frame`, `Content-Length`, CRLF CRLF, body, CRLF; each part
37 * is one write) and keeps the last complete part as views into the forwarded chunks, so frame
38 * bytes are never copied. Any other framing turns tracking off and the rest passes through.
39 */
40const partTracker = () => {
41 let pieces: Array<Uint8Array> = [];
42 let size = 0;
43 let partEnd: number | null = null;
44 let tracking = true;
45 return {
46 /** The last complete part, while the stream sits at a part boundary. */
47 still: null as Part | null,
48 feed(chunk: Uint8Array) {
49 if (!tracking) return;
50 let rest = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength);
51 let part: Part | undefined;
52 while (rest.length > 0) {
53 if (partEnd === null) {
54 // A header split across chunks is re-scanned per chunk. Only a header-sized
55 // prefix of the new chunk joins the scan, so frame bytes are never copied.
56 const head =
57 size === 0
58 ? rest
59 : Buffer.concat([...pieces, rest.subarray(0, MAX_HEADER_BYTES + HEADER_END.length)]);
60 const headerEnd = head.indexOf(HEADER_END);
61 if (headerEnd < 0 && head.length <= MAX_HEADER_BYTES) {
62 pieces.push(rest);
63 size += rest.length;
64 break;
65 }
66 const length =
67 headerEnd < 0 || headerEnd > MAX_HEADER_BYTES
68 ? undefined
69 : partLength(head.subarray(0, headerEnd).toString("latin1"));
70 if (length === undefined) {
71 tracking = false;
72 break;
73 }
74 partEnd = headerEnd + HEADER_END.length + length + 2;
75 }
76 const taken = rest.subarray(0, partEnd - size);
77 pieces.push(taken);
78 size += taken.length;
79 rest = rest.subarray(taken.length);
80 if (size < partEnd) break;
81 const complete = pieces;
82 if (!Arr.isArrayNonEmpty(complete) || !endsWithCrlf(complete)) {
83 tracking = false;
84 break;
85 }
86 part = complete;
87 pieces = [];
88 size = 0;
89 partEnd = null;
90 }
91 if (!tracking) pieces = [];
92 this.still = tracking && size === 0 ? (part ?? this.still) : null;
93 },
94 };
95};
96
97/** Re-sends the latest frame once when the stream goes quiet between frames. */
98export const repeatStillFrame = <E, R>(body: Stream.Stream<Uint8Array, E, R>) =>
99 Stream.transformPull(body, (pull, scope) =>
100 Effect.sync(() => {
101 const tracker = partTracker();
102 let pending: Fiber.Fiber<Effect.Success<typeof pull>, Effect.Error<typeof pull>> | null =
103 null;
104 const forward = (chunks: Effect.Success<typeof pull>) => {
105 pending = null;
106 for (const chunk of chunks) tracker.feed(chunk);
107 return chunks;
108 };
109 return Effect.gen(function* () {
110 const still = tracker.still;
111 if (still === null && pending === null) return forward(yield* pull);
112 pending ??= yield* Effect.forkIn(pull, scope);
113 if (still === null) return forward(yield* Fiber.join(pending));
114 // Timing out the join leaves the pull running, so a frame arriving mid-wait is kept.
115 const next = yield* Fiber.join(pending).pipe(Effect.timeoutOption(STILL_FRAME_DELAY));
116 if (Option.isSome(next)) return forward(next.value);
117 tracker.still = null;
118 return still;
119 });
120 }),
121 );