Open History to see the last commit
1import * as Arr from "effect/Array";
2import * as Effect from "effect/Effect";
3import * as Socket from "effect/socket/Socket";
4
5// Old clients omit this; old servers ignore it.
6export const SOCKET_FRAME_PARTS_QUERY_PARAM = "frameParts";
7
8// WebSocket messages arrive whole. At 4,000 bytes/s each part takes about
9// 12 seconds, leaving margin before the client's 15-second liveness timeout.
10const SOCKET_FRAME_PART_BYTES = 48 * 1024;
11
12// No JSON message starts with a record separator. It marks every part but the last.
13const PART_PREFIX = "\u001e";
14
15// The RPC client only counts frames that decode to messages as liveness, and
16// ignores pongs, so each held-back part reads to it as one.
17const PART_RECEIVED_FRAME = '{"_tag":"Pong"}';
18
19/** Splits on the UTF-8 size the frame has on the wire, never inside a surrogate pair. */
20export function splitSocketFrame(
21 frame: string,
22 partBytes = SOCKET_FRAME_PART_BYTES,
23): Arr.NonEmptyReadonlyArray<string> {
24 // UTF-8 spends at most 3 bytes on a UTF-16 code unit.
25 if (frame.length * 3 <= partBytes) return [frame];
26 const parts: Array<string> = [];
27 let start = 0;
28 let bytes = PART_PREFIX.length;
29 for (let index = 0; index < frame.length;) {
30 const code = frame.charCodeAt(index);
31 const pair = code >= 0xd800 && code <= 0xdbff && index + 1 < frame.length;
32 const size = pair ? 4 : code < 0x80 ? 1 : code < 0x800 ? 2 : 3;
33 if (bytes + size > partBytes) {
34 parts.push(PART_PREFIX + frame.slice(start, index));
35 start = index;
36 bytes = PART_PREFIX.length;
37 }
38 bytes += size;
39 index += pair ? 2 : 1;
40 }
41 return Arr.append(parts, frame.slice(start));
42}
43
44/** Server side: sends string frames longer than one part as consecutive parts. */
45export function splitLargeSocketFrames(socket: Socket.Socket): Socket.Socket {
46 const split = (chunk: string | Uint8Array): Arr.NonEmptyReadonlyArray<string | Uint8Array> =>
47 typeof chunk === "string" ? splitSocketFrame(chunk) : [chunk];
48 return Socket.make({
49 reader: socket.reader,
50 writer: Effect.map(socket.writer, (writer) => ({
51 write: (chunk) =>
52 typeof chunk === "string" ? writer.writeAll(splitSocketFrame(chunk)) : writer.write(chunk),
53 writeAll: (chunks) => writer.writeAll(Arr.flatMap(chunks, split)),
54 })),
55 });
56}
57
58/** Client side: joins parts back into whole frames. */
59export function joinSocketFrameParts(socket: Socket.Socket): Socket.Socket {
60 return Socket.make({
61 reader: Effect.map(socket.reader, (reader) => {
62 let pending = "";
63 const join = (frame: string | Uint8Array) => {
64 if (typeof frame !== "string") return frame;
65 if (frame.startsWith(PART_PREFIX)) {
66 pending += frame.slice(PART_PREFIX.length);
67 return PART_RECEIVED_FRAME;
68 }
69 if (pending === "") return frame;
70 const whole = pending + frame;
71 pending = "";
72 return whole;
73 };
74 return {
75 upgrade: reader.upgrade,
76 pull: Effect.map(reader.pull, Arr.map(join)),
77 };
78 }),
79 writer: socket.writer,
80 });
81}