Open History to see the last commit
1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Socket from "effect/socket/Socket";
4import { requireOptionalNativeModule } from "expo";
5
6type EventType = "open" | "message" | "error" | "close";
7
8interface NativeEvents {
9 readonly open: { readonly id: number };
10 readonly message: { readonly id: number; readonly text?: string; readonly data?: Uint8Array };
11 readonly error: { readonly id: number; readonly message: string };
12 readonly close: {
13 readonly id: number;
14 readonly code: number;
15 readonly reason: string;
16 readonly wasClean: boolean;
17 };
18}
19
20export interface NativeModule {
21 connect(url: string, protocols: ReadonlyArray<string>): number;
22 sendText(id: number, text: string): void;
23 sendBinary(id: number, bytes: Uint8Array): void;
24 close(id: number, code: number, reason: string): void;
25 addListener<K extends keyof NativeEvents>(
26 event: K,
27 listener: (event: NativeEvents[K]) => void,
28 ): { remove(): void };
29}
30
31interface Listener {
32 readonly listener: (event: Socket.WebSocketEvent) => void;
33 readonly once: boolean;
34}
35
36/** Subset of the WHATWG WebSocket API that Effect's socket layer touches, over the native module. */
37export class Connection implements Socket.WebSocketLike {
38 static readonly CONNECTING = 0;
39 static readonly OPEN = 1;
40 static readonly CLOSING = 2;
41 static readonly CLOSED = 3;
42
43 readonly url: string;
44 readyState: number = Connection.CONNECTING;
45
46 readonly #id: number;
47 readonly #sockets: Sockets;
48 readonly #listeners = new Map<EventType, Array<Listener>>();
49
50 constructor(sockets: Sockets, url: string, protocols?: string | ReadonlyArray<string>) {
51 this.url = url;
52 this.#sockets = sockets;
53 this.#id = sockets.native.connect(
54 url,
55 protocols === undefined ? [] : typeof protocols === "string" ? [protocols] : protocols,
56 );
57 sockets.register(this.#id, this);
58 }
59
60 addEventListener(
61 type: EventType,
62 listener: (event: Socket.WebSocketEvent) => void,
63 options?: { readonly once?: boolean },
64 ): void {
65 const entries = this.#listeners.get(type) ?? [];
66 if (entries.some((entry) => entry.listener === listener)) return;
67 entries.push({ listener, once: options?.once === true });
68 this.#listeners.set(type, entries);
69 }
70
71 removeEventListener(type: EventType, listener: (event: Socket.WebSocketEvent) => void): void {
72 const entries = this.#listeners.get(type);
73 if (entries === undefined) return;
74 this.#listeners.set(
75 type,
76 entries.filter((entry) => entry.listener !== listener),
77 );
78 }
79
80 send(data: string | Uint8Array<ArrayBuffer>): void {
81 if (this.readyState === Connection.CONNECTING) {
82 throw new Error("InvalidStateError: WebSocket is still connecting");
83 }
84 // Browsers drop writes after close instead of throwing.
85 if (this.readyState !== Connection.OPEN) return;
86 if (typeof data === "string") {
87 this.#sockets.native.sendText(this.#id, data);
88 } else {
89 this.#sockets.native.sendBinary(this.#id, data);
90 }
91 }
92
93 close(code = 1000, reason = ""): void {
94 if (this.readyState >= Connection.CLOSING) return;
95 this.readyState = Connection.CLOSING;
96 this.#sockets.native.close(this.#id, code, reason);
97 }
98
99 handleOpen(_event: NativeEvents["open"]): void {
100 if (this.readyState !== Connection.CONNECTING) return;
101 this.readyState = Connection.OPEN;
102 this.#dispatch("open", { type: "open" });
103 }
104
105 handleMessage(event: NativeEvents["message"]): void {
106 if (this.readyState !== Connection.OPEN) return;
107 this.#dispatch("message", { type: "message", data: event.text ?? event.data });
108 }
109
110 handleError(event: NativeEvents["error"]): void {
111 if (this.readyState === Connection.CLOSED) return;
112 this.#dispatch("error", { type: "error", message: event.message });
113 }
114
115 handleClose(event: NativeEvents["close"]): void {
116 if (this.readyState === Connection.CLOSED) return;
117 this.readyState = Connection.CLOSED;
118 this.#sockets.unregister(this.#id);
119 this.#dispatch("close", {
120 type: "close",
121 code: event.code,
122 reason: event.reason,
123 wasClean: event.wasClean,
124 });
125 }
126
127 #dispatch(type: EventType, event: Socket.WebSocketEvent & Record<string, unknown>): void {
128 const entries = this.#listeners.get(type);
129 if (entries === undefined || entries.length === 0) return;
130 this.#listeners.set(
131 type,
132 entries.filter((entry) => !entry.once),
133 );
134 for (const entry of entries) entry.listener(event);
135 }
136}
137
138/** Routes the module's per-id events to the sockets that own them. Native assigns the ids, so a reload cannot reuse one. */
139export class Sockets {
140 readonly native: NativeModule;
141 readonly #open = new Map<number, Connection>();
142 readonly #subscriptions: ReadonlyArray<{ remove(): void }>;
143
144 constructor(native: NativeModule) {
145 this.native = native;
146 this.#subscriptions = [
147 native.addListener("open", (event) => this.#open.get(event.id)?.handleOpen(event)),
148 native.addListener("message", (event) => this.#open.get(event.id)?.handleMessage(event)),
149 native.addListener("error", (event) => this.#open.get(event.id)?.handleError(event)),
150 native.addListener("close", (event) => this.#open.get(event.id)?.handleClose(event)),
151 ];
152 }
153
154 register(id: number, socket: Connection): void {
155 this.#open.set(id, socket);
156 }
157
158 unregister(id: number): void {
159 this.#open.delete(id);
160 }
161
162 dispose(): void {
163 for (const subscription of this.#subscriptions) subscription.remove();
164 for (const socket of this.#open.values()) {
165 socket.close(1001);
166 socket.handleClose({ id: 0, code: 1001, reason: "", wasClean: false });
167 }
168 }
169}
170
171function layerFromModule(native: NativeModule): Layer.Layer<Socket.WebSocketConstructor> {
172 return Layer.effect(
173 Socket.WebSocketConstructor,
174 Effect.map(
175 Effect.acquireRelease(
176 Effect.sync(() => new Sockets(native)),
177 (sockets) => Effect.sync(() => sockets.dispose()),
178 ),
179 (sockets) => (url, options) => {
180 if (options !== undefined && typeof options !== "string" && !Array.isArray(options)) {
181 throw new TypeError("WebSocket client options are not supported by the native WebSocket");
182 }
183 return new Connection(sockets, url, options);
184 },
185 ),
186 );
187}
188
189const native = requireOptionalNativeModule<NativeModule>("T3WebSocket");
190
191// A dev client built before this module shipped still has SocketRocket.
192export const layer: Layer.Layer<Socket.WebSocketConstructor> =
193 native === null ? Socket.layerWebSocketConstructorGlobal : layerFromModule(native);