1import { assert, describe, it } from "@effect/vitest"
2import * as Effect from "effect/Effect"
3import * as Exit from "effect/Exit"
4import * as Fiber from "effect/Fiber"
5import * as Latch from "effect/Latch"
6import * as Socket from "effect/socket/Socket"
7import * as Stream from "effect/Stream"
8
9type EventType = "open" | "message" | "error" | "close"
10
11class TestWebSocket implements Socket.WebSocketLike {
12 readonly listeners = new Map<
13 EventType,
14 Set<{ readonly listener: (event: Socket.WebSocketEvent) => void; readonly once: boolean }>
15 >()
16 readyState = 0
17 pauseCount = 0
18 resumeCount = 0
19
20 constructor(readonly openListenerAttached: Latch.Latch) {}
21
22 pause(): void {
23 this.pauseCount++
24 }
25
26 resume(): void {
27 this.resumeCount++
28 }
29
30 addEventListener(
31 type: EventType,
32 listener: (event: Socket.WebSocketEvent) => void,
33 options?: { readonly once?: boolean }
34 ): void {
35 const listeners = this.listeners.get(type) ?? new Set()
36 listeners.add({ listener, once: options?.once === true })
37 this.listeners.set(type, listeners)
38 if (type === "open") this.openListenerAttached.openUnsafe()
39 }
40
41 removeEventListener(type: EventType, listener: (event: Socket.WebSocketEvent) => void): void {
42 const listeners = this.listeners.get(type)
43 if (listeners === undefined) return
44 for (const entry of listeners) {
45 if (entry.listener === listener) listeners.delete(entry)
46 }
47 }
48
49 dispatch(type: EventType, event: Socket.WebSocketEvent): void {
50 const listeners = this.listeners.get(type)
51 if (listeners === undefined) return
52 for (const entry of Array.from(listeners)) {
53 if (entry.once) listeners.delete(entry)
54 entry.listener(event)
55 }
56 }
57
58 listenerCount(type: EventType): number {
59 return this.listeners.get(type)?.size ?? 0
60 }
61
62 close(): void {}
63 send(): void {}
64}
65
66const stubSocket = (pull: Socket.Reader["pull"]) => {
67 let upgraded = false
68 const socket = Socket.make({
69 reader: Effect.succeed({
70 pull,
71 upgrade: () =>
72 Effect.sync(() => {
73 upgraded = true
74 })
75 }),
76 writer: Effect.succeed({
77 write: () => Effect.void,
78 writeAll: () => Effect.void
79 })
80 })
81 return { socket, wasUpgraded: () => upgraded }
82}
83
84describe("Socket", () => {
85 describe("make", () => {
86 it.effect("exposes TLS upgrade on the acquired reader", () =>
87 Effect.gen(function*() {
88 const { socket, wasUpgraded } = stubSocket(Effect.never)
89
90 const { upgrade } = yield* socket.reader
91 yield* upgrade()
92
93 assert.isTrue(wasUpgraded())
94 }))
95
96 it.effect("readerBytes maps pulls to bytes", () =>
97 Effect.gen(function*() {
98 const { socket } = stubSocket(Effect.succeed(["hello"]))
99
100 const pull = yield* Socket.readerBytes(socket)
101 const [message] = yield* pull
102
103 assert.deepStrictEqual(message, new TextEncoder().encode("hello"))
104 }))
105
106 it.effect("readerString maps pulls to strings", () =>
107 Effect.gen(function*() {
108 const { socket } = stubSocket(Effect.succeed([new TextEncoder().encode("hello")]))
109
110 const pull = yield* Socket.readerString(socket)
111 const [message] = yield* pull
112
113 assert.strictEqual(message, "hello")
114 }))
115 })
116
117 describe("fromWebSocket", () => {
118 it.effect("removes the open listener when the socket closes while connecting", () =>
119 Effect.gen(function*() {
120 const openListenerAttached = Latch.makeUnsafe(false)
121 const ws = new TestWebSocket(openListenerAttached)
122 const socket = yield* Socket.fromWebSocket(Effect.succeed(ws))
123 const reader = yield* socket.reader.pipe(
124 Effect.exit,
125 Effect.forkChild({ startImmediately: true })
126 )
127
128 yield* openListenerAttached.await
129 ws.dispatch("close", { code: 1006, reason: "closed while connecting" })
130
131 assert.isTrue(Exit.isFailure(yield* Fiber.join(reader)))
132 assert.strictEqual(ws.listenerCount("open"), 0)
133 }))
134
135 it.effect("pauses at the highWaterMark and resumes after draining", () =>
136 Effect.gen(function*() {
137 const ws = new TestWebSocket(Latch.makeUnsafe(false))
138 ws.readyState = 1
139 const socket = yield* Socket.fromWebSocket(Effect.succeed(ws), { highWaterMark: 10 })
140 const { pull } = yield* socket.reader
141 const firstPull = yield* pull.pipe(Effect.forkChild({ startImmediately: true }))
142
143 ws.dispatch("message", { data: "first" })
144 assert.deepStrictEqual(yield* Fiber.join(firstPull), ["first"])
145
146 ws.dispatch("message", { data: "hello" })
147 assert.strictEqual(ws.pauseCount, 0)
148 ws.dispatch("message", { data: "world" })
149 assert.strictEqual(ws.pauseCount, 1)
150
151 assert.deepStrictEqual(yield* pull, ["hello", "world"])
152 assert.strictEqual(ws.resumeCount, 1)
153 }))
154
155 it.effect("counts text frames toward the highWaterMark by UTF-8 byte length", () =>
156 Effect.gen(function*() {
157 const ws = new TestWebSocket(Latch.makeUnsafe(false))
158 ws.readyState = 1
159 const socket = yield* Socket.fromWebSocket(Effect.succeed(ws), { highWaterMark: 3 })
160 const { pull } = yield* socket.reader
161
162 ws.dispatch("message", { data: "€" })
163
164 assert.strictEqual(ws.pauseCount, 1)
165 assert.deepStrictEqual(yield* pull, ["€"])
166 }))
167
168 it.effect("uses the default highWaterMark for pausable sockets", () =>
169 Effect.gen(function*() {
170 const ws = new TestWebSocket(Latch.makeUnsafe(false))
171 ws.readyState = 1
172 const socket = yield* Socket.fromWebSocket(Effect.succeed(ws))
173 const { pull } = yield* socket.reader
174 const payload = "a".repeat(64 * 1024)
175
176 ws.dispatch("message", { data: payload })
177 assert.strictEqual(ws.pauseCount, 1)
178 assert.deepStrictEqual(yield* pull, [payload])
179 assert.strictEqual(ws.resumeCount, 1)
180 }))
181 })
182
183 describe("toChannel", () => {
184 it.effect("reports a write failure while a pull is suspended", () =>
185 Effect.gen(function*() {
186 const writeError = new Socket.SocketError({
187 reason: new Socket.SocketWriteError({ cause: new Error("write failed") })
188 })
189 // an idle reader that only fails once its acquisition scope closes,
190 // which is the whole reason `toChannel` needs no per-pull race
191 const socket = Socket.make({
192 reader: Effect.gen(function*() {
193 const closed = Latch.makeUnsafe(false)
194 yield* Effect.addFinalizer(() => closed.open)
195 return {
196 pull: Effect.andThen(
197 closed.await,
198 Effect.fail(new Socket.SocketError({ reason: new Socket.SocketCloseError({ code: 1000 }) }))
199 ),
200 upgrade: Socket.SocketUpgradeError.unsupported
201 }
202 }),
203 writer: Effect.succeed({
204 write: () => Effect.fail(writeError),
205 writeAll: () => Effect.fail(writeError)
206 })
207 })
208
209 const exit = yield* Stream.make("hello").pipe(
210 Stream.encodeText,
211 Stream.pipeThroughChannel(Socket.toChannel(socket)),
212 Stream.runDrain,
213 Effect.exit
214 )
215
216 assert.deepStrictEqual(exit, Exit.fail(writeError))
217 }))
218 })
219})