1/** @effect-diagnostics preferSchemaOverJson:skip-file */
2import { NodeHttpServer } from "@effect/platform-node"
3import { NodeWS } from "@effect/platform-node/NodeSocket"
4import { assert, describe, expect, it } from "@effect/vitest"
5import { ByteSize, Effect, Option } from "effect"
6import * as Deferred from "effect/Deferred"
7import * as Duration from "effect/Duration"
8import * as Fiber from "effect/Fiber"
9import { constVoid } from "effect/Function"
10import {
11 Cookies,
12 FetchHttpClient,
13 HttpBody,
14 HttpClient,
15 HttpClientRequest,
16 HttpClientResponse,
17 HttpPlatform,
18 HttpRouter,
19 HttpServer,
20 HttpServerRequest,
21 HttpServerRespondable,
22 HttpServerResponse,
23 Multipart,
24 UrlParams
25} from "effect/http"
26import * as HttpApiError from "effect/http-api/HttpApiError"
27import * as Latch from "effect/Latch"
28import * as Layer from "effect/Layer"
29import * as ManagedRuntime from "effect/ManagedRuntime"
30import * as NetAddress from "effect/net/NetAddress"
31import * as Schema from "effect/Schema"
32import { Socket } from "effect/socket"
33import * as Stream from "effect/Stream"
34import * as Tracer from "effect/Tracer"
35import * as Buffer from "node:buffer"
36import { randomBytes } from "node:crypto"
37import { EventEmitter } from "node:events"
38import * as Http from "node:http"
39import * as Net from "node:net"
40
41const Todo = Schema.Struct({
42 id: Schema.Number,
43 title: Schema.String
44})
45const IdParams = Schema.Struct({
46 id: Schema.FiniteFromString
47})
48const todoResponse = HttpServerResponse.schemaJson(Todo)
49
50describe("HttpServer", () => {
51 it.effect("keeps routes isolated between independent servers", () =>
52 Effect.gen(function*() {
53 const publicServer = Http.createServer()
54 const internalServer = Http.createServer()
55
56 const Health = Layer.effectDiscard(
57 Effect.flatMap(HttpRouter.HttpRouter, (router) =>
58 router.add("GET", "/health", HttpServerResponse.text("healthy")))
59 )
60 const publicApp = Layer.mergeAll(
61 HttpRouter.add("GET", "/public", HttpServerResponse.text("public")),
62 Health
63 )
64 const internalApp = Layer.mergeAll(
65 HttpRouter.add("GET", "/internal", HttpServerResponse.text("internal")),
66 Health
67 )
68
69 yield* Layer.mergeAll(
70 HttpRouter.serve(publicApp, { disableListenLog: true, disableLogger: true }).pipe(
71 Layer.provide(NodeHttpServer.layer(() =>
72 publicServer, { port: 0 }))
73 ),
74 HttpRouter.serve(internalApp, { disableListenLog: true, disableLogger: true }).pipe(
75 Layer.provide(NodeHttpServer.layer(() => internalServer, { port: 0 }))
76 )
77 ).pipe(Layer.build)
78
79 const status = (port: number, path: string) =>
80 Effect.promise(() => fetch("http://localhost:" + port + path).then((response) => response.status))
81
82 assert.strictEqual(yield* status(tcpPort(publicServer), "/public"), 200)
83 assert.strictEqual(yield* status(tcpPort(internalServer), "/internal"), 200)
84 assert.strictEqual(yield* status(tcpPort(publicServer), "/health"), 200)
85 assert.strictEqual(yield* status(tcpPort(internalServer), "/health"), 200)
86 assert.strictEqual(yield* status(tcpPort(publicServer), "/internal"), 404)
87 assert.strictEqual(yield* status(tcpPort(internalServer), "/public"), 404)
88 }))
89
90 it.effect("schema", () =>
91 Effect.gen(function*() {
92 yield* HttpRouter.add(
93 "GET",
94 "/todos/:id",
95 Effect.flatMap(
96 HttpRouter.schemaParams(IdParams),
97 ({ id }) => todoResponse({ id, title: "test" })
98 )
99 ).pipe(
100 HttpRouter.serve,
101 Layer.build
102 )
103 const todo = yield* HttpClient.get("/todos/1").pipe(
104 Effect.flatMap(HttpClientResponse.schemaBodyJson(Todo))
105 )
106 expect(todo).toEqual({ id: 1, title: "test" })
107 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
108
109 it.effect("schemaJson applies a JSON reviver", () =>
110 Effect.gen(function*() {
111 yield* HttpRouter.add(
112 "POST",
113 "/json",
114 Effect.gen(function*() {
115 const decoded = yield* HttpRouter.schemaJson(
116 Schema.Struct({ body: Schema.Struct({ value: Schema.String }) }),
117 { reviver: (key, value) => key === "value" ? "revived" : value }
118 )
119 assert.deepStrictEqual(decoded, { body: { value: "revived" } })
120 return HttpServerResponse.empty()
121 })
122 ).pipe(HttpRouter.serve, Layer.build)
123
124 const response = yield* HttpClient.post("/json", {
125 body: HttpBody.jsonUnsafe({ value: "original" })
126 })
127 assert.strictEqual(response.status, 204)
128 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
129
130 it.effect.each(["Uploaded", ""] as const)("forwards status text %j", (statusText) =>
131 Effect.gen(function*() {
132 yield* HttpRouter.add(
133 "GET",
134 "/",
135 HttpServerResponse.empty({ status: 201, statusText })
136 ).pipe(
137 HttpRouter.serve,
138 Layer.build
139 )
140 const server = yield* HttpServer.HttpServer
141 assert.isTrue(NetAddress.isInetAddress(server.address))
142 if (!NetAddress.isInetAddress(server.address)) return
143 const port = server.address.port
144
145 assert.strictEqual(yield* getStatusText(port), statusText)
146 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
147
148 it.effect("formData", () =>
149 Effect.gen(function*() {
150 yield* HttpRouter.add(
151 "POST",
152 "/upload",
153 Effect.gen(function*() {
154 const request = yield* HttpServerRequest.HttpServerRequest
155 const formData = yield* request.multipart
156 const part = formData.file
157 assert(typeof part !== "string")
158 const file = part[0]
159 assert(typeof file !== "string")
160 expect(file.path.endsWith("/test.txt")).toEqual(true)
161 expect(file.contentType).toEqual("text/plain")
162 return yield* HttpServerResponse.json({ ok: "file" in formData })
163 })
164 ).pipe(
165 HttpRouter.serve,
166 Layer.build
167 )
168 const client = yield* HttpClient.HttpClient
169 const formData = new FormData()
170 formData.append("file", new Blob(["test"], { type: "text/plain" }), "test.txt")
171 const result = yield* client.post("/upload", { body: HttpBody.formData(formData) }).pipe(
172 Effect.flatMap((r) => r.json)
173 )
174 expect(result).toEqual({ ok: true })
175 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
176
177 it.effect("schemaBodyForm", () =>
178 Effect.gen(function*() {
179 yield* HttpRouter.add(
180 "POST",
181 "/upload",
182 Effect.gen(function*() {
183 const files = yield* HttpServerRequest.schemaBodyForm(Schema.Struct({
184 file: Multipart.FilesSchema,
185 test: Schema.String
186 }))
187 expect(files).toHaveProperty("file")
188 expect(files).toHaveProperty("test")
189 return HttpServerResponse.empty()
190 })
191 ).pipe(
192 HttpRouter.serve,
193 Layer.build
194 )
195 const client = yield* HttpClient.HttpClient
196 const formData = new FormData()
197 formData.append("file", new Blob(["test"], { type: "text/plain" }), "test.txt")
198 formData.append("test", "test")
199 const response = yield* client.post("/upload", { body: HttpBody.formData(formData) })
200 expect(response.status).toEqual(204)
201 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
202
203 it.effect("formData withMaxFileSize", () =>
204 Effect.gen(function*() {
205 yield* HttpRouter.add(
206 "POST",
207 "/upload",
208 Effect.gen(function*() {
209 const request = yield* HttpServerRequest.HttpServerRequest
210 yield* request.multipart
211 return HttpServerResponse.empty()
212 }).pipe(
213 Effect.catchTag("MultipartError", (error) =>
214 error.reason._tag === "FileTooLarge" ?
215 Effect.succeed(HttpServerResponse.empty({ status: 413 })) :
216 Effect.fail(error))
217 )
218 ).pipe(
219 HttpRouter.serve,
220 Layer.build,
221 Effect.provideService(Multipart.MaxFileSize, ByteSize.bytes(100))
222 )
223 const client = yield* HttpClient.HttpClient
224 const formData = new FormData()
225 const data = new Uint8Array(1000)
226 formData.append("file", new Blob([data], { type: "text/plain" }), "test.txt")
227 const response = yield* client.post("/upload", { body: HttpBody.formData(formData) })
228 expect(response.status).toEqual(413)
229 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
230
231 it.effect("formData withMaxFieldSize", () =>
232 Effect.gen(function*() {
233 yield* HttpRouter.add(
234 "POST",
235 "/upload",
236 Effect.gen(function*() {
237 const request = yield* HttpServerRequest.HttpServerRequest
238 yield* request.multipart
239 return HttpServerResponse.empty()
240 }).pipe(
241 Effect.catchTag("MultipartError", (error) =>
242 error.reason._tag === "FieldTooLarge" ?
243 Effect.succeed(HttpServerResponse.empty({ status: 413 })) :
244 Effect.fail(error))
245 )
246 ).pipe(
247 HttpRouter.serve,
248 Layer.build,
249 Effect.provideService(Multipart.MaxFieldSize, ByteSize.bytes(100))
250 )
251 const client = yield* HttpClient.HttpClient
252 const formData = new FormData()
253 const data = new Uint8Array(1000).fill(1)
254 formData.append("file", new TextDecoder().decode(data))
255 const response = yield* client.post("/upload", { body: HttpBody.formData(formData) })
256 expect(response.status).toEqual(413)
257 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
258
259 it.effect("mountApp", () =>
260 Effect.gen(function*() {
261 const child = Effect.map(HttpServerRequest.HttpServerRequest, (_) => HttpServerResponse.text(_.url))
262 yield* HttpRouter.use((router) => router.prefixed("/child").add("*", "*", child)).pipe(
263 HttpRouter.serve,
264 Layer.build
265 )
266 const client = yield* HttpClient.HttpClient
267 const todo = yield* client.get("/child/1").pipe(Effect.flatMap((_) => _.text))
268 expect(todo).toEqual("/1")
269 const root = yield* client.get("/child").pipe(Effect.flatMap((_) => _.text))
270 expect(root).toEqual("/")
271 const rootSearch = yield* client.get("/child?foo=bar").pipe(Effect.flatMap((_) => _.text))
272 expect(rootSearch).toEqual("?foo=bar")
273 const rootSlash = yield* client.get("/child/").pipe(Effect.flatMap((_) => _.text))
274 expect(rootSlash).toEqual("/")
275 const invalid = yield* client.get("/child1/", {
276 urlParams: { foo: "bar" }
277 }).pipe(Effect.map((_) => _.status))
278 expect(invalid).toEqual(404)
279 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
280
281 it.effect("file", () =>
282 Effect.gen(function*() {
283 yield* (yield* HttpServerResponse.file(`${__dirname}/fixtures/text.txt`).pipe(
284 Effect.updateService(
285 HttpPlatform.HttpPlatform,
286 (_) => ({
287 ..._,
288 fileResponse: (path, options) =>
289 Effect.map(
290 _.fileResponse(path, options),
291 (res) => {
292 ;(res as any).headers.etag = "\"etag\""
293 return res
294 }
295 )
296 })
297 )
298 )).pipe(
299 Effect.succeed,
300 HttpServer.serveEffect()
301 )
302 const client = yield* HttpClient.HttpClient
303 const res = yield* client.get("/")
304 expect(res.status).toEqual(200)
305 expect(res.headers["content-type"]).toEqual("text/plain")
306 expect(res.headers["content-length"]).toEqual("27")
307 expect(res.headers.etag).toEqual("\"etag\"")
308 const text = yield* res.text
309 expect(text.trim()).toEqual("lorem ipsum dolar sit amet")
310 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
311
312 it.effect("fileWeb", () =>
313 Effect.gen(function*() {
314 const now = new Date()
315 const file = new Buffer.File([new TextEncoder().encode("test")], "test.txt", {
316 type: "text/plain",
317 lastModified: now.getTime()
318 })
319 yield* HttpServerResponse.fileWeb(file).pipe(
320 Effect.updateService(
321 HttpPlatform.HttpPlatform,
322 (_) => ({
323 ..._,
324 fileWebResponse: (path, options) =>
325 Effect.map(
326 _.fileWebResponse(path, options),
327 (res) => ({ ...res, headers: { ...res.headers, etag: "W/\"etag\"" } })
328 )
329 })
330 ),
331 HttpServer.serveEffect()
332 )
333 const client = yield* HttpClient.HttpClient
334 const res = yield* client.get("/")
335 expect(res.status).toEqual(200)
336 expect(res.headers["content-type"]).toEqual("text/plain")
337 expect(res.headers["content-length"]).toEqual("4")
338 expect(res.headers["last-modified"]).toEqual(now.toUTCString())
339 expect(res.headers.etag).toEqual("W/\"etag\"")
340 const text = yield* res.text
341 expect(text.trim()).toEqual("test")
342 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
343
344 it.effect("schemaBodyUrlParams", () =>
345 Effect.gen(function*() {
346 yield* HttpRouter.add(
347 "POST",
348 "/todos",
349 Effect.flatMap(
350 HttpServerRequest.schemaBodyUrlParams(Schema.Struct({
351 id: Schema.FiniteFromString,
352 title: Schema.String
353 })),
354 ({ id, title }) => todoResponse({ id, title })
355 )
356 ).pipe(
357 HttpRouter.serve,
358 Layer.build
359 )
360 const todo = yield* HttpClientRequest.post("/todos").pipe(
361 HttpClientRequest.bodyUrlParams({ id: "1", title: "test" }),
362 HttpClient.execute,
363 Effect.flatMap(HttpClientResponse.schemaBodyJson(Todo))
364 )
365 expect(todo).toEqual({ id: 1, title: "test" })
366 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
367
368 it.effect("schemaBodyUrlParams error", () =>
369 Effect.gen(function*() {
370 yield* HttpRouter.add(
371 "GET",
372 "/todos",
373 Effect.flatMap(
374 HttpServerRequest.schemaBodyUrlParams(Schema.Struct({
375 id: Schema.FiniteFromString,
376 title: Schema.String
377 })),
378 ({ id, title }) => todoResponse({ id, title })
379 ).pipe(
380 Effect.catchTag("SchemaError", (error) =>
381 Effect.succeed(HttpServerResponse.jsonUnsafe({ error }, { status: 400 })))
382 )
383 ).pipe(
384 HttpRouter.serve,
385 Layer.build
386 )
387 const client = yield* HttpClient.HttpClient
388 const response = yield* client.get("/todos")
389 expect(response.status).toEqual(400)
390 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
391
392 it.effect("schemaBodyFormJson", () =>
393 Effect.gen(function*() {
394 yield* HttpRouter.add(
395 "POST",
396 "/upload",
397 Effect.gen(function*() {
398 const result = yield* HttpServerRequest.schemaBodyFormJson(
399 Schema.Struct({
400 test: Schema.String
401 }),
402 { reviver: (key, value) => key === "test" ? "revived" : value }
403 )("json")
404 expect(result.test).toEqual("revived")
405 return HttpServerResponse.empty()
406 })
407 ).pipe(
408 HttpRouter.serve,
409 Layer.build
410 )
411 const client = yield* HttpClient.HttpClient
412 const formData = new FormData()
413 formData.append("json", JSON.stringify({ test: "content" }))
414 const response = yield* client.post("/upload", { body: HttpBody.formData(formData) })
415 expect(response.status).toEqual(204)
416 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
417
418 it.effect("schemaBodyFormJson file", () =>
419 Effect.gen(function*() {
420 yield* HttpRouter.add(
421 "POST",
422 "/upload",
423 Effect.gen(function*() {
424 const result = yield* HttpServerRequest.schemaBodyFormJson(Schema.Struct({
425 test: Schema.String
426 }))("json")
427
428 expect(result.test).toEqual("content")
429 return HttpServerResponse.empty()
430 })
431 ).pipe(
432 HttpRouter.serve,
433 Layer.build
434 )
435 const client = yield* HttpClient.HttpClient
436 const formData = new FormData()
437 formData.append(
438 "json",
439 new Blob([JSON.stringify({ test: "content" })], { type: "application/json" }),
440 "test.json"
441 )
442 const response = yield* client.post("/upload", { body: HttpBody.formData(formData) })
443 expect(response.status).toEqual(204)
444 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
445
446 it.effect("schemaBodyFormJson url encoded", () =>
447 Effect.gen(function*() {
448 yield* HttpRouter.add(
449 "POST",
450 "/upload",
451 Effect.gen(function*() {
452 const result = yield* HttpServerRequest.schemaBodyFormJson(Schema.Struct({
453 test: Schema.String
454 }))("json")
455 expect(result.test).toEqual("content")
456 return HttpServerResponse.empty()
457 })
458 ).pipe(
459 HttpRouter.serve,
460 Layer.build
461 )
462 const client = yield* HttpClient.HttpClient
463 const response = yield* client.post("/upload", {
464 body: HttpBody.urlParams(UrlParams.fromInput({
465 json: JSON.stringify({ test: "content" })
466 }))
467 })
468 expect(response.status).toEqual(204)
469 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
470
471 it.effect("tracing", () =>
472 Effect.gen(function*() {
473 yield* HttpRouter.add(
474 "GET",
475 "/",
476 Effect.flatMap(
477 Effect.currentSpan,
478 (_) => HttpServerResponse.json({ spanId: _.spanId, parent: _.parent })
479 )
480 ).pipe(
481 HttpRouter.serve,
482 Layer.build
483 )
484 const client = yield* HttpClient.HttpClient
485 const requestSpan = yield* Effect.makeSpan("client request")
486 const body = yield* client.get("/").pipe(
487 Effect.flatMap((r) => r.json),
488 Effect.provideService(
489 Tracer.Tracer,
490 Tracer.make({
491 span(options) {
492 assert.strictEqual(options.name, "http.client GET")
493 assert.strictEqual(options.kind, "client")
494 assert(options.parent._tag === "Some")
495 if (options.parent.value._tag !== "Span") {
496 throw new Error("Expected span parent")
497 }
498 assert.strictEqual(options.parent.value.name, "request parent")
499 return requestSpan
500 }
501 })
502 ),
503 Effect.withSpan("request parent"),
504 Effect.repeat({ times: 2 })
505 )
506 expect((body as any).parent._tag).toEqual("Some")
507 expect((body as any).parent.value.spanId).toEqual(requestSpan.spanId)
508 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
509
510 it.effect("html", () =>
511 Effect.gen(function*() {
512 yield* HttpRouter.addAll([
513 HttpRouter.route("GET", "/home", HttpServerResponse.html("<html />")),
514 HttpRouter.route(
515 "GET",
516 "/about",
517 HttpServerResponse.html`<html>${Effect.succeed("<body />")}</html>`
518 ),
519 HttpRouter.route(
520 "GET",
521 "/stream",
522 HttpServerResponse.htmlStream`<html>${Stream.make("<body />", 123, "hello")}</html>`
523 )
524 ]).pipe(
525 HttpRouter.serve,
526 Layer.build
527 )
528 const client = yield* HttpClient.HttpClient
529 const home = yield* client.get("/home").pipe(Effect.flatMap((r) => r.text))
530 expect(home).toEqual("<html />")
531 const about = yield* client.get("/about").pipe(Effect.flatMap((r) => r.text))
532 expect(about).toEqual("<html><body /></html>")
533 const stream = yield* client.get("/stream").pipe(Effect.flatMap((r) => r.text))
534 expect(stream).toEqual("<html><body />123hello</html>")
535 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
536
537 it.effect("setCookie", () =>
538 Effect.gen(function*() {
539 yield* HttpRouter.add(
540 "GET",
541 "/home",
542 HttpServerResponse.empty().pipe(
543 HttpServerResponse.setCookieUnsafe("test", "value"),
544 HttpServerResponse.setCookieUnsafe("test2", "value2", {
545 httpOnly: true,
546 secure: true,
547 sameSite: "lax",
548 partitioned: true,
549 path: "/",
550 domain: "example.com",
551 expires: new Date(2022, 1, 1, 0, 0, 0, 0),
552 maxAge: "5 minutes"
553 })
554 )
555 ).pipe(
556 HttpRouter.serve,
557 Layer.build
558 )
559 const client = yield* HttpClient.HttpClient
560 const res = yield* client.get("/home")
561 assert.deepStrictEqual(
562 res.cookies.toJSON(),
563 Cookies.fromReadonlyRecord({
564 test: Cookies.makeCookieUnsafe("test", "value"),
565 test2: Cookies.makeCookieUnsafe("test2", "value2", {
566 httpOnly: true,
567 secure: true,
568 sameSite: "lax",
569 partitioned: true,
570 path: "/",
571 domain: "example.com",
572 expires: new Date(2022, 1, 1, 0, 0, 0, 0),
573 maxAge: Duration.minutes(5)
574 })
575 }).toJSON()
576 )
577 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
578
579 it.live("uninterruptible routes", () =>
580 Effect.gen(function*() {
581 yield* HttpRouter.add(
582 "GET",
583 "/home",
584 Effect.gen(function*() {
585 const fiber = Fiber.getCurrent()!
586 setTimeout(() => fiber.interruptUnsafe(fiber.id), 10)
587 yield* Effect.sleep(50)
588 return HttpServerResponse.empty()
589 }),
590 { uninterruptible: true }
591 ).pipe(
592 HttpRouter.serve,
593 Layer.build
594 )
595 const client = yield* HttpClient.HttpClient
596 const res = yield* client.get("/home")
597 assert.strictEqual(res.status, 204)
598 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
599
600 it.effect("completes a HEAD response once when close precedes the end callback", () =>
601 Effect.gen(function*() {
602 const scope = yield* Effect.scope
603 const handler = yield* NodeHttpServer.makeHandler(
604 Effect.succeed(HttpServerResponse.empty()),
605 { scope }
606 )
607 const completed = Latch.makeUnsafe()
608 let writableEnded = false
609 const nodeResponse = Object.defineProperty(new EventEmitter(), "writableEnded", {
610 get: () => writableEnded
611 }) as Http.ServerResponse
612 let closeListenerRemovals = 0
613 nodeResponse.writeHead = () => nodeResponse
614 nodeResponse.off = ((event: string | symbol, listener: (...args: Array<unknown>) => void) => {
615 if (event === "close") {
616 closeListenerRemovals++
617 }
618 return EventEmitter.prototype.off.call(nodeResponse, event, listener) as Http.ServerResponse
619 }) as Http.ServerResponse["off"]
620 nodeResponse.end = ((callback: () => void) => {
621 writableEnded = true
622 nodeResponse.emit("close")
623 callback()
624 completed.openUnsafe()
625 return nodeResponse
626 }) as Http.ServerResponse["end"]
627
628 handler(
629 { method: "HEAD", url: "/", headers: {}, socket: {} } as Http.IncomingMessage,
630 nodeResponse
631 )
632 yield* completed.await
633
634 assert.strictEqual(closeListenerRemovals, 1)
635 }))
636
637 it.effect("returns none when a late remoteAddress read finds no socket", () =>
638 Effect.gen(function*() {
639 const scope = yield* Effect.scope
640 let request: HttpServerRequest.HttpServerRequest | undefined
641 const handler = yield* NodeHttpServer.makeHandler(
642 Effect.gen(function*() {
643 request = yield* HttpServerRequest.HttpServerRequest
644 return HttpServerResponse.empty()
645 }),
646 { scope }
647 )
648 const completed = Latch.makeUnsafe()
649 let writableEnded = false
650 const nodeResponse = Object.defineProperty(new EventEmitter(), "writableEnded", {
651 get: () => writableEnded
652 }) as Http.ServerResponse
653 nodeResponse.writeHead = () => nodeResponse
654 nodeResponse.end = (() => {
655 writableEnded = true
656 completed.openUnsafe()
657 return nodeResponse
658 }) as Http.ServerResponse["end"]
659 const nodeRequest = {
660 method: "GET",
661 url: "/",
662 headers: {},
663 socket: { remoteAddress: "127.0.0.1" }
664 } as unknown as Http.IncomingMessage
665
666 handler(nodeRequest, nodeResponse)
667 yield* completed.await
668 assert(request !== undefined)
669 ;(nodeRequest as unknown as { socket: null }).socket = null
670
671 assert.deepStrictEqual(request.remoteAddress, Option.none())
672 }))
673
674 it.effect("coalesces streaming chunks from the same pull", () =>
675 Effect.gen(function*() {
676 const scope = yield* Effect.scope
677 const handler = yield* NodeHttpServer.makeHandler(
678 Effect.succeed(HttpServerResponse.stream(Stream.make(
679 Buffer.Buffer.from("a"),
680 Buffer.Buffer.from("b")
681 ))),
682 { scope }
683 )
684 const completed = Latch.makeUnsafe()
685 const writes: Array<Uint8Array> = []
686 let writableEnded = false
687 const nodeResponse = Object.defineProperty(new EventEmitter(), "writableEnded", {
688 get: () => writableEnded
689 }) as Http.ServerResponse
690 nodeResponse.writeHead = () => nodeResponse
691 nodeResponse.write = ((chunk: Uint8Array) => {
692 writes.push(chunk)
693 return true
694 }) as Http.ServerResponse["write"]
695 nodeResponse.end = (() => {
696 writableEnded = true
697 completed.openUnsafe()
698 return nodeResponse
699 }) as Http.ServerResponse["end"]
700
701 handler(
702 { method: "GET", url: "/", headers: {}, socket: {} } as Http.IncomingMessage,
703 nodeResponse
704 )
705 yield* completed.await
706
707 assert.deepStrictEqual(writes.map((chunk) => Buffer.Buffer.from(chunk).toString()), ["ab"])
708 }))
709
710 it.effect("waits for drain after a streaming write applies backpressure", () =>
711 Effect.gen(function*() {
712 const scope = yield* Effect.scope
713 const handler = yield* NodeHttpServer.makeHandler(
714 Effect.succeed(HttpServerResponse.stream(Stream.make(
715 Buffer.Buffer.from("a"),
716 Buffer.Buffer.from("b")
717 ))),
718 { scope }
719 )
720 const writeObserved = Latch.makeUnsafe()
721 const completed = Latch.makeUnsafe()
722 let writableEnded = false
723 const nodeResponse = Object.defineProperty(new EventEmitter(), "writableEnded", {
724 get: () => writableEnded
725 }) as Http.ServerResponse
726 let writeCount = 0
727 nodeResponse.writeHead = () => nodeResponse
728 nodeResponse.write = (() => {
729 writeCount++
730 queueMicrotask(() => writeObserved.openUnsafe())
731 return writeCount > 1
732 }) as Http.ServerResponse["write"]
733 nodeResponse.end = (() => {
734 writableEnded = true
735 completed.openUnsafe()
736 return nodeResponse
737 }) as Http.ServerResponse["end"]
738
739 handler(
740 { method: "GET", url: "/", headers: {}, socket: {} } as Http.IncomingMessage,
741 nodeResponse
742 )
743 yield* writeObserved.await
744 assert.strictEqual(nodeResponse.writableEnded, false)
745
746 nodeResponse.emit("drain")
747 yield* completed.await
748 assert.strictEqual(writeCount, 1)
749 }))
750
751 it.live("disposes after a client aborts a handler awaiting an upstream request", () => {
752 const upstreamStarted = Latch.makeUnsafe()
753 const upstream = Http.createServer(() => {
754 upstreamStarted.openUnsafe()
755 })
756 const server = Http.createServer()
757 const router = HttpRouter.use((router) =>
758 router.add(
759 "GET",
760 "/",
761 Effect.gen(function*() {
762 const request = yield* HttpServerRequest.HttpServerRequest
763 if (request.method !== "HEAD") {
764 return HttpServerResponse.text("ok")
765 }
766 yield* HttpClient.head(`http://localhost:${tcpPort(upstream)}`)
767 return HttpServerResponse.empty()
768 })
769 )
770 )
771 const runtime = ManagedRuntime.make(
772 Layer.effectDiscard(
773 HttpRouter.serve(router).pipe(
774 Layer.provide(NodeHttpServer.layer(() => server, {
775 port: 0,
776 gracefulShutdownTimeout: "100 millis"
777 })),
778 Layer.provide(FetchHttpClient.layer),
779 Layer.launch,
780 Effect.forkScoped,
781 Effect.asVoid
782 )
783 )
784 )
785
786 return Effect.gen(function*() {
787 yield* Effect.callback<void>((resume) => {
788 upstream.listen(0, () => resume(Effect.void))
789 })
790 yield* Effect.promise(() => runtime.context())
791 yield* Effect.callback<void>((resume) => {
792 if (server.listening) {
793 resume(Effect.void)
794 } else {
795 server.once("listening", () => resume(Effect.void))
796 }
797 })
798
799 const downstreamClosed = Latch.makeUnsafe()
800 const downstream = Http.request({ method: "HEAD", port: tcpPort(server), path: "/" })
801 downstream.on("error", () => {})
802 downstream.on("close", () => downstreamClosed.openUnsafe())
803 downstream.end()
804 yield* upstreamStarted.await
805 downstream.destroy()
806 yield* downstreamClosed.await
807
808 const disposed = yield* Effect.promise(() => runtime.dispose()).pipe(
809 Effect.timeoutOption("2 seconds")
810 )
811 assert.strictEqual(disposed._tag, "Some")
812 }).pipe(
813 Effect.ensuring(
814 Effect.sync(() => {
815 upstream.closeAllConnections()
816 upstream.close()
817 }).pipe(
818 Effect.andThen(Effect.promise(() => runtime.dispose())),
819 Effect.timeoutOrElse({ duration: "2 seconds", orElse: () => Effect.void })
820 )
821 )
822 )
823 })
824
825 describe("HttpServerRespondable", () => {
826 it.effect("error/schema", () =>
827 Effect.gen(function*() {
828 class CustomError extends Schema.Error<CustomError>("CustomError")({
829 _tag: Schema.tag("CustomError"),
830 name: Schema.String
831 }) {
832 [HttpServerRespondable.symbol]() {
833 return HttpServerResponse.schemaJson(CustomError)(this, { status: 599 })
834 }
835 }
836 yield* HttpRouter.add(
837 "GET",
838 "/home",
839 new CustomError({ name: "test" })
840 ).pipe(
841 HttpRouter.serve,
842 Layer.build
843 )
844 const client = yield* HttpClient.HttpClient
845 const res = yield* client.get("/home")
846 assert.strictEqual(res.status, 599)
847 const err = yield* HttpClientResponse.schemaBodyJson(CustomError)(res)
848 assert.deepStrictEqual(err, new CustomError({ name: "test" }))
849 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
850
851 it.effect("httpapi error", () =>
852 Effect.gen(function*() {
853 yield* HttpRouter.add(
854 "GET",
855 "/home",
856 new HttpApiError.BadRequest({})
857 ).pipe(
858 HttpRouter.serve,
859 Layer.build
860 )
861 const client = yield* HttpClient.HttpClient
862 const res = yield* client.get("/home")
863 assert.strictEqual(res.status, 400)
864 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
865 })
866
867 it.effect("RouterConfig", () =>
868 Effect.gen(function*() {
869 yield* HttpRouter.add(
870 "GET",
871 "/:param",
872 Effect.succeed(HttpServerResponse.empty())
873 ).pipe(
874 HttpRouter.serve,
875 Layer.build
876 )
877 let res = yield* HttpClient.get("/123456")
878 assert.strictEqual(res.status, 404)
879 res = yield* HttpClient.get("/12345")
880 assert.strictEqual(res.status, 204)
881 }).pipe(
882 Effect.provide([
883 NodeHttpServer.layerTest,
884 Layer.succeed(HttpRouter.RouterConfig)({ maxParamLength: 5 })
885 ])
886 ))
887
888 it.effect("HttpRouter prefixed", () =>
889 Effect.gen(function*() {
890 const handler = HttpRouter.serve(HttpRouter.use(Effect.fnUntraced(function*(router_) {
891 const router = router_.prefixed("/todos")
892 yield* router.add(
893 "GET",
894 "/:id",
895 Effect.flatMap(
896 HttpRouter.schemaParams(IdParams),
897 ({ id }) => todoResponse({ id, title: "test" })
898 )
899 )
900 yield* router.addAll([
901 HttpRouter.route("GET", "/", Effect.succeed(HttpServerResponse.text("root")))
902 ])
903 })))
904
905 yield* Layer.build(handler)
906
907 const todo = yield* HttpClient.get("/todos/1").pipe(
908 Effect.flatMap(HttpClientResponse.schemaBodyJson(Todo))
909 )
910 expect(todo).toEqual({ id: 1, title: "test" })
911 const root = yield* HttpClient.get("/todos").pipe(
912 Effect.flatMap((r) => r.text)
913 )
914 expect(root).toEqual("root")
915 }).pipe(Effect.provide(NodeHttpServer.layerTest)))
916
917 it.effect("websocket options are forwarded to the WebSocketServer", () =>
918 Effect.gen(function*() {
919 yield* HttpRouter.add(
920 "GET",
921 "/ws",
922 Effect.gen(function*() {
923 const request = yield* HttpServerRequest.HttpServerRequest
924 const socket = yield* Effect.orDie(request.upgrade)
925 yield* Effect.gen(function*() {
926 const { pull } = yield* socket.reader
927 while (true) {
928 yield* pull
929 }
930 }).pipe(
931 Effect.scoped,
932 Effect.catchTag("SocketError", () => Effect.void),
933 Effect.orDie
934 )
935 return HttpServerResponse.empty()
936 })
937 ).pipe(
938 HttpRouter.serve,
939 Layer.build
940 )
941 const server = yield* HttpServer.HttpServer
942 const port = (server.address as NetAddress.InetAddress).port
943
944 const connect = (perMessageDeflate: boolean) =>
945 Effect.acquireRelease(
946 Effect.callback<NodeWS.WebSocket, Error>((resume) => {
947 const ws = new NodeWS.WebSocket(`ws://127.0.0.1:${port}/ws`, { perMessageDeflate })
948 ws.on("open", () => resume(Effect.succeed(ws)))
949 ws.on("error", (error) => resume(Effect.fail(error)))
950 }),
951 (ws) => Effect.sync(() => ws.close())
952 )
953
954 // layerTest configures websocket: { perMessageDeflate: true }, so the
955 // server accepts the extension when the client offers it...
956 const compressed = yield* connect(true)
957 expect(compressed.extensions).toContain("permessage-deflate")
958
959 // ...and clients that do not offer it still connect uncompressed.
960 const plain = yield* connect(false)
961 expect(plain.extensions).not.toContain("permessage-deflate")
962 }).pipe(Effect.scoped, Effect.provide(layerTestWebsocket)))
963
964 for (
965 const [name, exit, code] of [
966 ["success", "success", 1000],
967 ["interrupt", "interrupt", 1001],
968 ["failure", "failure", 1011],
969 ["defect", "defect", 1011],
970 ["explicit close before failure", "explicit", 4400]
971 ] as const
972 ) {
973 it.effect(`closes a WebSocket with the handler's ${name} code`, () =>
974 Effect.gen(function*() {
975 const opened = yield* Deferred.make<void>()
976 yield* HttpRouter.add(
977 "GET",
978 "/ws",
979 Effect.gen(function*() {
980 const request = yield* HttpServerRequest.HttpServerRequest
981 const socket = yield* request.upgrade
982 yield* socket.reader
983 yield* Deferred.await(opened)
984 if (exit === "explicit") {
985 const writer = yield* socket.writer
986 yield* writer.write(new Socket.CloseEvent(4400, "handler closed"))
987 }
988 if (exit === "interrupt") return yield* Effect.interrupt
989 if (exit === "failure" || exit === "explicit") return yield* Effect.fail(new Error("handler failed"))
990 if (exit === "defect") return yield* Effect.die(new Error("handler defect"))
991 return HttpServerResponse.empty()
992 })
993 ).pipe((layer) => HttpRouter.serve(layer), Layer.build)
994 const server = yield* HttpServer.HttpServer
995 const port = (server.address as NetAddress.InetAddress).port
996 const actual = yield* Effect.callback<number, Error>((resume) => {
997 const ws = new NodeWS.WebSocket(`ws://127.0.0.1:${port}/ws`)
998 ws.on("open", () => {
999 Effect.runSync(Deferred.succeed(opened, undefined))
1000 })
1001 ws.on("close", (code) => resume(Effect.succeed(code)))
1002 ws.on("error", (error) => resume(Effect.fail(error)))
1003 return Effect.sync(() => ws.close())
1004 })
1005 assert.strictEqual(actual, code)
1006 }).pipe(Effect.timeout("5 seconds"), Effect.provide(layerTestWebsocket)), 10000)
1007 }
1008
1009 it.effect("an upgrade connection reset by the peer does not crash the process", () =>
1010 Effect.gen(function*() {
1011 yield* HttpRouter.add("GET", "/", HttpServerResponse.text("ok")).pipe(
1012 HttpRouter.serve,
1013 Layer.build
1014 )
1015 const server = yield* HttpServer.HttpServer
1016 const port = (server.address as NetAddress.InetAddress).port
1017
1018 const uncaught: Array<unknown> = []
1019 const onUncaught = (error: unknown) => uncaught.push(error)
1020 process.on("uncaughtException", onUncaught)
1021 yield* Effect.addFinalizer(() => Effect.sync(() => process.off("uncaughtException", onUncaught)))
1022
1023 yield* Effect.callback<void>((resume) => {
1024 const socket = Net.connect({ port, host: "127.0.0.1" }, () => {
1025 socket.write(
1026 "GET /ws HTTP/1.1\r\n" +
1027 "Host: 127.0.0.1\r\n" +
1028 "Upgrade: websocket\r\n" +
1029 "Connection: Upgrade\r\n" +
1030 "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" +
1031 "Sec-WebSocket-Version: 13\r\n\r\n"
1032 )
1033 setTimeout(() => {
1034 socket.resetAndDestroy()
1035 setTimeout(() => resume(Effect.void), 50)
1036 }, 50)
1037 })
1038 socket.on("error", () => {})
1039 })
1040
1041 expect(uncaught).toEqual([])
1042 const response = yield* HttpClient.get("/")
1043 expect(response.status).toEqual(200)
1044 }).pipe(Effect.provide(layerTestWebsocket)))
1045
1046 it.effect("fails refused websocket handshakes without hanging shutdown", () =>
1047 Effect.gen(function*() {
1048 const outcome = yield* Deferred.make<Socket.SocketError | undefined>()
1049 yield* HttpRouter.add(
1050 "GET",
1051 "/ws",
1052 Effect.gen(function*() {
1053 const request = yield* HttpServerRequest.HttpServerRequest
1054 const socket = yield* request.upgrade
1055 const error = yield* Effect.scoped(socket.reader).pipe(
1056 Effect.match({ onFailure: (error) => error, onSuccess: () => undefined })
1057 )
1058 yield* Deferred.succeed(outcome, error)
1059 return HttpServerResponse.empty()
1060 })
1061 ).pipe(
1062 HttpRouter.serve,
1063 Layer.build
1064 )
1065 const server = yield* HttpServer.HttpServer
1066 const port = (server.address as NetAddress.InetAddress).port
1067
1068 const status = yield* Effect.callback<number | undefined, Error>((resume) => {
1069 const req = Http.request({
1070 hostname: "127.0.0.1",
1071 port,
1072 agent: false,
1073 path: "/ws",
1074 method: "GET",
1075 headers: {
1076 Connection: "Upgrade",
1077 Upgrade: "websocket",
1078 "Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
1079 "Sec-WebSocket-Version": "12" // Unsupported version forces handshake rejection.
1080 }
1081 })
1082 req.on("response", (res) => {
1083 res.resume()
1084 res.on("end", () => resume(Effect.succeed(res.statusCode)))
1085 })
1086 req.on("error", (error) => resume(Effect.fail(error)))
1087 req.end()
1088 return Effect.sync(() => req.destroy())
1089 })
1090 assert.strictEqual(status, 400)
1091 assert.isTrue(Socket.SocketError.is(yield* Deferred.await(outcome)))
1092 }).pipe(Effect.provide(layerTestWebsocket), Effect.scoped))
1093
1094 it.effect("does not write the HTTP response to an upgraded connection", () =>
1095 Effect.gen(function*() {
1096 yield* HttpRouter.add(
1097 "GET",
1098 "/ws",
1099 Effect.gen(function*() {
1100 const request = yield* HttpServerRequest.HttpServerRequest
1101 const socket = yield* Effect.orDie(request.upgrade)
1102 yield* Effect.asVoid(socket.reader)
1103 const writer = yield* socket.writer
1104 yield* writer.write("refused")
1105 yield* writer.write(new Socket.CloseEvent(4400, "refused"))
1106 return HttpServerResponse.empty()
1107 }).pipe(Effect.scoped)
1108 ).pipe(
1109 HttpRouter.serve,
1110 Layer.build
1111 )
1112 const server = yield* HttpServer.HttpServer
1113 const port = (server.address as NetAddress.InetAddress).port
1114 const { frames, trailing } = yield* Effect.promise(() => rawWebSocket(port, "/ws"))
1115 assert.strictEqual(frames.length, 2)
1116 assert.strictEqual(frames[0].opcode, 1)
1117 assert.strictEqual(frames[0].payload.toString(), "refused")
1118 assert.strictEqual(frames[1].opcode, 8)
1119 assert.strictEqual(frames[1].payload.readUInt16BE(0), 4400)
1120 assert.strictEqual(trailing.toString(), "")
1121 }).pipe(Effect.provide(layerTestWebsocket)))
1122
1123 it.effect("writes the HTTP response when the upgrade is not consumed", () =>
1124 Effect.gen(function*() {
1125 yield* HttpRouter.add(
1126 "GET",
1127 "/no-ws",
1128 HttpServerResponse.text("upgrade refused", { status: 426 })
1129 ).pipe(
1130 HttpRouter.serve,
1131 Layer.build
1132 )
1133 const server = yield* HttpServer.HttpServer
1134 const port = (server.address as NetAddress.InetAddress).port
1135 const response = yield* Effect.promise(() => rawUpgradeRequest(port, "/no-ws"))
1136 assert.match(response, /^HTTP\/1\.1 426/)
1137 assert.match(response, /upgrade refused/)
1138 }).pipe(Effect.provide(layerTestWebsocket)))
1139})
1140
1141const layerTestWebsocket = HttpServer.layerTestClient.pipe(
1142 Layer.provide(
1143 Layer.fresh(FetchHttpClient.layer).pipe(
1144 Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false }))
1145 )
1146 ),
1147 Layer.provideMerge(NodeHttpServer.layer(Http.createServer, {
1148 port: 0,
1149 websocket: { perMessageDeflate: true }
1150 }))
1151)
1152
1153const getStatusText = (port: number) =>
1154 Effect.callback<string | undefined, Error>((resume) => {
1155 const request = Http.get({ hostname: "127.0.0.1", port, agent: false }, (response) => {
1156 response.resume()
1157 response.on("end", () => resume(Effect.succeed(response.statusMessage)))
1158 })
1159 request.on("error", (error) => resume(Effect.fail(error)))
1160 return Effect.sync(() => request.destroy())
1161 })
1162
1163const tcpPort = (server: Http.Server): number => {
1164 const address = server.address()
1165 assert(address !== null && typeof address !== "string")
1166 return address.port
1167}
1168
1169interface WebSocketFrame {
1170 readonly opcode: number
1171 readonly payload: Buffer.Buffer
1172}
1173
1174const parseWebSocketFrames = (
1175 stream: Buffer.Buffer
1176): { readonly frames: ReadonlyArray<WebSocketFrame>; readonly trailing: Buffer.Buffer } => {
1177 const frames: Array<WebSocketFrame> = []
1178 let offset = 0
1179 while (stream.length - offset >= 2) {
1180 const first = stream[offset]
1181 const second = stream[offset + 1]
1182 const opcode = first & 0x0f
1183 let length = second & 0x7f
1184 let headerSize = 2
1185 if (length === 126) {
1186 if (stream.length - offset < 4) break
1187 length = stream.readUInt16BE(offset + 2)
1188 headerSize = 4
1189 } else if (length === 127) {
1190 break
1191 }
1192 const controlFrame = (opcode & 0x8) !== 0
1193 if (
1194 (first & 0x70) !== 0 || (second & 0x80) !== 0 ||
1195 ![0x0, 0x1, 0x2, 0x8, 0x9, 0xa].includes(opcode) ||
1196 (controlFrame && length > 125)
1197 ) {
1198 break
1199 }
1200 if (stream.length - offset < headerSize + length) break
1201 frames.push({ opcode, payload: stream.subarray(offset + headerSize, offset + headerSize + length) })
1202 offset += headerSize + length
1203 }
1204 return { frames, trailing: stream.subarray(offset) }
1205}
1206
1207const upgradeRequest = (path: string): string =>
1208 [
1209 `GET ${path} HTTP/1.1`,
1210 "Host: 127.0.0.1",
1211 "Upgrade: websocket",
1212 "Connection: Upgrade",
1213 `Sec-WebSocket-Key: ${randomBytes(16).toString("base64")}`,
1214 "Sec-WebSocket-Version: 13",
1215 "",
1216 ""
1217 ].join("\r\n")
1218
1219const closeFrame = (code: number): Buffer.Buffer => {
1220 const payload = Buffer.Buffer.alloc(2)
1221 payload.writeUInt16BE(code, 0)
1222 const mask = randomBytes(4)
1223 const masked = Buffer.Buffer.from(payload)
1224 for (let index = 0; index < masked.length; index++) {
1225 masked[index] = masked[index] ^ mask[index % 4]
1226 }
1227 return Buffer.Buffer.concat([Buffer.Buffer.from([0x88, 0x80 | payload.length]), mask, masked])
1228}
1229
1230const rawWebSocket = (
1231 port: number,
1232 path: string
1233): Promise<{ readonly frames: ReadonlyArray<WebSocketFrame>; readonly trailing: Buffer.Buffer }> =>
1234 new Promise((resolve, reject) => {
1235 const socket = Net.createConnection({ host: "127.0.0.1", port })
1236 const timer = setTimeout(() => {
1237 socket.destroy()
1238 reject(new Error("the websocket conversation did not finish in time"))
1239 }, 5000)
1240 let upgraded = false
1241 let stream = Buffer.Buffer.alloc(0)
1242 let closeEchoed = false
1243 socket.on("connect", () => socket.write(upgradeRequest(path)))
1244 socket.on("data", (data) => {
1245 stream = Buffer.Buffer.concat([stream, Buffer.Buffer.from(data)])
1246 if (!upgraded) {
1247 const headerEnd = stream.indexOf("\r\n\r\n")
1248 if (headerEnd === -1) return
1249 if (!stream.subarray(0, headerEnd).toString().includes("101")) {
1250 clearTimeout(timer)
1251 socket.destroy()
1252 reject(new Error("the upgrade was refused"))
1253 return
1254 }
1255 upgraded = true
1256 stream = Buffer.Buffer.from(stream.subarray(headerEnd + 4))
1257 }
1258 if (!closeEchoed) {
1259 const close = parseWebSocketFrames(stream).frames.find((frame) => frame.opcode === 8)
1260 if (close !== undefined) {
1261 closeEchoed = true
1262 socket.write(closeFrame(close.payload.length >= 2 ? close.payload.readUInt16BE(0) : 1000))
1263 }
1264 }
1265 })
1266 socket.on("error", constVoid)
1267 socket.on("close", () => {
1268 clearTimeout(timer)
1269 resolve(parseWebSocketFrames(stream))
1270 })
1271 })
1272
1273const rawUpgradeRequest = (port: number, path: string): Promise<string> =>
1274 new Promise((resolve, reject) => {
1275 const socket = Net.createConnection({ host: "127.0.0.1", port })
1276 const timer = setTimeout(() => {
1277 socket.destroy()
1278 reject(new Error("the upgrade request was not answered in time"))
1279 }, 5000)
1280 let stream = Buffer.Buffer.alloc(0)
1281 socket.on("connect", () => socket.write(upgradeRequest(path)))
1282 socket.on("data", (data) => {
1283 stream = Buffer.Buffer.concat([stream, Buffer.Buffer.from(data)])
1284 })
1285 socket.on("error", constVoid)
1286 socket.on("close", () => {
1287 clearTimeout(timer)
1288 resolve(stream.toString())
1289 })
1290 })