1import { NodeSocket, NodeSocketServer } from "@effect/platform-node"
2import { assert, describe, it } from "@effect/vitest"
3import { Deferred, Effect, Option, Queue, Redacted } from "effect"
4import * as Exit from "effect/Exit"
5import * as Fiber from "effect/Fiber"
6import type * as NetAddress from "effect/net/NetAddress"
7import * as Scope from "effect/Scope"
8import { Socket } from "effect/socket"
9import * as Stream from "effect/Stream"
10import * as Fs from "node:fs"
11import * as Net from "node:net"
12import { Duplex, PassThrough } from "node:stream"
13import * as Tls from "node:tls"
14import { fileURLToPath } from "node:url"
15import { vi } from "vitest"
16import { WS } from "vitest-websocket-mock"
17
18vi.mock("node:net", async (importOriginal) => {
19 const original = await importOriginal<typeof Net>()
20 return { ...original, createConnection: vi.fn(original.createConnection) }
21})
22
23vi.mock("node:tls", async (importOriginal) => {
24 const original = await importOriginal<typeof Tls>()
25 return { ...original, connect: vi.fn(original.connect) }
26})
27
28// Regenerate with:
29// openssl req -x509 -newkey ed25519 -nodes -days 36500 -subj /CN=localhost \
30// -addext subjectAltName=DNS:localhost,IP:127.0.0.1 -keyout key.pem -out cert.pem
31const cert = Fs.readFileSync(fileURLToPath(new URL("./fixtures/tls/cert.pem", import.meta.url)))
32const key = Fs.readFileSync(fileURLToPath(new URL("./fixtures/tls/key.pem", import.meta.url)))
33
34const echoHandler = (socket: Socket.Socket, upgradeOptions?: Socket.TlsUpgradeOptions) =>
35 Effect.gen(function*() {
36 const writer = yield* socket.writer
37 const { pull, upgrade } = yield* socket.reader
38 if (upgradeOptions !== undefined) {
39 yield* upgrade(upgradeOptions)
40 }
41 while (true) {
42 yield* writer.writeAll(yield* pull)
43 }
44 }).pipe(
45 Effect.scoped,
46 Effect.catchTag("SocketError", () => Effect.void)
47 )
48
49const makePassThrough = Effect.acquireRelease(
50 Effect.sync(() => new PassThrough()),
51 (duplex) => Effect.sync(() => duplex.destroy())
52)
53
54// A duplex-backed reader with one pull parked, waiting for bytes.
55const parkedPull = Effect.gen(function*() {
56 const duplex = yield* makePassThrough
57 const { pull } = yield* (yield* NodeSocket.fromDuplex(Effect.succeed(duplex))).reader
58 const parked = yield* Effect.forkChild(pull, { startImmediately: true })
59 return { duplex, pull, parked }
60})
61
62const decode = (batch: ReadonlyArray<Uint8Array | string>) =>
63 Buffer.concat(batch.map((chunk) => Buffer.from(chunk))).toString()
64
65const sendHelloWorld = (socket: Socket.Socket, upgradeOptions?: Socket.TlsUpgradeOptions) =>
66 Effect.gen(function*() {
67 const writer = yield* socket.writer
68 const { pull, upgrade } = yield* socket.reader
69 if (upgradeOptions !== undefined) {
70 yield* upgrade(upgradeOptions)
71 }
72 yield* writer.writeAll(["Hello", "World"])
73 const decoder = new TextDecoder()
74 let text = ""
75 while (text.length < 10) {
76 for (const chunk of yield* pull) {
77 text += typeof chunk === "string" ? chunk : decoder.decode(chunk)
78 }
79 }
80 return text
81 }).pipe(Effect.scoped)
82
83const makeServer = Effect.gen(function*() {
84 const server = yield* NodeSocketServer.make({ port: 0 })
85 yield* server.run(echoHandler).pipe(Effect.forkScoped)
86 return server
87})
88
89describe("Socket", () => {
90 it.live("preserves a native IPv6 listener address", () =>
91 Effect.gen(function*() {
92 const server = yield* NodeSocketServer.make({ host: "::1", port: 0 })
93 assert.strictEqual(server.address._tag, "InetAddressV6")
94 if (server.address._tag !== "InetAddressV6") return
95 assert.strictEqual(server.address.address.toString(), "::1")
96 assert.notStrictEqual(server.address.port, 0)
97 }))
98
99 it.live("closes with a pending pre-run socket", () =>
100 Effect.gen(function*() {
101 const scope = yield* Scope.make()
102 const server = yield* NodeSocketServer.make({ host: "127.0.0.1", port: 0 }).pipe(Scope.provide(scope))
103 assert.strictEqual(server.address._tag, "InetAddressV4")
104 if (server.address._tag !== "InetAddressV4") return
105 const socket = Net.createConnection({ host: "127.0.0.1", port: server.address.port })
106 yield* Effect.promise(() =>
107 new Promise<void>((resolve, reject) => {
108 socket.once("connect", resolve)
109 socket.once("error", reject)
110 })
111 )
112 const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild)
113 const exit = yield* Fiber.await(closing).pipe(
114 Effect.timeout("1 second"),
115 Effect.ensuring(Effect.sync(() => socket.destroy()))
116 )
117 assert.isTrue(Exit.isSuccess(exit))
118 }))
119
120 it.live("closes with a pending pre-run WebSocket", () =>
121 Effect.gen(function*() {
122 const scope = yield* Scope.make()
123 const server = yield* NodeSocketServer.makeWebSocket({ host: "127.0.0.1", port: 0 }).pipe(Scope.provide(scope))
124 assert.strictEqual(server.address._tag, "InetAddressV4")
125 if (server.address._tag !== "InetAddressV4") return
126 const socket = new NodeSocket.NodeWS.WebSocket(`ws://127.0.0.1:${server.address.port}`)
127 yield* Effect.promise(() =>
128 new Promise<void>((resolve, reject) => {
129 socket.once("open", resolve)
130 socket.once("error", reject)
131 })
132 )
133 const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild)
134 const exit = yield* Fiber.await(closing).pipe(
135 Effect.timeout("1 second"),
136 Effect.ensuring(Effect.sync(() => socket.terminate()))
137 )
138 assert.isTrue(Exit.isSuccess(exit))
139 }))
140
141 it.effect("open", () =>
142 Effect.gen(function*() {
143 const server = yield* makeServer
144 const channel = NodeSocket.makeNetChannel({ port: (server.address as NetAddress.InetAddress).port })
145
146 const outputEffect = Stream.make("Hello", "World").pipe(
147 Stream.encodeText,
148 Stream.pipeThroughChannel(channel),
149 Stream.catchIf(
150 (error) => error.reason._tag === "SocketCloseError",
151 () => Stream.empty
152 ),
153 Stream.decodeText(),
154 Stream.mkString
155 )
156
157 const output = yield* outputEffect
158 assert.strictEqual(output, "HelloWorld")
159 }))
160
161 it.effect("pull batches", () =>
162 Effect.gen(function*() {
163 const server = yield* makeServer
164 const socket = yield* NodeSocket.makeNet({ port: (server.address as NetAddress.InetAddress).port })
165
166 const received = yield* Effect.gen(function*() {
167 const writer = yield* socket.writer
168 const pull = yield* Socket.readerBytes(socket)
169 yield* writer.writeAll(["Hello", "World"])
170 const received: Array<Uint8Array> = []
171 let length = 0
172 while (length < 10) {
173 const chunk = yield* pull
174 for (const data of chunk) {
175 received.push(data)
176 length += data.byteLength
177 }
178 }
179 return received
180 }).pipe(Effect.scoped)
181
182 const text = new TextDecoder().decode(
183 received.reduce((acc, chunk) => {
184 const next = new Uint8Array(acc.byteLength + chunk.byteLength)
185 next.set(acc, 0)
186 next.set(chunk, acc.byteLength)
187 return next
188 }, new Uint8Array(0))
189 )
190 assert.strictEqual(text, "HelloWorld")
191 }))
192
193 it.effect("pull drains every buffered duplex chunk", () =>
194 Effect.gen(function*() {
195 const first = Buffer.from("aa")
196 const second = Buffer.from("bb")
197 const duplex = new Duplex({
198 read() {},
199 write() {}
200 })
201 duplex.pause()
202 const socket = yield* NodeSocket.fromDuplex(Effect.succeed(duplex))
203 const { pull } = yield* socket.reader
204 duplex.push(first)
205 duplex.push(second)
206 const batch = yield* pull
207 assert.strictEqual(Buffer.concat(batch.map((chunk) => chunk as Uint8Array)).toString(), "aabb")
208 // Node 26 returns one buffered chunk per `read()`, so the batch carries
209 // the pushed buffers with no copy. Earlier Node concatenates them.
210 if (batch.length === 2) {
211 assert.strictEqual(batch[0], first)
212 assert.strictEqual(batch[1], second)
213 }
214 }))
215
216 it.live("handles reader scope closure during a parked pull read", () =>
217 Effect.gen(function*() {
218 const duplex = yield* makePassThrough
219 const socket = yield* NodeSocket.fromDuplex(Effect.succeed(duplex))
220 const readScope = yield* Scope.make()
221 yield* Effect.addFinalizer(() => Scope.close(readScope, Exit.void))
222 const { pull } = yield* Scope.provide(socket.reader, readScope)
223 const parked = yield* Effect.forkChild(pull, { startImmediately: true })
224 let observed = ""
225 duplex.once("data", (chunk) => {
226 observed = chunk.toString()
227 Effect.runSync(Scope.close(readScope, Exit.void))
228 })
229
230 // Deliver synchronously so an exception stays in the test, rather than
231 // escaping from Node's next-tick readable notification.
232 duplex.write("hello")
233 duplex.emit("readable")
234 const error = yield* Fiber.join(parked).pipe(Effect.flip, Effect.timeout("1 second"))
235 assert.strictEqual(observed, "hello")
236 assert.strictEqual(error.reason._tag, "SocketCloseError")
237 if (error.reason._tag === "SocketCloseError") {
238 assert.strictEqual(error.reason.code, 1006)
239 }
240 }))
241
242 it.live("preserves a chunk when a parked pull is interrupted during read", () =>
243 Effect.gen(function*() {
244 const { duplex, parked, pull } = yield* parkedPull
245 duplex.once("data", () => Effect.runSync(Fiber.interrupt(parked)))
246 duplex.write("hello")
247 duplex.emit("readable")
248 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(parked).pipe(Effect.timeout("1 second"))))
249 const next = yield* pull.pipe(Effect.timeout("1 second"))
250 assert.strictEqual(decode(next), "hello")
251 }))
252
253 it.live("drains a retained chunk when EOF follows parked pull interruption", () =>
254 Effect.gen(function*() {
255 const { duplex, parked, pull } = yield* parkedPull
256 duplex.once("data", () => Effect.runSync(Fiber.interrupt(parked)))
257 const ended = new Promise<void>((resolve) => duplex.once("end", resolve))
258 duplex.end("hello")
259 duplex.emit("readable")
260 yield* Effect.promise(() => ended).pipe(Effect.timeout("1 second"))
261 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(parked)))
262 const next = yield* pull.pipe(Effect.timeout("1 second"))
263 assert.strictEqual(decode(next), "hello")
264 const error = yield* Effect.flip(pull).pipe(Effect.timeout("1 second"))
265 assert.strictEqual(error.reason._tag, "SocketCloseError")
266 if (error.reason._tag === "SocketCloseError") {
267 assert.strictEqual(error.reason.code, 1000)
268 }
269 }))
270
271 it.live("preserves order when a parked pull is replaced synchronously during read", () =>
272 Effect.gen(function*() {
273 const { duplex, parked, pull } = yield* parkedPull
274 const scope = yield* Effect.scope
275 let replacement: Fiber.Fiber<ReadonlyArray<Uint8Array | string>, Socket.SocketError> | undefined
276 duplex.once("data", () => {
277 Effect.runSync(Fiber.interrupt(parked))
278 // Make the second chunk available only after read() consumes the first,
279 // so this also exercises separate reads on Node versions that coalesce.
280 duplex.write("second")
281 replacement = Effect.runSync(Effect.forkIn(pull, scope, { startImmediately: true }))
282 })
283 duplex.write("first")
284 duplex.emit("readable")
285 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(parked)))
286 assert.isDefined(replacement)
287 let received = decode(yield* Fiber.join(replacement).pipe(Effect.timeout("1 second")))
288 while (received.length < "firstsecond".length) {
289 received += decode(yield* pull.pipe(Effect.timeout("1 second")))
290 }
291 assert.strictEqual(received, "firstsecond")
292 }))
293
294 it.live("respects a zero open timeout", () =>
295 Effect.gen(function*() {
296 const socket = yield* NodeSocket.fromDuplex(Effect.never, { openTimeout: 0 })
297 const error = yield* Effect.scoped(Effect.asVoid(socket.reader)).pipe(
298 Effect.flip,
299 Effect.timeout("1 second")
300 )
301 assert.strictEqual(error.reason._tag, "SocketOpenError")
302 if (error.reason._tag === "SocketOpenError") {
303 assert.strictEqual(error.reason.kind, "Timeout")
304 }
305 }))
306
307 it.live("destroys a pending TCP connection on open timeout", () => {
308 const conn = new Net.Socket()
309 const destroy = vi.spyOn(conn, "destroy")
310 const destroySoon = vi.spyOn(conn, "destroySoon").mockImplementation(() => {})
311 vi.mocked(Net.createConnection).mockReturnValueOnce(conn)
312
313 return Effect.gen(function*() {
314 const socket = yield* NodeSocket.makeNet({ host: "127.0.0.1", port: 1, openTimeout: 0 })
315 yield* Effect.scoped(Effect.asVoid(socket.reader)).pipe(Effect.flip)
316 assert.isTrue(destroy.mock.calls.length > 0)
317 assert.strictEqual(destroySoon.mock.calls.length, 0)
318 }).pipe(
319 Effect.ensuring(Effect.sync(() => {
320 destroy.mockRestore()
321 destroySoon.mockRestore()
322 conn.destroy()
323 }))
324 )
325 })
326
327 it.live("gracefully closes an established TCP connection", () => {
328 const conn = new Net.Socket()
329 const destroy = vi.spyOn(conn, "destroy")
330 const destroySoon = vi.spyOn(conn, "destroySoon").mockImplementation(() => {})
331 vi.mocked(Net.createConnection).mockImplementationOnce(() => {
332 queueMicrotask(() => conn.emit("connect"))
333 return conn
334 })
335
336 return Effect.gen(function*() {
337 const socket = yield* NodeSocket.makeNet({ host: "127.0.0.1", port: 1 })
338 yield* Effect.scoped(Effect.asVoid(socket.reader))
339 assert.strictEqual(destroy.mock.calls.length, 0)
340 assert.isTrue(destroySoon.mock.calls.length > 0)
341 }).pipe(
342 Effect.ensuring(Effect.sync(() => {
343 destroy.mockRestore()
344 destroySoon.mockRestore()
345 conn.destroy()
346 }))
347 )
348 })
349
350 describe("TLS", () => {
351 const makeTlsServer = Effect.suspend(() => {
352 const sockets = new Set<Tls.TLSSocket>()
353 return Effect.acquireRelease(
354 Effect.callback<Tls.Server>((resume) => {
355 const server = Tls.createServer({ cert, key }, (socket) => {
356 sockets.add(socket)
357 socket.on("close", () => sockets.delete(socket))
358 socket.pipe(socket)
359 })
360 server.listen(0, "127.0.0.1", () => resume(Effect.succeed(server)))
361 }),
362 (server) =>
363 Effect.sync(() => {
364 for (const socket of sockets) socket.destroy()
365 server.close()
366 })
367 )
368 })
369
370 const serverPort = (server: Tls.Server) => (server.address() as Net.AddressInfo).port
371
372 it.effect("open", () =>
373 Effect.gen(function*() {
374 const server = yield* makeTlsServer
375 const channel = NodeSocket.makeTlsChannel({
376 host: "127.0.0.1",
377 port: serverPort(server),
378 ca: [cert]
379 })
380
381 const output = yield* Stream.make("Hello", "World").pipe(
382 Stream.encodeText,
383 Stream.pipeThroughChannel(channel),
384 Stream.catchIf(
385 (error) => error.reason._tag === "SocketCloseError",
386 () => Stream.empty
387 ),
388 Stream.decodeText(),
389 Stream.mkString
390 )
391 assert.strictEqual(output, "HelloWorld")
392 }))
393
394 it.effect("echoes over the reader and writer", () =>
395 Effect.gen(function*() {
396 const server = yield* makeTlsServer
397 const socket = yield* NodeSocket.makeTls({
398 host: "127.0.0.1",
399 port: serverPort(server),
400 ca: [cert]
401 })
402
403 const received = yield* sendHelloWorld(socket)
404
405 assert.strictEqual(received, "HelloWorld")
406 }))
407
408 it.effect("fails to open against an untrusted certificate", () =>
409 Effect.gen(function*() {
410 const server = yield* makeTlsServer
411 const socket = yield* NodeSocket.makeTls({
412 host: "127.0.0.1",
413 port: serverPort(server)
414 })
415
416 const error = yield* Effect.scoped(Effect.asVoid(socket.reader)).pipe(Effect.flip)
417 assert.strictEqual(error.reason._tag, "SocketOpenError")
418 if (error.reason._tag === "SocketOpenError") {
419 assert.strictEqual(error.reason.kind, "Unknown")
420 assert.strictEqual((error.reason.cause as { code?: string }).code, "DEPTH_ZERO_SELF_SIGNED_CERT")
421 }
422 }))
423
424 it.live("destroys a pending TLS connection on open timeout", () => {
425 const conn = new Tls.TLSSocket(new Net.Socket())
426 const destroy = vi.spyOn(conn, "destroy")
427 const destroySoon = vi.spyOn(conn, "destroySoon").mockImplementation(() => {})
428 vi.mocked(Tls.connect).mockReturnValueOnce(conn)
429
430 return Effect.gen(function*() {
431 const socket = yield* NodeSocket.makeTls({ host: "127.0.0.1", port: 1, openTimeout: 0 })
432 yield* Effect.scoped(Effect.asVoid(socket.reader)).pipe(Effect.flip)
433 assert.isTrue(destroy.mock.calls.length > 0)
434 assert.strictEqual(destroySoon.mock.calls.length, 0)
435 }).pipe(
436 Effect.ensuring(Effect.sync(() => {
437 destroy.mockRestore()
438 destroySoon.mockRestore()
439 conn.destroy()
440 }))
441 )
442 })
443
444 it.effect("echoes through a NodeSocketServer.makeTls server", () =>
445 Effect.gen(function*() {
446 const server = yield* NodeSocketServer.makeTls({ host: "127.0.0.1", port: 0, cert, key })
447 yield* server.run(echoHandler).pipe(Effect.forkScoped)
448
449 const socket = yield* NodeSocket.fromDuplex(Effect.acquireRelease(
450 Effect.callback<Tls.TLSSocket>((resume) => {
451 const conn = Tls.connect({
452 host: "127.0.0.1",
453 port: (server.address as NetAddress.InetAddress).port,
454 ca: [cert]
455 })
456 conn.once("secureConnect", () => resume(Effect.succeed(conn)))
457 }),
458 (conn) => Effect.sync(() => conn.destroy())
459 ))
460
461 const received = yield* sendHelloWorld(socket)
462
463 assert.strictEqual(received, "HelloWorld")
464 }))
465
466 it.effect("upgrades a plain NodeSocketServer connection to TLS", () =>
467 Effect.gen(function*() {
468 const server = yield* NodeSocketServer.make({ host: "127.0.0.1", port: 0 })
469 yield* server.run((socket) => echoHandler(socket, { cert, key: Redacted.make(key) })).pipe(
470 Effect.forkScoped
471 )
472
473 const socket = yield* NodeSocket.makeTls({
474 host: "127.0.0.1",
475 port: (server.address as NetAddress.InetAddress).port,
476 ca: [cert]
477 })
478
479 const received = yield* sendHelloWorld(socket)
480
481 assert.strictEqual(received, "HelloWorld")
482 }))
483
484 it.effect("upgrades a makeNet client connection to TLS", () =>
485 Effect.gen(function*() {
486 const server = yield* NodeSocketServer.makeTls({ host: "127.0.0.1", port: 0, cert, key })
487 yield* server.run(echoHandler).pipe(Effect.forkScoped)
488
489 const socket = yield* NodeSocket.makeNet({
490 host: "127.0.0.1",
491 port: (server.address as NetAddress.InetAddress).port
492 })
493
494 const received = yield* sendHelloWorld(socket, {
495 ca: [cert],
496 rejectUnauthorized: true
497 })
498
499 assert.strictEqual(received, "HelloWorld")
500 }))
501
502 it.live("requires both key and cert for server TLS upgrades", () =>
503 Effect.gen(function*() {
504 const raw = new Duplex({
505 read() {},
506 write(_chunk, _encoding, callback) {
507 callback()
508 }
509 })
510 const socket = yield* NodeSocket.fromDuplex(Effect.succeed(raw), { tlsServer: true })
511
512 const error = yield* Effect.gen(function*() {
513 const { upgrade } = yield* socket.reader
514 return yield* upgrade().pipe(Effect.flip)
515 }).pipe(Effect.scoped)
516
517 assert.strictEqual(error.reason._tag, "SocketUpgradeError")
518 if (error.reason._tag === "SocketUpgradeError") {
519 assert.strictEqual((error.reason.cause as Error).message, "server TLS upgrade requires both key and cert")
520 }
521 }))
522
523 it.live("rejects incomplete client credentials", () =>
524 Effect.gen(function*() {
525 const raw = new Duplex({
526 read() {},
527 write(_chunk, _encoding, callback) {
528 callback()
529 }
530 })
531 const socket = yield* NodeSocket.fromDuplex(Effect.succeed(raw))
532
533 const error = yield* Effect.gen(function*() {
534 const { upgrade } = yield* socket.reader
535 return yield* upgrade({ cert }).pipe(Effect.flip)
536 }).pipe(Effect.scoped)
537
538 assert.strictEqual(error.reason._tag, "SocketUpgradeError")
539 if (error.reason._tag === "SocketUpgradeError") {
540 assert.strictEqual(
541 (error.reason.cause as Error).message,
542 "TLS upgrade credentials must include both key and cert"
543 )
544 }
545 }))
546
547 it.live("fails pulls after an interrupted TLS upgrade", () =>
548 Effect.gen(function*() {
549 const raw = new Duplex({
550 read() {},
551 write(_chunk, _encoding, callback) {
552 callback()
553 }
554 })
555 const tls = new Duplex({
556 read() {},
557 write(_chunk, _encoding, callback) {
558 callback()
559 }
560 })
561 vi.mocked(Tls.connect).mockReturnValueOnce(tls as Tls.TLSSocket)
562 const socket = yield* NodeSocket.fromDuplex(Effect.succeed(raw))
563
564 const error = yield* Effect.gen(function*() {
565 const { pull, upgrade } = yield* socket.reader
566 const upgradeFiber = yield* upgrade().pipe(Effect.forkChild({ startImmediately: true }))
567 yield* Fiber.interrupt(upgradeFiber)
568 return yield* pull.pipe(Effect.flip, Effect.timeout("1 second"))
569 }).pipe(Effect.scoped)
570
571 assert.strictEqual(error.reason._tag, "SocketCloseError")
572 }))
573
574 it.live("rejects a second TLS upgrade", () =>
575 Effect.gen(function*() {
576 const server = yield* NodeSocketServer.makeTls({ host: "127.0.0.1", port: 0, cert, key })
577 yield* server.run((socket) =>
578 Effect.gen(function*() {
579 yield* Effect.asVoid(socket.reader)
580 return yield* Effect.never
581 }).pipe(Effect.scoped)
582 ).pipe(Effect.forkScoped)
583 const socket = yield* NodeSocket.makeNet({
584 host: "127.0.0.1",
585 port: (server.address as NetAddress.InetAddress).port
586 })
587
588 const error = yield* Effect.gen(function*() {
589 const { upgrade } = yield* socket.reader
590 const options = {
591 cert,
592 key: Redacted.make(key),
593 ca: [cert],
594 rejectUnauthorized: true
595 }
596 yield* upgrade(options)
597 return yield* upgrade(options).pipe(
598 Effect.flip,
599 Effect.timeout("1 second")
600 )
601 }).pipe(Effect.scoped)
602
603 assert.strictEqual(error.reason._tag, "SocketUpgradeError")
604 }))
605
606 it.effect("reports TLS handshake failures as SocketUpgradeError", () =>
607 Effect.gen(function*() {
608 const server = yield* NodeSocketServer.makeTls({ host: "127.0.0.1", port: 0, cert, key })
609 yield* server.run(() => Effect.never).pipe(Effect.forkScoped)
610 const socket = yield* NodeSocket.makeNet({
611 host: "127.0.0.1",
612 port: (server.address as NetAddress.InetAddress).port
613 })
614
615 const error = yield* Effect.gen(function*() {
616 const { upgrade } = yield* socket.reader
617 return yield* upgrade({
618 cert,
619 key: Redacted.make(key),
620 rejectUnauthorized: true
621 }).pipe(Effect.flip)
622 }).pipe(Effect.scoped)
623
624 assert.strictEqual(error.reason._tag, "SocketUpgradeError")
625 if (error.reason._tag === "SocketUpgradeError") {
626 assert.strictEqual((error.reason.cause as { code?: string }).code, "DEPTH_ZERO_SELF_SIGNED_CERT")
627 }
628 }))
629 })
630
631 describe("WebSocket", { concurrent: false }, () => {
632 const url = `ws://localhost:1234`
633
634 const makeServer = Effect.acquireRelease(
635 Effect.sync(() => new WS(url)),
636 (ws) =>
637 Effect.sync(() => {
638 ws.close()
639 WS.clean()
640 })
641 )
642
643 it.live("receives messages after handing a paused WebSocket to the reader", () =>
644 Effect.gen(function*() {
645 const serverScope = yield* Scope.make()
646 const connections = new Set<NodeSocket.NodeWS.WebSocket>()
647 yield* Effect.addFinalizer(() =>
648 Effect.gen(function*() {
649 for (const connection of connections) connection.terminate()
650 yield* Scope.close(serverScope, Exit.void)
651 })
652 )
653 const received = yield* Deferred.make<ReadonlyArray<Uint8Array | string>>()
654 const server = yield* NodeSocketServer.makeWebSocket({ host: "127.0.0.1", port: 0 }).pipe(
655 Scope.provide(serverScope)
656 )
657 assert.strictEqual(server.address._tag, "InetAddressV4")
658 if (server.address._tag !== "InetAddressV4") return
659
660 yield* server.run((socket) =>
661 Effect.gen(function*() {
662 const native = Option.getOrThrow(yield* Effect.serviceOption(Socket.WebSocket))
663 if (native instanceof NodeSocket.NodeWS.WebSocket) connections.add(native)
664 const { pull } = yield* socket.reader
665 const writer = yield* socket.writer
666 yield* writer.write("reader-ready")
667 yield* Deferred.succeed(received, yield* pull)
668 }).pipe(Effect.scoped)
669 ).pipe(Effect.forkIn(serverScope))
670
671 const client = new NodeSocket.NodeWS.WebSocket(`ws://127.0.0.1:${server.address.port}`)
672 connections.add(client)
673 const open = new Promise<void>((resolve, reject) => {
674 client.once("open", resolve)
675 client.once("error", reject)
676 })
677 const ready = new Promise<string>((resolve) => {
678 client.once("message", (data) => resolve(data.toString()))
679 })
680
681 yield* Effect.promise(() => open).pipe(Effect.timeout("1 second"))
682 assert.strictEqual(yield* Effect.promise(() => ready).pipe(Effect.timeout("1 second")), "reader-ready")
683 client.send("after-ready")
684
685 assert.deepStrictEqual(
686 yield* Deferred.await(received).pipe(Effect.timeout("1 second")),
687 ["after-ready"]
688 )
689 }))
690
691 it.effect("passes headers to the opening handshake", () =>
692 Effect.gen(function*() {
693 const authorization = yield* Deferred.make<string | undefined>()
694 const server = yield* NodeSocketServer.makeWebSocket({
695 host: "127.0.0.1",
696 port: 0,
697 verifyClient: ({ req }: Parameters<NodeSocket.NodeWS.VerifyClientCallbackSync>[0]) => {
698 Deferred.doneUnsafe(authorization, Effect.succeed(req.headers.authorization))
699 return true
700 }
701 })
702 assert.strictEqual(server.address._tag, "InetAddressV4")
703 if (server.address._tag !== "InetAddressV4") return
704 const port = server.address.port
705
706 const makeWebSocket = yield* Socket.WebSocketConstructor
707 const client = yield* Effect.acquireRelease(
708 Effect.sync(() =>
709 makeWebSocket(`ws://127.0.0.1:${port}`, {
710 headers: { Authorization: "Bearer test" }
711 })
712 ),
713 (client) => Effect.sync(() => client.close())
714 )
715 client.addEventListener("error", () => {})
716
717 assert.strictEqual(yield* Deferred.await(authorization).pipe(Effect.timeout("1 second")), "Bearer test")
718 }).pipe(Effect.provide(NodeSocket.layerWebSocketConstructorWS)))
719
720 it.effect("fails TLS upgrade with SocketUpgradeError", () =>
721 Effect.gen(function*() {
722 yield* Effect.asVoid(makeServer)
723 const socket = yield* Socket.makeWebSocket(Effect.succeed(url))
724 const { upgrade } = yield* socket.reader
725 const error = yield* upgrade({ cert, key: Redacted.make(key) }).pipe(Effect.flip)
726 assert.strictEqual(error.reason._tag, "SocketUpgradeError")
727 }).pipe(
728 Effect.provideService(Socket.WebSocketConstructor, (url) => new globalThis.WebSocket(url))
729 ))
730
731 it.effect("messages", () =>
732 Effect.gen(function*() {
733 const server = yield* makeServer
734 const socket = yield* Socket.makeWebSocket(Effect.succeed(url))
735 const messages = yield* Queue.unbounded<Uint8Array>()
736 const fiber = yield* Effect.gen(function*() {
737 const pull = yield* Socket.readerBytes(socket)
738 while (true) {
739 yield* Queue.offerAll(messages, yield* pull)
740 }
741 }).pipe(Effect.scoped, Effect.forkChild)
742 yield* Effect.gen(function*() {
743 const writer = yield* socket.writer
744 yield* writer.write(new TextEncoder().encode("Hello"))
745 yield* writer.write(new TextEncoder().encode("World"))
746 }).pipe(Effect.scoped)
747 assert.deepStrictEqual(yield* Effect.promise(() => server.nextMessage), new TextEncoder().encode("Hello"))
748 assert.deepStrictEqual(yield* Effect.promise(() => server.nextMessage), new TextEncoder().encode("World"))
749
750 server.send("Right back at you!")
751 let message = yield* Queue.take(messages)
752 assert.deepStrictEqual(message, new TextEncoder().encode("Right back at you!"))
753
754 server.send(new Blob(["A Blob message"]))
755 message = yield* Queue.take(messages)
756 assert.deepStrictEqual(message, new TextEncoder().encode("A Blob message"))
757
758 server.close()
759 const exit = yield* Fiber.await(fiber)
760 assert.isTrue(Exit.isFailure(exit))
761 if (Exit.isFailure(exit)) {
762 const failure = exit.cause.reasons[0]
763 assert.strictEqual(failure._tag, "Fail")
764 if (failure._tag === "Fail" && Socket.SocketError.is(failure.error)) {
765 assert.strictEqual(failure.error.reason._tag, "SocketCloseError")
766 } else {
767 assert.fail("expected a SocketError")
768 }
769 }
770 }).pipe(
771 Effect.provideService(Socket.WebSocketConstructor, (url) => new globalThis.WebSocket(url))
772 ))
773
774 it.effect("clean closes are errors", () =>
775 Effect.gen(function*() {
776 const server = yield* makeServer
777 const socket = yield* Socket.makeWebSocket(Effect.succeed(url))
778 const fiber = yield* Effect.gen(function*() {
779 const { pull } = yield* socket.reader
780 while (true) {
781 yield* pull
782 }
783 }).pipe(Effect.scoped, Effect.forkChild)
784
785 yield* Effect.promise(() => server.connected)
786 server.close({ code: 1000, reason: "done", wasClean: true })
787
788 const exit = yield* Effect.exit(Fiber.join(fiber))
789 assert.isTrue(exit._tag === "Failure")
790 if (exit._tag === "Failure") {
791 const failure = exit.cause.reasons[0]
792 if (failure._tag === "Fail" && failure.error instanceof Socket.SocketError) {
793 assert.strictEqual(failure.error.reason._tag, "SocketCloseError")
794 if (failure.error.reason._tag === "SocketCloseError") {
795 assert.strictEqual(failure.error.reason.code, 1000)
796 assert.strictEqual(failure.error.reason.closeReason, "done")
797 }
798 } else {
799 assert.fail("expected a SocketError")
800 }
801 }
802 }).pipe(
803 Effect.provideService(Socket.WebSocketConstructor, (url) => new globalThis.WebSocket(url))
804 ))
805
806 it.effect("reports send errors as SocketError", () =>
807 Effect.gen(function*() {
808 class ThrowingWebSocket extends EventTarget {
809 readonly readyState = globalThis.WebSocket.OPEN
810
811 close(): void {}
812
813 send(): void {
814 throw new Error("send failed")
815 }
816 }
817 const webSocket = new ThrowingWebSocket()
818 const socket = yield* Socket.makeWebSocket(Effect.succeed(url)).pipe(
819 Effect.provideService(
820 Socket.WebSocketConstructor,
821 () => webSocket
822 )
823 )
824 const exit = yield* Effect.scoped(Effect.gen(function*() {
825 yield* Effect.asVoid(socket.reader)
826 const writer = yield* socket.writer
827 return yield* Effect.exit(writer.write(new Uint8Array([1])))
828 }))
829 assert.strictEqual(exit._tag, "Failure")
830 if (exit._tag === "Failure") {
831 assert.strictEqual(exit.cause.reasons[0]?._tag, "Fail")
832 const reason = exit.cause.reasons[0]
833 if (reason?._tag === "Fail") {
834 assert.isTrue(Socket.SocketError.is(reason.error))
835 assert.strictEqual(reason.error.reason._tag, "SocketWriteError")
836 }
837 }
838 }))
839 })
840
841 describe("TransformStream", () => {
842 it.effect("works", () =>
843 Effect.gen(function*() {
844 const readable = Stream.make("A", "B", "C").pipe(
845 Stream.tap(() => Effect.sleep(50)),
846 Stream.toReadableStream()
847 )
848 const decoder = new TextDecoder()
849 const chunks: Array<string> = []
850 const writable = new WritableStream<Uint8Array>({
851 write(chunk) {
852 chunks.push(decoder.decode(chunk))
853 }
854 })
855
856 const socket = yield* Socket.fromTransformStream(
857 Effect.succeed({
858 readable,
859 writable
860 })
861 )
862 yield* socket.writer.pipe(
863 Effect.tap((writer) =>
864 writer.write("Hello").pipe(
865 Effect.andThen(writer.write("World"))
866 )
867 ),
868 Effect.scoped,
869 Effect.forkChild
870 )
871 const received: Array<string> = []
872 yield* Effect.gen(function*() {
873 const pull = yield* Socket.readerString(socket)
874 while (true) {
875 const chunk = yield* pull
876 for (const data of chunk) {
877 received.push(data)
878 }
879 }
880 }).pipe(
881 Effect.scoped,
882 Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void)
883 )
884
885 assert.deepStrictEqual(chunks, ["Hello", "World"])
886 assert.deepStrictEqual(received, ["A", "B", "C"])
887 }))
888
889 it.effect("reports writable stream rejection as SocketError", () =>
890 Effect.gen(function*() {
891 const socket = yield* Socket.fromTransformStream(Effect.succeed({
892 readable: new ReadableStream<Uint8Array>({}),
893 writable: new WritableStream<Uint8Array>({
894 write: () => Promise.reject(new Error("write failed"))
895 })
896 }))
897 const exit = yield* Effect.scoped(Effect.gen(function*() {
898 yield* Effect.asVoid(socket.reader)
899 const writer = yield* socket.writer
900 return yield* Effect.exit(writer.write(new Uint8Array([1])))
901 }))
902 assert.strictEqual(exit._tag, "Failure")
903 if (exit._tag === "Failure") {
904 assert.strictEqual(exit.cause.reasons[0]?._tag, "Fail")
905 const reason = exit.cause.reasons[0]
906 if (reason?._tag === "Fail") {
907 assert.isTrue(Socket.SocketError.is(reason.error))
908 assert.strictEqual(reason.error.reason._tag, "SocketWriteError")
909 }
910 }
911 }))
912 })
913})