1import * as DenoSocket from "@effect/platform-deno/DenoSocket"
2import { assert, describe, it } from "@effect/vitest"
3import { Deferred, Effect, Fiber, Queue } from "effect"
4import * as Socket from "effect/socket/Socket"
5import * as Stream from "effect/Stream"
6
7const ca = Deno.readTextFileSync(new URL("./fixtures/tls/ca.pem", import.meta.url))
8const cert = Deno.readTextFileSync(new URL("./fixtures/tls/cert.pem", import.meta.url))
9const key = Deno.readTextFileSync(new URL("./fixtures/tls/key.pem", import.meta.url))
10
11const makeTcpServer = Effect.acquireRelease(
12 Effect.sync(() => Deno.listen({ hostname: "127.0.0.1", port: 0 })),
13 (listener) => Effect.sync(() => listener.close())
14)
15
16const makeTlsServer = Effect.acquireRelease(
17 Effect.sync(() => Deno.listenTls({ hostname: "127.0.0.1", port: 0, cert, key })),
18 (listener) => Effect.sync(() => listener.close())
19)
20
21const runTlsEchoServer = (listener: Deno.TlsListener) =>
22 Effect.tryPromise({
23 try: () => listener.accept().then((conn) => conn.readable.pipeTo(conn.writable)),
24 catch: (cause) => cause
25 }).pipe(Effect.ignore)
26
27const makeTempDir = Effect.acquireRelease(
28 Effect.promise(() => Deno.makeTempDir()),
29 (path) => Effect.promise(() => Deno.remove(path, { recursive: true }))
30)
31
32const runEchoServer = (listener: Deno.Listener) =>
33 Effect.forever(
34 Effect.promise(() => listener.accept().then((conn) => conn.readable.pipeTo(conn.writable)))
35 )
36
37const makeTestConn = (options?: {
38 readonly data?: Uint8Array | undefined
39 readonly pendingRead?: boolean | undefined
40 readonly closeReadableOnCloseWrite?: boolean | undefined
41 readonly setNoDelay?: ((value?: boolean) => void) | undefined
42 readonly setKeepAlive?: ((value?: boolean) => void) | undefined
43}) => {
44 let controller!: ReadableStreamDefaultController<Uint8Array>
45 let readableClosed = false
46 let closeWrites = 0
47 const readable = new ReadableStream<Uint8Array>({
48 start(value) {
49 controller = value
50 if (options?.data) value.enqueue(options.data)
51 if (options?.pendingRead !== true) {
52 readableClosed = true
53 value.close()
54 }
55 }
56 })
57 const conn = {
58 readable,
59 writable: new WritableStream<Uint8Array>(),
60 read: () => Promise.resolve(null),
61 write: (data: Uint8Array) => Promise.resolve(data.length),
62 close() {
63 if (readableClosed) return
64 readableClosed = true
65 controller.close()
66 },
67 closeWrite() {
68 closeWrites++
69 if (options?.closeReadableOnCloseWrite && !readableClosed) {
70 readableClosed = true
71 controller.close()
72 }
73 return Promise.resolve()
74 },
75 ref() {},
76 unref() {},
77 localAddr: { transport: "tcp", hostname: "127.0.0.1", port: 0 },
78 remoteAddr: { transport: "tcp", hostname: "127.0.0.1", port: 0 },
79 [Symbol.dispose]() {
80 this.close()
81 },
82 ...(options?.setNoDelay ? { setNoDelay: options.setNoDelay } : {}),
83 ...(options?.setKeepAlive ? { setKeepAlive: options.setKeepAlive } : {})
84 } as Deno.Conn
85 return { conn, closeWrites: () => closeWrites }
86}
87
88const replaceDenoConnect = (
89 connect: (options: Deno.ConnectOptions | Deno.UnixConnectOptions) => Promise<Deno.Conn>
90) =>
91 Effect.acquireRelease(
92 Effect.sync(() => {
93 const descriptor = Object.getOwnPropertyDescriptor(Deno, "connect")!
94 Object.defineProperty(Deno, "connect", { ...descriptor, value: connect })
95 return descriptor
96 }),
97 (descriptor) => Effect.sync(() => Object.defineProperty(Deno, "connect", descriptor))
98 )
99
100describe("DenoSocket", () => {
101 it.effect("upgrades a TCP reader to TLS", () =>
102 Effect.gen(function*() {
103 const listener = yield* makeTlsServer
104 yield* runTlsEchoServer(listener).pipe(Effect.forkScoped)
105 const address = listener.addr as Deno.NetAddr
106 const socket = yield* DenoSocket.makeTcp({ hostname: address.hostname, port: address.port })
107 const writer = yield* socket.writer
108 const { pull, upgrade } = yield* socket.reader
109
110 yield* upgrade({ ca: [ca] })
111 yield* writer.writeAll(["Hello", "World"])
112
113 const decoder = new TextDecoder()
114 let output = ""
115 while (output.length < 10) {
116 for (const chunk of yield* pull) output += typeof chunk === "string" ? chunk : decoder.decode(chunk)
117 }
118 assert.strictEqual(output, "HelloWorld")
119 }))
120
121 it.effect("reports TLS handshake failures as SocketUpgradeError", () =>
122 Effect.gen(function*() {
123 const listener = yield* makeTlsServer
124 yield* runTlsEchoServer(listener).pipe(Effect.forkScoped)
125 const address = listener.addr as Deno.NetAddr
126 const socket = yield* DenoSocket.makeTcp({ hostname: address.hostname, port: address.port })
127 const { pull, upgrade } = yield* socket.reader
128
129 const error = yield* upgrade().pipe(Effect.flip)
130 assert.strictEqual(error.reason._tag, "SocketUpgradeError")
131 if (error.reason._tag === "SocketUpgradeError") assert.instanceOf(error.reason.cause, Error)
132 assert.strictEqual(yield* pull.pipe(Effect.flip), error)
133 }))
134
135 it.effect("echoes over TCP", () =>
136 Effect.gen(function*() {
137 const listener = yield* makeTcpServer
138 yield* runEchoServer(listener).pipe(Effect.forkScoped)
139 const address = listener.addr as Deno.NetAddr
140
141 const output = yield* Stream.make("Hello", "World").pipe(
142 Stream.encodeText,
143 Stream.pipeThroughChannel(DenoSocket.makeTcpChannel({ hostname: address.hostname, port: address.port })),
144 Stream.catchIf(
145 (error) => error.reason._tag === "SocketCloseError",
146 () => Stream.empty
147 ),
148 Stream.decodeText(),
149 Stream.mkString
150 )
151
152 assert.strictEqual(output, "HelloWorld")
153 }))
154
155 it.effect("keeps reading after the write side closes", () =>
156 Effect.gen(function*() {
157 const listener = yield* makeTcpServer
158 const address = listener.addr as Deno.NetAddr
159 const writeClosed = yield* Deferred.make<void>()
160 const encoder = new TextEncoder()
161 const decoder = new TextDecoder()
162
163 yield* Effect.gen(function*() {
164 const conn = yield* Effect.promise(() => listener.accept())
165 const chunks: Array<string> = []
166 const read: Effect.Effect<void> = Effect.suspend(() => {
167 const buffer = new Uint8Array(5)
168 return Effect.promise(() => conn.read(buffer)).pipe(
169 Effect.flatMap((size) => {
170 if (size === null) return Effect.void
171 chunks.push(decoder.decode(buffer.subarray(0, size)))
172 return read
173 })
174 )
175 })
176 yield* read
177 assert.strictEqual(chunks.join(""), "Hello")
178 yield* Deferred.succeed(writeClosed, undefined)
179 const writer = conn.writable.getWriter()
180 yield* Effect.promise(() => writer.ready.then(() => writer.write(encoder.encode("Hello"))))
181 writer.releaseLock()
182 yield* Effect.promise(() => conn.closeWrite())
183 }).pipe(Effect.forkScoped)
184
185 const socket = yield* DenoSocket.makeTcp({ hostname: address.hostname, port: address.port })
186 const messages = yield* Queue.unbounded<Uint8Array>()
187 const runFiber = yield* Effect.gen(function*() {
188 const pull = yield* Socket.readerBytes(socket)
189 while (true) {
190 yield* Queue.offerAll(messages, yield* pull)
191 }
192 }).pipe(Effect.scoped, Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void), Effect.forkChild)
193 yield* Effect.scoped(
194 socket.writer.pipe(Effect.flatMap((writer) => writer.write(encoder.encode("Hello"))))
195 )
196 yield* Deferred.await(writeClosed)
197
198 assert.deepStrictEqual(yield* Queue.take(messages), encoder.encode("Hello"))
199 yield* Fiber.join(runFiber)
200 }))
201
202 it.effect("half-closes an empty channel input", () =>
203 Effect.gen(function*() {
204 const listener = yield* makeTcpServer
205 const address = listener.addr as Deno.NetAddr
206 const encoder = new TextEncoder()
207
208 yield* Effect.gen(function*() {
209 const conn = yield* Effect.promise(() => listener.accept())
210 assert.strictEqual(yield* Effect.promise(() => conn.read(new Uint8Array(1))), null)
211 const writer = conn.writable.getWriter()
212 yield* Effect.promise(() => writer.write(encoder.encode("Closed")))
213 writer.releaseLock()
214 yield* Effect.promise(() => conn.closeWrite())
215 }).pipe(Effect.forkScoped)
216
217 const output = yield* Stream.empty.pipe(
218 Stream.pipeThroughChannel(DenoSocket.makeTcpChannel({ hostname: address.hostname, port: address.port })),
219 Stream.catchIf(
220 (error) => error.reason._tag === "SocketCloseError",
221 () => Stream.empty
222 ),
223 Stream.decodeText(),
224 Stream.mkString
225 )
226
227 assert.strictEqual(output, "Closed")
228 }))
229
230 it.effect("does not carry a consumed half-close into a second acquisition", () =>
231 Effect.gen(function*() {
232 const encoder = new TextEncoder()
233 const first = makeTestConn({ pendingRead: true, closeReadableOnCloseWrite: true })
234 const second = makeTestConn({ data: encoder.encode("Second") })
235 const connections = [first.conn, second.conn]
236 let index = 0
237 const socket = yield* DenoSocket.fromConn(Effect.sync(() => connections[index++]!))
238 const opened = yield* Deferred.make<void>()
239
240 const firstRun = yield* Effect.gen(function*() {
241 const { pull } = yield* socket.reader
242 yield* Deferred.succeed(opened, undefined)
243 while (true) {
244 yield* pull
245 }
246 }).pipe(Effect.scoped, Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void), Effect.forkChild)
247 yield* Deferred.await(opened)
248 yield* Effect.scoped(socket.writer)
249 yield* Fiber.join(firstRun)
250
251 const received: Array<Uint8Array> = []
252 yield* Effect.gen(function*() {
253 const pull = yield* Socket.readerBytes(socket)
254 while (true) {
255 const chunk = yield* pull
256 for (const data of chunk) {
257 received.push(data)
258 }
259 }
260 }).pipe(Effect.scoped, Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void))
261
262 assert.deepStrictEqual(received, [encoder.encode("Second")])
263 assert.strictEqual(first.closeWrites(), 1)
264 assert.strictEqual(second.closeWrites(), 0)
265 }))
266
267 it.effect("maps connection refusal to SocketOpenError", () =>
268 Effect.gen(function*() {
269 const listener = Deno.listen({ hostname: "127.0.0.1", port: 0 })
270 const address = listener.addr as Deno.NetAddr
271 listener.close()
272 const socket = yield* DenoSocket.makeTcp({ hostname: address.hostname, port: address.port })
273
274 const error = yield* Effect.scoped(Effect.asVoid(socket.reader)).pipe(Effect.flip)
275
276 assert.instanceOf(error, Socket.SocketError)
277 assert.strictEqual(error.reason._tag, "SocketOpenError")
278 if (error.reason._tag === "SocketOpenError") {
279 assert.strictEqual(error.reason.kind, "Unknown")
280 assert.instanceOf(error.reason.cause, Deno.errors.ConnectionRefused)
281 }
282 }))
283
284 it.effect("applies TCP tuning options and ignores them for Unix", () =>
285 Effect.gen(function*() {
286 const noDelay: Array<boolean | undefined> = []
287 const keepAlive: Array<boolean | undefined> = []
288 const tcp = makeTestConn({
289 setNoDelay: (value) => noDelay.push(value),
290 setKeepAlive: (value) => keepAlive.push(value)
291 })
292 const unix = makeTestConn()
293 const connections = [tcp.conn, unix.conn]
294 const connectOptions: Array<Deno.ConnectOptions | Deno.UnixConnectOptions> = []
295 let index = 0
296 yield* replaceDenoConnect((options) => {
297 connectOptions.push(options)
298 return Promise.resolve(connections[index++]!)
299 })
300
301 const tcpSocket = yield* DenoSocket.makeTcp({ port: 1, noDelay: false, keepAlive: true })
302 yield* Effect.scoped(Effect.asVoid(tcpSocket.reader))
303 const unixSocket = yield* DenoSocket.makeTcp({
304 transport: "unix",
305 path: "/unused.sock",
306 noDelay: true,
307 keepAlive: false
308 })
309 yield* Effect.scoped(Effect.asVoid(unixSocket.reader))
310
311 assert.deepStrictEqual(noDelay, [false])
312 assert.deepStrictEqual(keepAlive, [true])
313 assert.deepStrictEqual(connectOptions, [
314 { port: 1 },
315 { transport: "unix", path: "/unused.sock" }
316 ])
317 }))
318
319 it.effect("echoes over Unix sockets", () =>
320 Effect.gen(function*() {
321 const directory = yield* makeTempDir
322 const path = `${directory}/echo.sock`
323 const listener = yield* Effect.acquireRelease(
324 Effect.sync(() => Deno.listen({ transport: "unix", path })),
325 (listener) => Effect.sync(() => listener.close())
326 )
327 yield* runEchoServer(listener).pipe(Effect.forkScoped)
328
329 const output = yield* Stream.make("Hello", "Unix").pipe(
330 Stream.encodeText,
331 Stream.pipeThroughChannel(DenoSocket.makeTcpChannel({ transport: "unix", path })),
332 Stream.catchIf(
333 (error) => error.reason._tag === "SocketCloseError",
334 () => Stream.empty
335 ),
336 Stream.decodeText(),
337 Stream.mkString
338 )
339
340 assert.strictEqual(output, "HelloUnix")
341 }))
342
343 it.effect("rejects TLS upgrades on Unix sockets without disrupting the connection", () =>
344 Effect.gen(function*() {
345 const directory = yield* makeTempDir
346 const path = `${directory}/upgrade.sock`
347 const listener = yield* Effect.acquireRelease(
348 Effect.sync(() => Deno.listen({ transport: "unix", path })),
349 (listener) => Effect.sync(() => listener.close())
350 )
351 yield* runEchoServer(listener).pipe(Effect.forkScoped)
352
353 const socket = yield* DenoSocket.makeTcp({ transport: "unix", path })
354 const writer = yield* socket.writer
355 const { pull, upgrade } = yield* socket.reader
356
357 const error = yield* upgrade().pipe(Effect.flip)
358 assert.strictEqual(error.reason._tag, "SocketUpgradeError")
359
360 yield* writer.write("HelloUnix")
361 const [received] = yield* pull
362 assert.strictEqual(new TextDecoder().decode(received as Uint8Array), "HelloUnix")
363 }))
364
365 it.effect("uses Deno's native WebSocket", () =>
366 Effect.gen(function*() {
367 const server = yield* Effect.acquireRelease(
368 Effect.sync(() =>
369 Deno.serve(
370 { hostname: "127.0.0.1", port: 0, onListen: () => {} },
371 (request) => {
372 const { response, socket } = Deno.upgradeWebSocket(request)
373 socket.onmessage = (event) => socket.send(event.data)
374 return response
375 }
376 )
377 ),
378 (server) => Effect.promise(() => server.shutdown())
379 )
380 const address = server.addr as Deno.NetAddr
381 const messages = yield* Queue.unbounded<Uint8Array>()
382
383 yield* Effect.gen(function*() {
384 const socket = yield* Socket.Socket
385 const runFiber = yield* Effect.gen(function*() {
386 const pull = yield* Socket.readerBytes(socket)
387 while (true) {
388 yield* Queue.offerAll(messages, yield* pull)
389 }
390 }).pipe(
391 Effect.scoped,
392 Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void),
393 Effect.forkChild
394 )
395 const writer = yield* socket.writer
396 yield* writer.write("Hello WebSocket")
397
398 assert.deepStrictEqual(yield* Queue.take(messages), new TextEncoder().encode("Hello WebSocket"))
399 yield* Fiber.interrupt(runFiber)
400 }).pipe(
401 Effect.scoped,
402 Effect.provide(DenoSocket.layerWebSocket(`ws://${address.hostname}:${address.port}`))
403 )
404 }))
405
406 it.effect("adapts a TransformStream", () =>
407 Effect.gen(function*() {
408 const readable = Stream.make("A", "B", "C").pipe(
409 Stream.tap(() => Effect.sleep(50)),
410 Stream.toReadableStream()
411 )
412 const decoder = new TextDecoder()
413 const chunks: Array<string> = []
414 const writable = new WritableStream<Uint8Array>({
415 write(chunk) {
416 chunks.push(decoder.decode(chunk))
417 }
418 })
419
420 const socket = yield* Socket.fromTransformStream(
421 Effect.succeed({ readable, writable })
422 )
423 yield* socket.writer.pipe(
424 Effect.tap((writer) => writer.write("Hello").pipe(Effect.andThen(writer.write("World")))),
425 Effect.scoped,
426 Effect.forkChild
427 )
428 const received: Array<string> = []
429 yield* Effect.gen(function*() {
430 const pull = yield* Socket.readerString(socket)
431 while (true) {
432 const chunk = yield* pull
433 for (const data of chunk) {
434 received.push(data)
435 }
436 }
437 }).pipe(Effect.scoped, Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void))
438
439 assert.deepStrictEqual(chunks, ["Hello", "World"])
440 assert.deepStrictEqual(received, ["A", "B", "C"])
441 }))
442})