1/**
2 * Bun implementation of the Effect `HttpServer`.
3 *
4 * `make` creates a scoped HTTP server from `Bun.serve`, converting Bun
5 * `Request` values into `HttpServerRequest` values and Effect
6 * `HttpServerResponse` values back into Web `Response` values. The server
7 * supports streaming bodies, multipart requests, file responses through
8 * `BunHttpPlatform`, and WebSocket upgrades. This module also provides layers
9 * for the server alone, the Bun HTTP support services, the combined server,
10 * configurable server options, and a test server with an HTTP client.
11 *
12 * @since 4.0.0
13 */
14import type { Server as BunServer, ServerWebSocket } from "bun"
15import type * as Arr from "effect/Array"
16import * as Cause from "effect/Cause"
17import * as Config from "effect/Config"
18import type { ConfigError } from "effect/Config"
19import * as Context from "effect/Context"
20import * as Deferred from "effect/Deferred"
21import * as Duration from "effect/Duration"
22import * as Effect from "effect/Effect"
23import * as Exit from "effect/Exit"
24import * as Fiber from "effect/Fiber"
25import type * as FileSystem from "effect/FileSystem"
26import { constVoid, flow } from "effect/Function"
27import * as Cookies from "effect/http/Cookies"
28import * as Etag from "effect/http/Etag"
29import * as FetchHttpClient from "effect/http/FetchHttpClient"
30import * as Headers from "effect/http/Headers"
31import type { HttpClient } from "effect/http/HttpClient"
32import * as HttpEffect from "effect/http/HttpEffect"
33import * as IncomingMessage from "effect/http/HttpIncomingMessage"
34import type { HttpMethod } from "effect/http/HttpMethod"
35import type { HttpPlatform } from "effect/http/HttpPlatform"
36import * as Server from "effect/http/HttpServer"
37import * as Error from "effect/http/HttpServerError"
38import * as ServerRequest from "effect/http/HttpServerRequest"
39import * as ServerResponse from "effect/http/HttpServerResponse"
40import type * as Multipart from "effect/http/Multipart"
41import * as UrlParams from "effect/http/UrlParams"
42import * as Inspectable from "effect/Inspectable"
43import * as Layer from "effect/Layer"
44import * as NetAddress from "effect/net/NetAddress"
45import * as Option from "effect/Option"
46import type * as Path from "effect/Path"
47import type * as Record from "effect/Record"
48import * as Result from "effect/Result"
49import * as Scheduler from "effect/Scheduler"
50import type * as Schema from "effect/Schema"
51import * as Scope from "effect/Scope"
52import * as Semaphore from "effect/Semaphore"
53import * as Socket from "effect/socket/Socket"
54import * as Stream from "effect/Stream"
55import * as Platform from "./BunHttpPlatform.ts"
56import * as BunMultipart from "./BunMultipart.ts"
57import * as BunServices from "./BunServices.ts"
58import * as BunStream from "./BunStream.ts"
59
60/**
61 * Bun serve options accepted by the HTTP server, extended with typed route definitions.
62 *
63 * @category options
64 * @since 4.0.0
65 */
66export type ServeOptions<R extends string> =
67 & (
68 | Bun.Serve.UnixServeOptions<WebSocketContext>
69 | Bun.Serve.HostnamePortServeOptions<WebSocketContext>
70 )
71 & { readonly routes?: Bun.Serve.Routes<WebSocketContext, R> }
72
73/**
74 * WebSocket tuning options forwarded to `Bun.serve`'s `websocket` handler.
75 *
76 * **Details**
77 *
78 * The lifecycle handlers (`open`, `message`, `close`, ...) are managed by the
79 * server and cannot be overridden; everything else — such as
80 * `perMessageDeflate` compression, payload limits, and idle timeouts — passes
81 * through, e.g.
82 * `BunHttpServer.layer({ port: 3000, websocket: { perMessageDeflate: true } })`.
83 *
84 * The `compressionThreshold` option controls the minimum message size in bytes
85 * that is compressed when per-message deflate is negotiated. It defaults to
86 * 1024, matching the default threshold of Node's `ws` server.
87 *
88 * @category options
89 * @since 4.0.0
90 */
91export type WebSocketOptions =
92 & Omit<
93 Bun.WebSocketHandler<WebSocketContext>,
94 "open" | "message" | "close" | "drain" | "ping" | "pong" | "data" | "binaryType"
95 >
96 & {
97 /**
98 * The minimum message size in bytes that is compressed when per-message
99 * deflate is negotiated.
100 *
101 * @default 1024
102 */
103 readonly compressionThreshold?: number | undefined
104 }
105
106/**
107 * Creates a scoped Bun `HttpServer` from `Bun.serve` options, stopping the server on scope finalization with optional graceful shutdown settings.
108 *
109 * @category constructors
110 * @since 4.0.0
111 */
112export const make = Effect.fnUntraced(
113 function*<R extends string>(
114 options: ServeOptions<R> & {
115 readonly disablePreemptiveShutdown?: boolean | undefined
116 readonly gracefulShutdownTimeout?: Duration.Input | undefined
117 readonly websocket?: WebSocketOptions | undefined
118 }
119 ) {
120 const scope = yield* Effect.scope
121 let listenOptions = options
122 if (!("unix" in options) || options.unix === undefined) {
123 const internetOptions = options as Bun.Serve.HostnamePortServeOptions<WebSocketContext>
124 let hostname = internetOptions.hostname ?? "::"
125 if (Result.isFailure(NetAddress.ipFromString(hostname))) {
126 hostname = yield* Effect.tryPromise({
127 try: async () => {
128 const result = await Bun.dns.lookup(hostname, { socketType: "tcp" })
129 if (result.length === 0) throw new globalThis.Error(`Could not resolve hostname: ${hostname}`)
130 return result[0].address
131 },
132 catch: (cause) => new Error.ServeError({ cause })
133 })
134 }
135 listenOptions = { ...options, hostname }
136 }
137 const { compressionThreshold = MIN_COMPRESSIBLE_SIZE, ...websocket } = options.websocket ?? {}
138 const handlerStack: Array<(request: Request, server: BunServer<WebSocketContext>) => Response | Promise<Response>> =
139 [
140 function(_request, _server) {
141 return new Response("not found", { status: 404 })
142 }
143 ]
144 const server = Bun.serve<WebSocketContext, R>({
145 ...listenOptions as ServeOptions<R>,
146 fetch: handlerStack[0],
147 websocket: {
148 ...websocket,
149 open(ws) {
150 Deferred.doneUnsafe(ws.data.deferred, Exit.succeed(ws))
151 },
152 message(ws, message) {
153 ws.data.run(message)
154 },
155 close(ws, code, closeReason) {
156 code = typeof code === "number" ? code : 1001
157 const error = new Socket.SocketError({
158 reason: new Socket.SocketCloseError({ code, closeReason })
159 })
160 ws.data.closeError = error
161 ws.data.onClose(error)
162 }
163 }
164 })
165
166 const shutdown = yield* Effect.promise(() => server.stop()).pipe(
167 Effect.cached
168 )
169 const preemptiveShutdown = options.disablePreemptiveShutdown ? Effect.void : Effect.timeoutOrElse(shutdown, {
170 duration: options.gracefulShutdownTimeout ?? Duration.seconds(20),
171 orElse: () => Effect.void
172 })
173
174 yield* Scope.addFinalizer(scope, shutdown)
175
176 const address = "unix" in options && options.unix !== undefined
177 ? NetAddress.unixPathAddress(options.unix)
178 : yield* Effect.fromResult(NetAddress.inetAddressFromIpString(server.hostname!, server.port!)).pipe(
179 Effect.mapError((cause) => new Error.ServeError({ cause }))
180 )
181
182 return Server.make({
183 address,
184 serve: Effect.fnUntraced(function*(httpApp, middleware) {
185 const parent = yield* Effect.fiber
186 const services = parent.context
187 const serveScope = Context.getUnsafe(services, Scope.Scope)
188 const scope = Scope.forkUnsafe(serveScope, "parallel")
189
190 const httpEffect = HttpEffect.toHandled(httpApp, (request, response) =>
191 Effect.sync(() => {
192 ;(request as BunServerRequest).resolve(makeResponse(request, response, services, scope))
193 }), middleware)
194
195 function handler(request: Request, server: BunServer<WebSocketContext>) {
196 return new Promise<Response>((resolve, _reject) => {
197 const context = Context.add(
198 services,
199 ServerRequest.HttpServerRequest,
200 new BunServerRequest(request, resolve, removeHost(request.url), server, compressionThreshold)
201 )
202 const fiber = Fiber.runIn(Effect.runForkWith(context)(httpEffect), scope)
203 request.signal.addEventListener("abort", () => {
204 fiber.interruptUnsafe(parent.id, Error.ClientAbort.annotation)
205 }, { once: true })
206 })
207 }
208
209 yield* Scope.addFinalizerExit(serveScope, () => {
210 const index = handlerStack.indexOf(handler)
211 if (index !== -1) handlerStack.splice(index, 1)
212 server.reload({
213 fetch: handlerStack[handlerStack.length - 1],
214 ...(options.routes === undefined ? undefined : { routes: options.routes })
215 })
216 return handlerStack.length === 1 ? preemptiveShutdown : Effect.void
217 })
218 handlerStack.push(handler)
219 server.reload({
220 fetch: handler,
221 ...(options.routes === undefined ? undefined : { routes: options.routes })
222 })
223 })
224 })
225 }
226)
227
228const MIN_COMPRESSIBLE_SIZE = 1024
229
230const makeResponse = (
231 request: ServerRequest.HttpServerRequest,
232 response: ServerResponse.HttpServerResponse,
233 context: Context.Context<never>,
234 scope: Scope.Scope
235): Response => {
236 if (ServerResponse.omitsBody(response, request.method === "HEAD")) {
237 return ServerResponse.toWeb(response, { withoutBody: true })
238 }
239 const fields: {
240 headers: globalThis.Headers
241 status?: number
242 statusText?: string
243 } = {
244 headers: new globalThis.Headers(response.headers),
245 status: response.status
246 }
247
248 if (!Cookies.isEmpty(response.cookies)) {
249 for (const header of Cookies.toSetCookieHeaders(response.cookies)) {
250 fields.headers.append("set-cookie", header)
251 }
252 }
253
254 if (response.statusText !== undefined) {
255 fields.statusText = response.statusText
256 }
257
258 response = HttpEffect.scopeTransferToStream(response)
259 const body = response.body
260 switch (body._tag) {
261 case "Empty": {
262 return new Response(undefined, fields)
263 }
264 case "Uint8Array": {
265 return new Response(body.text ?? body.body as any, fields)
266 }
267 case "Raw": {
268 if (body.body instanceof Response) {
269 for (const [key, value] of fields.headers.entries()) {
270 body.body.headers.set(key, value)
271 }
272 return body.body
273 }
274 return new Response(body.body as any, fields)
275 }
276 case "FormData": {
277 return new Response(body.formData as any, fields)
278 }
279 case "Stream": {
280 return new Response(
281 Stream.toReadableStreamWith(
282 Stream.unwrap(Effect.withFiber((fiber) => {
283 Fiber.runIn(fiber, scope)
284 return Effect.succeed(body.stream)
285 })),
286 context
287 ),
288 fields
289 )
290 }
291 }
292}
293
294/**
295 * Layer that provides only `HttpServer` by constructing a scoped Bun server from the supplied serve options.
296 *
297 * @category layers
298 * @since 4.0.0
299 */
300export const layerServer: <R extends string>(
301 options: ServeOptions<R> & {
302 readonly disablePreemptiveShutdown?: boolean | undefined
303 readonly gracefulShutdownTimeout?: Duration.Input | undefined
304 readonly websocket?: WebSocketOptions | undefined
305 }
306) => Layer.Layer<Server.HttpServer, Error.ServeError> = flow(make, Layer.effect(Server.HttpServer)) as any
307
308/**
309 * Layer that provides Bun HTTP support services: `HttpPlatform`, weak ETag generation, and `BunServices`.
310 *
311 * @category layers
312 * @since 4.0.0
313 */
314export const layerHttpServices: Layer.Layer<
315 | HttpPlatform
316 | Etag.Generator
317 | BunServices.BunServices
318> = Layer.mergeAll(
319 Platform.layer,
320 Etag.layerWeak,
321 BunServices.layer
322)
323
324/**
325 * Layer that provides a Bun `HttpServer` together with the Bun HTTP platform, ETag generator, and Bun services.
326 *
327 * @category layers
328 * @since 4.0.0
329 */
330export const layer = <R extends string>(
331 options: ServeOptions<R> & {
332 readonly disablePreemptiveShutdown?: boolean | undefined
333 readonly gracefulShutdownTimeout?: Duration.Input | undefined
334 readonly websocket?: WebSocketOptions | undefined
335 }
336): Layer.Layer<
337 | Server.HttpServer
338 | HttpPlatform
339 | Etag.Generator
340 | BunServices.BunServices,
341 Error.ServeError
342> => Layer.mergeAll(layerServer(options), layerHttpServices)
343
344/**
345 * Layer that starts a Bun HTTP server on an ephemeral port for tests.
346 *
347 * @category testing
348 * @since 4.0.0
349 */
350export const layerTest: Layer.Layer<
351 Server.HttpServer | HttpPlatform | FileSystem.FileSystem | Etag.Generator | Path.Path | HttpClient
352> = Server.layerTestClient.pipe(
353 Layer.provide(FetchHttpClient.layer.pipe(
354 Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false }))
355 )),
356 Layer.provideMerge(Layer.orDie(layer({ hostname: "127.0.0.1", port: 0 })))
357)
358
359/**
360 * Creates the Bun HTTP server and support-services layer from configurable serve options.
361 *
362 * @category layers
363 * @since 4.0.0
364 */
365export const layerConfig = <R extends string>(
366 options: Config.Wrap<
367 ServeOptions<R> & {
368 readonly disablePreemptiveShutdown?: boolean | undefined
369 readonly gracefulShutdownTimeout?: Duration.Input | undefined
370 readonly websocket?: WebSocketOptions | undefined
371 }
372 >
373): Layer.Layer<
374 Server.HttpServer | HttpPlatform | FileSystem.FileSystem | Etag.Generator | Path.Path,
375 ConfigError | Error.ServeError
376> =>
377 Layer.mergeAll(
378 Layer.effect(Server.HttpServer)(Effect.flatMap(Config.unwrap(options), make)),
379 layerHttpServices
380 )
381
382// -----------------------------------------------------------------------------
383// Internal
384// -----------------------------------------------------------------------------
385
386interface WebSocketContext {
387 readonly deferred: Deferred.Deferred<ServerWebSocket<WebSocketContext>>
388 readonly buffer: Array<Uint8Array | string>
389 closeError: Socket.SocketError | undefined
390 run: (_: Uint8Array | string) => void
391 onClose: (error: Socket.SocketError) => void
392}
393
394function wsDefaultRun(this: WebSocketContext, _: Uint8Array | string) {
395 this.buffer.push(_)
396}
397
398class BunServerRequest extends Inspectable.Class implements ServerRequest.HttpServerRequest {
399 readonly [ServerRequest.TypeId]: typeof ServerRequest.TypeId
400 readonly [IncomingMessage.TypeId]: typeof IncomingMessage.TypeId
401 readonly source: Request
402 public resolve: (response: Response) => void
403 readonly url: string
404 private bunServer: BunServer<WebSocketContext>
405 private compressionThreshold: number
406 public headersOverride?: Headers.Headers | undefined
407 private remoteAddressOverride?: Option.Option<string> | undefined
408
409 constructor(
410 source: Request,
411 resolve: (response: Response) => void,
412 url: string,
413 bunServer: BunServer<WebSocketContext>,
414 compressionThreshold: number,
415 headersOverride?: Headers.Headers,
416 remoteAddressOverride?: Option.Option<string>
417 ) {
418 super()
419 this[ServerRequest.TypeId] = ServerRequest.TypeId
420 this[IncomingMessage.TypeId] = IncomingMessage.TypeId
421 this.source = source
422 this.resolve = resolve
423 this.url = url
424 this.bunServer = bunServer
425 this.compressionThreshold = compressionThreshold
426 this.headersOverride = headersOverride
427 this.remoteAddressOverride = remoteAddressOverride
428 }
429 toJSON(): unknown {
430 return IncomingMessage.inspect(this, {
431 _id: "HttpServerRequest",
432 method: this.method,
433 url: this.originalUrl
434 })
435 }
436 modify(
437 options: {
438 readonly url?: string | undefined
439 readonly headers?: Headers.Headers | undefined
440 readonly remoteAddress?: Option.Option<string> | undefined
441 }
442 ) {
443 return new BunServerRequest(
444 this.source,
445 this.resolve,
446 options.url ?? this.url,
447 this.bunServer,
448 this.compressionThreshold,
449 options.headers ?? this.headersOverride,
450 "remoteAddress" in options ? options.remoteAddress : this.remoteAddressOverride
451 )
452 }
453 get method(): HttpMethod {
454 return this.source.method.toUpperCase() as HttpMethod
455 }
456 get originalUrl() {
457 return this.source.url
458 }
459 get remoteAddress(): Option.Option<string> {
460 return this.remoteAddressOverride ?? Option.fromNullishOr(this.bunServer.requestIP(this.source)?.address)
461 }
462 get headers(): Headers.Headers {
463 this.headersOverride ??= Headers.fromInput(this.source.headers)
464 return this.headersOverride
465 }
466
467 private cachedCookies: Record.ReadonlyRecord<string, string> | undefined
468 get cookies() {
469 if (this.cachedCookies) {
470 return this.cachedCookies
471 }
472 return this.cachedCookies = Cookies.parseHeader(this.headers.cookie ?? "")
473 }
474
475 get stream(): Stream.Stream<Uint8Array, Error.HttpServerError> {
476 return this.source.body
477 ? BunStream.fromReadableStream({
478 evaluate: () => this.source.body ?? emptyReadbleStream,
479 onError: (cause) =>
480 new Error.HttpServerError({
481 reason: new Error.RequestParseError({
482 request: this,
483 cause
484 })
485 })
486 })
487 : Stream.fail(
488 new Error.HttpServerError({
489 reason: new Error.RequestParseError({
490 request: this,
491 description: "can not create stream from empty body"
492 })
493 })
494 )
495 }
496
497 private textEffect: Effect.Effect<string, Error.HttpServerError> | undefined
498 get text(): Effect.Effect<string, Error.HttpServerError> {
499 if (this.textEffect) {
500 return this.textEffect
501 }
502 this.textEffect = Effect.runSync(Effect.cached(
503 Effect.tryPromise({
504 try: () => this.source.text(),
505 catch: (cause) =>
506 new Error.HttpServerError({
507 reason: new Error.RequestParseError({
508 request: this,
509 cause
510 })
511 })
512 })
513 ))
514 return this.textEffect
515 }
516
517 get json(): Effect.Effect<Schema.Json, Error.HttpServerError> {
518 return Effect.flatMap(this.text, (_) =>
519 Effect.try({
520 try: () => JSON.parse(_) as Schema.Json,
521 catch: (cause) =>
522 new Error.HttpServerError({
523 reason: new Error.RequestParseError({
524 request: this,
525 cause
526 })
527 })
528 }))
529 }
530
531 get urlParamsBody(): Effect.Effect<UrlParams.UrlParams, Error.HttpServerError> {
532 return Effect.flatMap(this.text, (_) =>
533 Effect.try({
534 try: () => UrlParams.fromInput(new URLSearchParams(_)),
535 catch: (cause) =>
536 new Error.HttpServerError({
537 reason: new Error.RequestParseError({
538 request: this,
539 cause
540 })
541 })
542 }))
543 }
544
545 private multipartEffect:
546 | Effect.Effect<
547 Multipart.Persisted,
548 Multipart.MultipartError,
549 Scope.Scope | FileSystem.FileSystem | Path.Path
550 >
551 | undefined
552 get multipart(): Effect.Effect<
553 Multipart.Persisted,
554 Multipart.MultipartError,
555 Scope.Scope | FileSystem.FileSystem | Path.Path
556 > {
557 if (this.multipartEffect) {
558 return this.multipartEffect
559 }
560 this.multipartEffect = Effect.runSync(Effect.cached(
561 BunMultipart.persisted(this.source)
562 ))
563 return this.multipartEffect
564 }
565
566 get multipartStream(): Stream.Stream<Multipart.Part, Multipart.MultipartError> {
567 return BunMultipart.stream(this.source)
568 }
569
570 private arrayBufferEffect: Effect.Effect<ArrayBuffer, Error.HttpServerError> | undefined
571 get arrayBuffer(): Effect.Effect<ArrayBuffer, Error.HttpServerError> {
572 if (this.arrayBufferEffect) {
573 return this.arrayBufferEffect
574 }
575 this.arrayBufferEffect = Effect.runSync(Effect.cached(
576 Effect.tryPromise({
577 try: () => this.source.arrayBuffer(),
578 catch: (cause) =>
579 new Error.HttpServerError({
580 reason: new Error.RequestParseError({
581 request: this,
582 cause
583 })
584 })
585 })
586 ))
587 this.textEffect = Effect.map(this.arrayBufferEffect, (_) => new TextDecoder().decode(_))
588 return this.arrayBufferEffect
589 }
590
591 get upgrade(): Effect.Effect<Socket.Socket, Error.HttpServerError> {
592 return Effect.callback<Socket.Socket, Error.HttpServerError>((resume) => {
593 const deferred = Deferred.makeUnsafe<ServerWebSocket<WebSocketContext>>()
594 const semaphore = Semaphore.makeUnsafe(1)
595
596 const success = this.bunServer.upgrade(this.source, {
597 data: {
598 deferred,
599 buffer: [],
600 closeError: undefined,
601 run: wsDefaultRun,
602 onClose: constVoid
603 }
604 })
605 if (!success) {
606 resume(Effect.fail(
607 new Error.HttpServerError({
608 reason: new Error.RequestParseError({
609 request: this,
610 description: "Not an upgradeable ServerRequest"
611 })
612 })
613 ))
614 return
615 }
616 const compressionThreshold = this.compressionThreshold
617 resume(Effect.map(Deferred.await(deferred), (ws) => {
618 const write = (chunk: Uint8Array | string | Socket.CloseEvent) =>
619 Effect.sync(() => {
620 if (typeof chunk === "string") {
621 ws.sendText(chunk, chunk.length >= compressionThreshold)
622 } else if (Socket.isCloseEvent(chunk)) {
623 ws.close(chunk.code, chunk.reason)
624 } else {
625 ws.sendBinary(chunk, chunk.byteLength >= compressionThreshold)
626 }
627 })
628 const writeAll = (chunks: ReadonlyArray<Uint8Array | string>) =>
629 Effect.sync(() => {
630 for (let i = 0; i < chunks.length; i++) {
631 const chunk = chunks[i]
632 if (typeof chunk === "string") {
633 ws.sendText(chunk, chunk.length >= compressionThreshold)
634 } else {
635 ws.sendBinary(chunk, chunk.byteLength >= compressionThreshold)
636 }
637 }
638 })
639 const writer: Socket.Socket["writer"] = Effect.succeed({ write, writeAll })
640
641 const reader: Socket.Socket["reader"] = Effect.gen(function*() {
642 const dispatcher = (yield* Scheduler.Scheduler).makeDispatcher()
643 yield* Effect.acquireRelease(semaphore.take(1), () => semaphore.release(1))
644 const closeError = ws.data.closeError ?? (ws.readyState >= 2
645 ? new Socket.SocketError({
646 reason: new Socket.SocketCloseError({ code: 1006 })
647 })
648 : undefined)
649 if (closeError !== undefined && ws.data.buffer.length === 0) {
650 return yield* closeError
651 }
652 const scope = yield* Effect.scope
653
654 type ReadResume = (
655 effect: Effect.Effect<Arr.NonEmptyReadonlyArray<Uint8Array | string>, Socket.SocketError>
656 ) => void
657
658 let buffer: Array<Uint8Array | string> = ws.data.buffer.splice(0)
659 let error: Socket.SocketError | undefined = closeError
660 let waiter: ReadResume | undefined
661 let flushScheduled = false
662
663 function takeBuffer(): Arr.NonEmptyReadonlyArray<Uint8Array | string> {
664 const chunk = buffer
665 buffer = []
666 return chunk as unknown as Arr.NonEmptyReadonlyArray<Uint8Array | string>
667 }
668 function deliver() {
669 flushScheduled = false
670 if (waiter === undefined || buffer.length === 0) return
671 const resumeRead = waiter
672 waiter = undefined
673 resumeRead(Effect.succeed(takeBuffer()))
674 }
675 function push(data: Uint8Array | string) {
676 buffer.push(data)
677 if (waiter !== undefined && !flushScheduled) {
678 flushScheduled = true
679 dispatcher.scheduleTask(deliver, 0)
680 }
681 }
682 function fail(err: Socket.SocketError) {
683 if (error === undefined) error = err
684 if (waiter !== undefined) {
685 const resumeRead = waiter
686 waiter = undefined
687 resumeRead(buffer.length > 0 ? Effect.succeed(takeBuffer()) : Effect.fail(error))
688 }
689 }
690
691 ws.data.run = push
692 ws.data.onClose = fail
693 yield* Scope.addFinalizerExit(
694 scope,
695 (exit) =>
696 Effect.suspend(() => {
697 // resume a pull blocked in another fiber before detaching
698 fail(
699 new Socket.SocketError({
700 reason: new Socket.SocketCloseError({ code: 1006 })
701 })
702 )
703 ws.data.run = wsDefaultRun
704 ws.data.onClose = constVoid
705 ws.close(closeCode(exit))
706 return Effect.void
707 })
708 )
709
710 return {
711 pull: Effect.callback<
712 Arr.NonEmptyReadonlyArray<Uint8Array | string>,
713 Socket.SocketError
714 >((resumeRead) => {
715 if (buffer.length > 0) return resumeRead(Effect.succeed(takeBuffer()))
716 if (error !== undefined) return resumeRead(Effect.fail(error))
717 waiter = resumeRead
718 return Effect.sync(() => {
719 if (waiter === resumeRead) waiter = undefined
720 })
721 }),
722 upgrade: Socket.SocketUpgradeError.unsupported
723 }
724 })
725
726 return Socket.make({ reader, writer })
727 }))
728 })
729 }
730}
731
732const closeCode = (exit: Exit.Exit<unknown, unknown>): number =>
733 Exit.isSuccess(exit) ? 1000 : Cause.hasInterruptsOnly(exit.cause) ? 1001 : 1011
734
735const emptyReadbleStream = new ReadableStream({
736 start(controller) {
737 controller.enqueue(new Uint8Array())
738 controller.close()
739 }
740})
741
742const removeHost = (url: string) => {
743 if (url[0] === "/") {
744 return url
745 }
746 const index = url.indexOf("/", url.indexOf("//") + 2)
747 return index === -1 ? "/" : url.slice(index)
748}