1/**
2 * Models bidirectional socket connections in Effect.
3 *
4 * A `Socket` exposes a pull-based `reader` and a scoped `writer`. Acquiring
5 * the reader dials the connection and returns an `Effect` that yields batches
6 * of incoming frames with end-to-end backpressure: nothing is read from the
7 * transport until the consumer pulls. Every termination, including clean
8 * closes, surfaces as a `SocketError`, so reconnecting is a plain
9 * `Effect.retry` around the scoped consume loop.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import type { NonEmptyArray, NonEmptyReadonlyArray } from "../Array.ts"
15import type * as Cause from "../Cause.ts"
16import * as Channel from "../Channel.ts"
17import * as Context from "../Context.ts"
18import type * as Duration from "../Duration.ts"
19import * as Effect from "../Effect.ts"
20import * as Exit from "../Exit.ts"
21import { constVoid, dual, flow } from "../Function.ts"
22import * as Latch from "../Latch.ts"
23import * as Layer from "../Layer.ts"
24import * as Predicate from "../Predicate.ts"
25import * as Pull from "../Pull.ts"
26import type * as Redacted from "../Redacted.ts"
27import * as Schema from "../Schema.ts"
28import * as Scope from "../Scope.ts"
29import * as Stream from "../Stream.ts"
30
31/**
32 * Runtime type identifier attached to `Socket` services.
33 *
34 * @stability unstable
35 * @category type IDs
36 * @since 4.0.0
37 */
38export const TypeId = "~effect/socket/Socket"
39
40/**
41 * Returns `true` when a value is a `Socket`.
42 *
43 * @stability unstable
44 * @category guards
45 * @since 4.0.0
46 */
47export const isSocket = (u: unknown): u is Socket => Predicate.hasProperty(u, TypeId)
48
49/**
50 * Service tag for bidirectional socket transports.
51 *
52 * **When to use**
53 *
54 * Use to access or provide the socket implementation used by programs that
55 * read and write frames through the Effect environment.
56 *
57 * @stability unstable
58 * @category services
59 * @since 4.0.0
60 */
61export const Socket: Context.Service<Socket, Socket> = Context.Service<Socket>("effect/socket/Socket")
62
63/**
64 * Effect-based socket abstraction exposing a pull-based read side and a
65 * scoped writer.
66 *
67 * **Details**
68 *
69 * Acquiring `reader` establishes the connection; the scope of the acquisition
70 * owns the connection lifecycle. Its `pull` yields non-empty batches of incoming
71 * frames and never completes via `Cause.Done`: every termination, clean close
72 * included, fails with a `SocketError` wrapping the close reason. Code placed
73 * between the acquisition and the first pull runs exactly once per
74 * (re)connection, which makes handshakes plain code placement.
75 *
76 * Closing the acquisition scope must fail a pull that is currently suspended,
77 * rather than leaving it blocked. Consumers such as `toChannel` rely on this
78 * to shut a connection down without racing every pull against a separate
79 * failure signal.
80 *
81 * The writer is detached from any single connection: acquiring it cannot
82 * fail, and writes made while disconnected suspend until the next connection
83 * is established. Releasing the writer scope half-closes the write side
84 * where the transport supports it.
85 *
86 * **Example** (Consuming with automatic reconnect)
87 *
88 * ```ts skip-type-checking
89 * Effect.gen(function*() {
90 * const { pull } = yield* socket.reader
91 * while (true) {
92 * yield* handle(yield* pull)
93 * }
94 * }).pipe(
95 * Effect.scoped,
96 * Effect.retry({ schedule: Schedule.exponential(200) })
97 * )
98 * ```
99 *
100 * @stability unstable
101 * @category models
102 * @since 4.0.0
103 */
104export interface Socket {
105 readonly [TypeId]: typeof TypeId
106 readonly reader: Effect.Effect<Reader, SocketError, Scope.Scope>
107 readonly writer: Effect.Effect<Writer, never, Scope.Scope>
108}
109
110/**
111 * The read side of a live `Socket` connection.
112 *
113 * **Details**
114 *
115 * `pull` reads the next non-empty batch. `upgrade` wraps this connection with
116 * TLS when the transport supports it. Unsupported readers fail with a
117 * `SocketUpgradeError`. The upgrade also fails with `SocketUpgradeError` when
118 * the selected TLS role requires an identity but `key` and `cert` are not both
119 * provided. Calling `upgrade()` without an options object uses the adapter's
120 * client defaults.
121 *
122 * @stability unstable
123 * @category models
124 * @since 4.0.0
125 */
126export interface Reader<A extends Uint8Array | string = Uint8Array | string> {
127 readonly pull: Effect.Effect<NonEmptyReadonlyArray<A>, SocketError>
128 readonly upgrade: (options?: TlsUpgradeOptions) => Effect.Effect<void, SocketError>
129}
130
131/**
132 * The write side of a `Socket`.
133 *
134 * **Details**
135 *
136 * `write` sends a single frame or a `CloseEvent`; `writeAll` sends a batch of
137 * frames, allowing transports to coalesce them into a single flush. Both
138 * apply the transport's native backpressure before succeeding.
139 *
140 * @stability unstable
141 * @category models
142 * @since 4.0.0
143 */
144export interface Writer {
145 readonly write: (chunk: Uint8Array | string | CloseEvent) => Effect.Effect<void, SocketError>
146 readonly writeAll: (chunks: NonEmptyReadonlyArray<Uint8Array | string>) => Effect.Effect<void, SocketError>
147}
148
149/**
150 * TLS credentials and handshake settings used to upgrade a live socket.
151 *
152 * **Details**
153 *
154 * `key` and `cert` are optional for client upgrades that do not present a
155 * client certificate. Server upgrades require both, and providing only one is
156 * invalid. Missing or incomplete credentials fail the upgrade with a
157 * `SocketUpgradeError` when the adapter needs an identity.
158 *
159 * @stability unstable
160 * @category models
161 * @since 4.0.0
162 */
163export interface TlsUpgradeOptions {
164 readonly key?:
165 | Redacted.Redacted<string | Uint8Array>
166 | ReadonlyArray<Redacted.Redacted<string | Uint8Array>>
167 | undefined
168 readonly cert?: string | Uint8Array | ReadonlyArray<string | Uint8Array> | undefined
169 readonly ca?: string | Uint8Array | ReadonlyArray<string | Uint8Array> | undefined
170 readonly passphrase?: Redacted.Redacted<string> | undefined
171 readonly alpnProtocols?: ReadonlyArray<string> | undefined
172 readonly requestCert?: boolean | undefined
173 readonly rejectUnauthorized?: boolean | undefined
174}
175
176/**
177 * Constructs a `Socket` from a reader acquisition and a scoped writer.
178 *
179 * **Details**
180 *
181 * The reader must fail a suspended pull when its acquisition scope closes; see
182 * `Socket` for why. A reader that leaves a pull blocked forever will hang any
183 * consumer that shuts the socket down by closing that scope.
184 *
185 * @stability unstable
186 * @category constructors
187 * @since 4.0.0
188 */
189export const make = (options: {
190 readonly reader: Socket["reader"]
191 readonly writer: Socket["writer"]
192}): Socket =>
193 Socket.of({
194 [TypeId]: TypeId,
195 reader: options.reader,
196 writer: options.writer
197 })
198
199const encoder = new TextEncoder()
200
201/**
202 * Acquires the socket's binary `pull`, encoding any string frames as UTF-8
203 * bytes.
204 *
205 * **Details**
206 *
207 * When a pulled batch contains no string frames it is returned as-is, so
208 * transports that only emit bytes (TCP) pay no per-chunk cost.
209 *
210 * @stability unstable
211 * @category combinators
212 * @since 4.0.0
213 */
214export const readerBytes = (
215 self: Socket
216): Effect.Effect<
217 Effect.Effect<NonEmptyReadonlyArray<Uint8Array>, SocketError>,
218 SocketError,
219 Scope.Scope
220> =>
221 Effect.map(self.reader, ({ pull }) =>
222 Effect.map(pull, (chunk) => {
223 for (let i = 0; i < chunk.length; i++) {
224 if (typeof chunk[i] === "string") {
225 const out = new Array<Uint8Array>(chunk.length) as NonEmptyArray<Uint8Array>
226 for (let j = 0; j < chunk.length; j++) {
227 const item = chunk[j]
228 out[j] = typeof item === "string" ? encoder.encode(item) : item
229 }
230 return out
231 }
232 }
233 return chunk as NonEmptyReadonlyArray<Uint8Array>
234 }))
235
236/**
237 * Acquires the socket's string `pull`, decoding binary frames with the optional
238 * text encoding.
239 *
240 * **Details**
241 *
242 * The `TextDecoder` is created once per acquisition.
243 *
244 * @stability unstable
245 * @category combinators
246 * @since 4.0.0
247 */
248export const readerString = (
249 self: Socket,
250 encoding?: string | undefined
251): Effect.Effect<
252 Effect.Effect<NonEmptyReadonlyArray<string>, SocketError>,
253 SocketError,
254 Scope.Scope
255> =>
256 Effect.map(self.reader, ({ pull }) => {
257 const decoder = new TextDecoder(encoding)
258 return Effect.map(pull, (chunk) => {
259 const out = new Array<string>(chunk.length)
260 for (let i = 0; i < chunk.length; i++) {
261 const item = chunk[i]
262 out[i] = typeof item === "string" ? item : decoder.decode(item)
263 }
264 return out as unknown as NonEmptyReadonlyArray<string>
265 })
266 })
267
268const CloseEventTypeId = "~effect/socket/Socket/CloseEvent"
269
270/**
271 * Represents a socket close event value carrying a close code and optional
272 * reason.
273 *
274 * @stability unstable
275 * @category models
276 * @since 4.0.0
277 */
278export class CloseEvent {
279 /**
280 * Marks this value as a socket close event for runtime guards.
281 *
282 * @stability unstable
283 * @since 4.0.0
284 */
285 readonly [CloseEventTypeId]: typeof CloseEventTypeId
286 readonly code: number
287 readonly reason?: string | undefined
288
289 constructor(code = 1000, reason?: string) {
290 this[CloseEventTypeId] = CloseEventTypeId
291 this.code = code
292 this.reason = reason
293 }
294 /**
295 * Formats the close code and optional reason for display.
296 *
297 * @stability unstable
298 * @since 4.0.0
299 */
300 toString() {
301 return this.reason ? `${this.code}: ${this.reason}` : `${this.code}`
302 }
303}
304
305/**
306 * Returns `true` when a value is a `CloseEvent`.
307 *
308 * @stability unstable
309 * @category guards
310 * @since 4.0.0
311 */
312export const isCloseEvent = (u: unknown): u is CloseEvent => Predicate.hasProperty(u, CloseEventTypeId)
313
314/**
315 * Type-level identifier used to mark `SocketError` values.
316 *
317 * @stability unstable
318 * @category type IDs
319 * @since 4.0.0
320 */
321export type SocketErrorTypeId = "~effect/socket/Socket/SocketError"
322
323/**
324 * Runtime type identifier attached to `SocketError` values.
325 *
326 * @stability unstable
327 * @category type IDs
328 * @since 4.0.0
329 */
330export const SocketErrorTypeId: SocketErrorTypeId = "~effect/socket/Socket/SocketError"
331
332/**
333 * Returns `true` when a value is a `SocketError`.
334 *
335 * @stability unstable
336 * @category guards
337 * @since 4.0.0
338 */
339export const isSocketError = (u: unknown): u is SocketError => Predicate.hasProperty(u, SocketErrorTypeId)
340
341/**
342 * Typed error for failures that occur while reading from a socket.
343 *
344 * @stability unstable
345 * @category errors
346 * @since 4.0.0
347 */
348export class SocketReadError extends Schema.Error<SocketReadError>("effect/socket/Socket/SocketReadError")({
349 _tag: Schema.tag("SocketReadError"),
350 cause: Schema.Defect()
351}) {
352 /**
353 * Default message used for socket read failures.
354 *
355 * @stability unstable
356 * @since 4.0.0
357 */
358 override readonly message = `An error occurred during Read`
359}
360
361/**
362 * Typed error for failures that occur while writing to a socket.
363 *
364 * @stability unstable
365 * @category errors
366 * @since 4.0.0
367 */
368export class SocketWriteError extends Schema.Error<SocketWriteError>("effect/socket/Socket/SocketWriteError")({
369 _tag: Schema.tag("SocketWriteError"),
370 cause: Schema.Defect()
371}) {
372 /**
373 * Default message used for socket write failures.
374 *
375 * @stability unstable
376 * @since 4.0.0
377 */
378 override readonly message = `An error occurred during Write`
379}
380
381/**
382 * Typed error for failures that occur while opening a socket, including
383 * unknown open failures and open timeouts.
384 *
385 * @stability unstable
386 * @category errors
387 * @since 4.0.0
388 */
389export class SocketOpenError extends Schema.Error<SocketOpenError>("effect/socket/Socket/SocketOpenError")({
390 _tag: Schema.tag("SocketOpenError"),
391 kind: Schema.Literals(["Unknown", "Timeout"]),
392 cause: Schema.Defect()
393}) {
394 /**
395 * Formats timeout and unknown open failures for display.
396 *
397 * @stability unstable
398 * @since 4.0.0
399 */
400 override get message() {
401 return this.kind === "Timeout"
402 ? `timeout waiting for "open"`
403 : `An error occurred during Open`
404 }
405}
406
407/**
408 * Typed error for an unsupported or failed in-place TLS upgrade.
409 *
410 * @stability unstable
411 * @category errors
412 * @since 4.0.0
413 */
414export class SocketUpgradeError extends Schema.Error<SocketUpgradeError>(
415 "effect/socket/Socket/SocketUpgradeError"
416)({
417 _tag: Schema.tag("SocketUpgradeError"),
418 cause: Schema.optional(Schema.Defect())
419}) {
420 /**
421 * An upgrade implementation for transports that cannot wrap the connection
422 * with TLS.
423 *
424 * @stability unstable
425 * @since 4.0.0
426 */
427 static readonly unsupported: Reader["upgrade"] = () =>
428 Effect.fail(new SocketError({ reason: new SocketUpgradeError({}) }))
429
430 override get message() {
431 return this.cause === undefined
432 ? `Socket does not support TLS upgrade`
433 : `An error occurred during TLS upgrade`
434 }
435}
436
437/**
438 * Typed error for a socket close, carrying the close code and optional close
439 * reason.
440 *
441 * **Details**
442 *
443 * Sockets never classify closes: any close, whatever the code, fails the
444 * reader with a `SocketError` wrapping this reason. Consumers that treat a
445 * close as normal catch it.
446 *
447 * @stability unstable
448 * @category errors
449 * @since 4.0.0
450 */
451export class SocketCloseError extends Schema.Error<SocketCloseError>("effect/socket/Socket/SocketCloseError")({
452 _tag: Schema.tag("SocketCloseError"),
453 code: Schema.Int,
454 closeReason: Schema.optional(Schema.String)
455}) {
456 override get message() {
457 if (this.closeReason) {
458 return `${this.code}: ${this.closeReason}`
459 }
460 return `${this.code}`
461 }
462}
463
464/**
465 * Schema for all socket-specific error reasons.
466 *
467 * @stability unstable
468 * @category errors
469 * @since 4.0.0
470 */
471export const SocketErrorReason = Schema.Union([
472 SocketReadError,
473 SocketWriteError,
474 SocketOpenError,
475 SocketUpgradeError,
476 SocketCloseError
477])
478
479/**
480 * Union of socket-specific read, write, open, upgrade, and close error reasons.
481 *
482 * @stability unstable
483 * @category errors
484 * @since 4.0.0
485 */
486export type SocketErrorReason =
487 | SocketReadError
488 | SocketWriteError
489 | SocketOpenError
490 | SocketUpgradeError
491 | SocketCloseError
492
493/**
494 * Tagged error that wraps socket read, write, open, upgrade, and close failures
495 * while preserving the underlying reason.
496 *
497 * @stability unstable
498 * @category errors
499 * @since 4.0.0
500 */
501export class SocketError extends Schema.TaggedError<SocketError>(SocketErrorTypeId)("SocketError", {
502 _tag: Schema.tag("SocketError"),
503 reason: SocketErrorReason
504}) {
505 // @effect-diagnostics-next-line overriddenSchemaConstructor:off
506 constructor(props: {
507 readonly reason: SocketReadError | SocketWriteError | SocketOpenError | SocketUpgradeError | SocketCloseError
508 }) {
509 if ("cause" in props.reason) {
510 super({
511 ...props,
512 cause: props.reason.cause
513 } as any)
514 } else {
515 super(props)
516 }
517 }
518
519 /**
520 * Marks this value as a socket error wrapper for runtime guards.
521 *
522 * @stability unstable
523 * @since 4.0.0
524 */
525 readonly [SocketErrorTypeId]: SocketErrorTypeId = SocketErrorTypeId
526
527 /**
528 * Returns `true` when the value is a `SocketError`.
529 *
530 * @stability unstable
531 * @since 4.0.0
532 */
533 static is(u: unknown): u is SocketError {
534 return isSocketError(u)
535 }
536
537 override readonly message = this.reason.message
538}
539
540const closeError = (code: number, closeReason?: string | undefined) =>
541 new SocketError({ reason: new SocketCloseError({ code, closeReason }) })
542
543const writeChunk = (
544 writer: Writer,
545 chunk: NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>
546): Effect.Effect<void, SocketError> => {
547 for (let i = 0; i < chunk.length; i++) {
548 if (isCloseEvent(chunk[i])) {
549 return Effect.gen(function*() {
550 for (let index = 0; index <= i; index++) {
551 yield* writer.write(chunk[index])
552 }
553 })
554 }
555 }
556 return writer.writeAll(chunk as NonEmptyReadonlyArray<Uint8Array | string>)
557}
558
559const toChannelWithReader = <A extends Uint8Array | string, IE>(
560 self: Socket,
561 reader: Effect.Effect<
562 Effect.Effect<NonEmptyReadonlyArray<A>, SocketError>,
563 SocketError,
564 Scope.Scope
565 >
566): Channel.Channel<
567 NonEmptyReadonlyArray<A>,
568 SocketError | IE,
569 void,
570 NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>,
571 IE
572> =>
573 Channel.fromTransform(Effect.fnUntraced(function*(upstream, scope) {
574 const readScope = yield* Scope.fork(scope)
575 const pull = yield* Scope.provide(reader, readScope)
576 const writeScope = yield* Scope.fork(scope)
577 const writer = yield* Scope.provide(self.writer, writeScope)
578
579 let writeFailure: Cause.Cause<SocketError | IE> | undefined
580
581 yield* upstream.pipe(
582 Effect.flatMap((chunk) => writeChunk(writer, chunk)),
583 Effect.forever({ disableYield: true }),
584 Effect.catchCauseFilter(
585 Pull.filterNoDone,
586 (cause) =>
587 Effect.suspend(() => {
588 writeFailure = cause as Cause.Cause<SocketError | IE>
589 // closing the read scope fails a suspended pull, which is the
590 // reader contract, so the read side needs no per-pull failure race
591 return Scope.close(readScope, Exit.void)
592 })
593 ),
594 Effect.ensuring(Scope.close(writeScope, Exit.void)),
595 Effect.forkIn(scope)
596 )
597
598 // @effect-diagnostics-next-line returnEffectInGen:off
599 return Effect.catchCause(
600 Effect.suspend(
601 (): Pull.Pull<NonEmptyReadonlyArray<A>, SocketError | IE> =>
602 writeFailure !== undefined ? Effect.failCause(writeFailure) : pull
603 ),
604 (cause) =>
605 Effect.failCause(
606 (writeFailure ?? cause) as Cause.Cause<SocketError | IE | Cause.Done<void>>
607 )
608 )
609 }))
610
611/**
612 * Converts a `Socket` into a bidirectional binary `Channel`, encoding
613 * incoming string frames as UTF-8 bytes and writing outgoing frame batches to
614 * the socket.
615 *
616 * **Details**
617 *
618 * The read side is the socket's pull, so the channel is backpressured
619 * end-to-end.
620 *
621 * @stability unstable
622 * @category combinators
623 * @since 4.0.0
624 */
625export const toChannel = <IE = never>(
626 self: Socket
627): Channel.Channel<
628 NonEmptyReadonlyArray<Uint8Array>,
629 SocketError | IE,
630 void,
631 NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>,
632 IE
633> => toChannelWithReader<Uint8Array, IE>(self, readerBytes(self))
634
635/**
636 * Converts a `Socket` into a bidirectional string `Channel`, decoding binary
637 * frames with the optional text encoding.
638 *
639 * @stability unstable
640 * @category combinators
641 * @since 4.0.0
642 */
643export const toChannelString: {
644 (encoding?: string | undefined): <IE>(self: Socket) => Channel.Channel<
645 NonEmptyReadonlyArray<string>,
646 SocketError | IE,
647 void,
648 NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>,
649 IE
650 >
651 <IE>(
652 self: Socket,
653 encoding?: string | undefined
654 ): Channel.Channel<
655 NonEmptyReadonlyArray<string>,
656 SocketError | IE,
657 void,
658 NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>,
659 IE
660 >
661} = dual((args) => isSocket(args[0]), <IE>(
662 self: Socket,
663 encoding?: string | undefined
664): Channel.Channel<
665 NonEmptyReadonlyArray<string>,
666 SocketError | IE,
667 void,
668 NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>,
669 IE
670> => toChannelWithReader<string, IE>(self, readerString(self, encoding)))
671
672/**
673 * Creates a `Socket` to binary `Channel` adapter with a fixed upstream error
674 * type.
675 *
676 * @stability unstable
677 * @category combinators
678 * @since 4.0.0
679 */
680export const toChannelWith = <IE = never>() =>
681(
682 self: Socket
683): Channel.Channel<
684 NonEmptyReadonlyArray<Uint8Array>,
685 SocketError | IE,
686 void,
687 NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>,
688 IE
689> => toChannel(self)
690
691/**
692 * Converts a `Socket` into a read-only binary `Stream` backed by the socket's
693 * pull, so consumption is backpressured end-to-end.
694 *
695 * @stability unstable
696 * @category combinators
697 * @since 4.0.0
698 */
699export const toStream = (self: Socket): Stream.Stream<Uint8Array, SocketError> =>
700 Stream.fromChannel(
701 Channel.fromTransform((_, scope) => Scope.provide(readerBytes(self), scope))
702 )
703
704/**
705 * Creates a binary socket `Channel` from the `Socket` service in the
706 * environment.
707 *
708 * @stability unstable
709 * @category constructors
710 * @since 4.0.0
711 */
712export const makeChannel = <IE = never>(): Channel.Channel<
713 NonEmptyReadonlyArray<Uint8Array>,
714 SocketError | IE,
715 void,
716 NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>,
717 IE,
718 unknown,
719 Socket
720> => Channel.unwrap(Effect.map(Socket, toChannelWith<IE>()))
721
722/**
723 * Event payload exposed by a WebSocket implementation.
724 *
725 * **Details**
726 *
727 * The socket adapter only reads `data`, `code`, and `reason`; implementations
728 * may expose additional fields.
729 *
730 * @stability unstable
731 * @category models
732 * @since 4.0.0
733 */
734export interface WebSocketEvent {
735 readonly type?: string
736 readonly data?: unknown
737 readonly code?: number
738 readonly reason?: string
739}
740
741/**
742 * The subset of the WebSocket API required by `Socket`.
743 *
744 * **Details**
745 *
746 * This structural interface is intentionally independent of the DOM
747 * `WebSocket` type. Node implementations such as `ws` expose the same
748 * event-target methods, but are not assignable to `globalThis.WebSocket`
749 * because their event payload types are runtime-specific.
750 *
751 * @stability unstable
752 * @category models
753 * @since 4.0.0
754 */
755export interface WebSocketLike {
756 readonly readyState: number
757 addEventListener(
758 type: "open" | "message" | "error" | "close",
759 listener: (event: WebSocketEvent) => void,
760 options?: {
761 readonly once?: boolean
762 }
763 ): void
764 removeEventListener(
765 type: "open" | "message" | "error" | "close",
766 listener: (event: WebSocketEvent) => void
767 ): void
768 close(code?: number, reason?: string): void
769 send(data: string | Uint8Array<ArrayBuffer>): void
770}
771
772/**
773 * Common options understood by a WebSocket client implementation.
774 *
775 * @stability unstable
776 * @category models
777 * @since 4.0.0
778 */
779export interface WebSocketClientOptions {
780 /**
781 * Headers to include in the opening handshake.
782 *
783 * **Details**
784 *
785 * This is supported by Node and Bun WebSocket clients. Browser constructors
786 * cannot set arbitrary handshake headers and must reject this option.
787 */
788 readonly headers?: Readonly<Record<string, string>> | undefined
789}
790
791/**
792 * Options accepted by a `WebSocketConstructor`.
793 *
794 * **Details**
795 *
796 * Browser-compatible constructors accept a protocol string or list. Node and
797 * Bun constructors additionally accept `WebSocketClientOptions`.
798 *
799 * @stability unstable
800 * @category models
801 * @since 4.0.0
802 */
803export type WebSocketConstructorOptions = string | Array<string> | WebSocketClientOptions
804
805/**
806 * Context service for the active `WebSocket` instance.
807 *
808 * @stability unstable
809 * @category services
810 * @since 4.0.0
811 */
812export class WebSocket extends Context.Service<WebSocket, WebSocketLike>()(
813 "~effect/socket/Socket/WebSocket"
814) {}
815
816/**
817 * Context service for constructing `WebSocket` instances from a URL and
818 * optional protocols or platform-specific options.
819 *
820 * @stability unstable
821 * @category services
822 * @since 4.0.0
823 */
824export class WebSocketConstructor extends Context.Service<
825 WebSocketConstructor,
826 (url: string, options?: WebSocketConstructorOptions | undefined) => WebSocketLike
827>()("@effect/platform/Socket/WebSocketConstructor") {}
828
829/**
830 * Layer that provides `WebSocketConstructor` using `globalThis.WebSocket`.
831 *
832 * @stability unstable
833 * @category layers
834 * @since 4.0.0
835 */
836export const layerWebSocketConstructorGlobal: Layer.Layer<WebSocketConstructor> = Layer.succeed(WebSocketConstructor)(
837 (url, options) => {
838 if (options !== undefined && typeof options !== "string" && !Array.isArray(options)) {
839 throw new TypeError("WebSocket client options are not supported by the global WebSocket constructor")
840 }
841 return new globalThis.WebSocket(url, options)
842 }
843)
844
845/**
846 * Creates a `Socket` backed by a `WebSocketConstructor`, dialing the
847 * WebSocket for each reader acquisition.
848 *
849 * @stability unstable
850 * @category constructors
851 * @since 4.0.0
852 */
853export const makeWebSocket = (url: string | Effect.Effect<string>, options?: {
854 readonly openTimeout?: Duration.Input | undefined
855 readonly protocols?: string | Array<string> | undefined
856 readonly highWaterMark?: number | undefined
857}): Effect.Effect<Socket, never, WebSocketConstructor> =>
858 WebSocketConstructor.use((makeWs) =>
859 fromWebSocket(
860 Effect.acquireRelease(
861 (typeof url === "string" ? Effect.succeed(url) : url).pipe(
862 Effect.map((url) => makeWs(url, options?.protocols))
863 ),
864 (ws) => Effect.sync(() => ws.close(1000))
865 ),
866 options
867 )
868 )
869
870interface Pausable {
871 pause(): void
872 resume(): void
873}
874
875const isPausable = (ws: WebSocketLike): ws is WebSocketLike & Pausable =>
876 "pause" in ws && typeof (ws as any).pause === "function"
877
878/**
879 * Builds a `Socket` from a scoped WebSocket acquisition effect.
880 *
881 * **Details**
882 *
883 * Reader acquisition runs `acquire`, attaches event listeners, then waits for
884 * the socket to open. Implementations exposing `pause`/`resume` (the `ws`
885 * package) remain flowing until buffered frames reach the configured
886 * `highWaterMark` (64 KiB by default). Draining the buffer resumes the
887 * transport. Incoming frames that arrive in the same tick are coalesced into
888 * one batch. Implementations without pause (browsers) buffer incoming frames,
889 * optionally failing the socket with a `SocketReadError` when `highWaterMark`
890 * bytes are exceeded (default unbounded). Message boundaries survive: each
891 * pulled batch contains one element per frame.
892 *
893 * @stability unstable
894 * @category constructors
895 * @since 4.0.0
896 */
897export const fromWebSocket = <RO, WS extends WebSocketLike>(
898 acquire: Effect.Effect<WS, SocketError, RO>,
899 options?: {
900 readonly openTimeout?: Duration.Input | undefined
901 readonly highWaterMark?: number | undefined
902 } | undefined
903): Effect.Effect<Socket, never, Exclude<RO, Scope.Scope>> =>
904 Effect.withFiber((fiber) => {
905 let currentWS: WebSocketLike | undefined
906 const latch = Latch.makeUnsafe(false)
907 const acquireContext = fiber.context as Context.Context<RO>
908
909 const reader: Socket["reader"] = Effect.gen(function*() {
910 const scope = yield* Effect.scope
911 const ws = yield* Scope.provide(acquire, scope)
912 if ("binaryType" in ws) {
913 ;(ws as { binaryType: string }).binaryType = "arraybuffer"
914 }
915 const pausable = isPausable(ws)
916 const highWaterMark = options?.highWaterMark ?? (pausable ? defaultHighWaterMark : undefined)
917
918 type ReadResume = (
919 effect: Effect.Effect<NonEmptyReadonlyArray<Uint8Array | string>, SocketError>
920 ) => void
921
922 let open = ws.readyState === 1
923 let paused = pausable && "isPaused" in ws && ws.isPaused === true
924 let buffer: Array<Uint8Array | string> = []
925 let bufferSize = 0
926 let error: SocketError | undefined
927 let waiter: ReadResume | undefined
928 let openWaiter:
929 | {
930 readonly resume: (effect: Effect.Effect<void, SocketError>) => void
931 readonly cleanup: () => void
932 }
933 | undefined
934
935 function pauseWebSocket() {
936 if (!pausable || paused) return
937 ws.pause()
938 paused = true
939 }
940
941 function resumeWebSocket() {
942 if (!pausable || !paused) return
943 ws.resume()
944 paused = false
945 }
946
947 function takeBuffer(): NonEmptyReadonlyArray<Uint8Array | string> {
948 const chunk = buffer
949 buffer = []
950 bufferSize = 0
951 if (error === undefined) resumeWebSocket()
952 return chunk as unknown as NonEmptyReadonlyArray<Uint8Array | string>
953 }
954
955 function deliver() {
956 if (waiter === undefined || buffer.length === 0) return
957 const resume = waiter
958 waiter = undefined
959 resume(Effect.succeed(takeBuffer()))
960 }
961
962 function push(data: Uint8Array | string) {
963 buffer.push(data)
964 if (highWaterMark !== undefined) {
965 bufferSize += typeof data === "string" ? encoder.encode(data).byteLength : data.byteLength
966 }
967 if (waiter !== undefined) deliver()
968 if (pausable && bufferSize >= highWaterMark!) {
969 pauseWebSocket()
970 } else if (!pausable && !waiter && highWaterMark !== undefined && bufferSize > highWaterMark) {
971 fail(
972 new SocketError({
973 reason: new SocketReadError({
974 cause: new Error(`Socket highWaterMark of ${highWaterMark} bytes exceeded`)
975 })
976 })
977 )
978 }
979 }
980
981 function fail(err: SocketError) {
982 if (error === undefined) error = err
983 if (openWaiter !== undefined) {
984 const { cleanup, resume } = openWaiter
985 openWaiter = undefined
986 cleanup()
987 resume(Effect.fail(error))
988 }
989 if (waiter !== undefined) {
990 const resume = waiter
991 waiter = undefined
992 resume(buffer.length > 0 ? Effect.succeed(takeBuffer()) : Effect.fail(error))
993 }
994 }
995
996 function onMessage(event: WebSocketEvent) {
997 const data = event.data as Uint8Array | ArrayBuffer | Blob | string
998 if (typeof Blob !== "undefined" && data instanceof Blob) {
999 data.arrayBuffer().then((buf) => push(new Uint8Array(buf)), (cause) => {
1000 fail(new SocketError({ reason: new SocketReadError({ cause }) }))
1001 })
1002 return
1003 }
1004 push(data instanceof ArrayBuffer ? new Uint8Array(data) : data as Uint8Array | string)
1005 }
1006 function onError(event: WebSocketEvent) {
1007 fail(
1008 new SocketError({
1009 reason: open
1010 ? new SocketReadError({ cause: event })
1011 : new SocketOpenError({ kind: "Unknown", cause: event })
1012 })
1013 )
1014 }
1015 function onClose(event: WebSocketEvent) {
1016 fail(closeError(typeof event.code === "number" ? event.code : 1006, event.reason))
1017 }
1018
1019 ws.addEventListener("message", onMessage)
1020 ws.addEventListener("error", onError, { once: true })
1021 ws.addEventListener("close", onClose, { once: true })
1022 yield* Scope.addFinalizer(
1023 scope,
1024 Effect.sync(() => {
1025 // resume a pull blocked in another fiber before detaching
1026 fail(closeError(1006))
1027 ws.removeEventListener("message", onMessage)
1028 ws.removeEventListener("error", onError)
1029 ws.removeEventListener("close", onClose)
1030 // let the close handshake proceed once nothing is pulling
1031 if (pausable) {
1032 try {
1033 resumeWebSocket()
1034 } catch {
1035 // the underlying stream may already be gone
1036 }
1037 }
1038 latch.closeUnsafe()
1039 currentWS = undefined
1040 })
1041 )
1042
1043 if (ws.readyState >= 2) {
1044 // @effect-diagnostics-next-line unnecessaryFailYieldableError:off
1045 return yield* Effect.fail(error ?? closeError(1006))
1046 }
1047
1048 if (!open) {
1049 yield* Effect.callback<void, SocketError>((resume) => {
1050 const onOpen = () => {
1051 if (openWaiter?.resume !== resume) return
1052 openWaiter = undefined
1053 ws.removeEventListener("open", onOpen)
1054 resume(Effect.void)
1055 }
1056 const cleanup = () => ws.removeEventListener("open", onOpen)
1057 openWaiter = { resume, cleanup }
1058 ws.addEventListener("open", onOpen, { once: true })
1059 return Effect.sync(() => {
1060 if (openWaiter?.resume === resume) openWaiter = undefined
1061 cleanup()
1062 })
1063 }).pipe(
1064 Effect.timeoutOrElse({
1065 duration: options?.openTimeout ?? 10000,
1066 orElse: () =>
1067 Effect.fail(
1068 new SocketError({
1069 reason: new SocketOpenError({
1070 kind: "Timeout",
1071 cause: new Error("timeout waiting for \"open\"")
1072 })
1073 })
1074 )
1075 })
1076 )
1077 open = true
1078 }
1079
1080 if (error === undefined && bufferSize < highWaterMark!) resumeWebSocket()
1081 currentWS = ws
1082 latch.openUnsafe()
1083
1084 return {
1085 pull: Effect.callback<NonEmptyReadonlyArray<Uint8Array | string>, SocketError>((resume) => {
1086 if (buffer.length > 0) return resume(Effect.succeed(takeBuffer()))
1087 if (error !== undefined) return resume(Effect.fail(error))
1088 waiter = resume
1089 return Effect.sync(() => {
1090 if (waiter === resume) waiter = undefined
1091 })
1092 }),
1093 upgrade: SocketUpgradeError.unsupported
1094 }
1095 }).pipe(
1096 Effect.updateContext((input: Context.Context<Scope.Scope>) => Context.merge(acquireContext, input))
1097 ) as Socket["reader"]
1098
1099 const write = (chunk: Uint8Array | string | CloseEvent): Effect.Effect<void, SocketError> =>
1100 Effect.suspend(() => {
1101 const ws = currentWS
1102 if (ws === undefined) return latch.whenOpen(write(chunk))
1103 try {
1104 if (isCloseEvent(chunk)) {
1105 ws.close(chunk.code, chunk.reason)
1106 } else {
1107 ws.send(chunk as string | Uint8Array<ArrayBuffer>)
1108 }
1109 return Effect.void
1110 } catch (cause) {
1111 return Effect.fail(new SocketError({ reason: new SocketWriteError({ cause }) }))
1112 }
1113 })
1114 const writeAll = (
1115 chunks: NonEmptyReadonlyArray<Uint8Array | string>
1116 ): Effect.Effect<void, SocketError> =>
1117 Effect.suspend(() => {
1118 const ws = currentWS
1119 if (ws === undefined) return latch.whenOpen(writeAll(chunks))
1120 try {
1121 for (let i = 0; i < chunks.length; i++) {
1122 ws.send(chunks[i] as string | Uint8Array<ArrayBuffer>)
1123 }
1124 return Effect.void
1125 } catch (cause) {
1126 return Effect.fail(new SocketError({ reason: new SocketWriteError({ cause }) }))
1127 }
1128 })
1129 const writer: Socket["writer"] = Effect.succeed({ write, writeAll })
1130
1131 return Effect.succeed(make({ reader, writer }))
1132 })
1133
1134const defaultHighWaterMark = 64 * 1024
1135
1136/**
1137 * Creates a binary `Channel` backed by a WebSocket URL, requiring a
1138 * `WebSocketConstructor` service.
1139 *
1140 * @stability unstable
1141 * @category constructors
1142 * @since 4.0.0
1143 */
1144export const makeWebSocketChannel = <IE = never>(
1145 url: string,
1146 options?: {
1147 readonly openTimeout?: Duration.Input | undefined
1148 readonly protocols?: string | Array<string> | undefined
1149 readonly highWaterMark?: number | undefined
1150 }
1151): Channel.Channel<
1152 NonEmptyReadonlyArray<Uint8Array>,
1153 SocketError | IE,
1154 void,
1155 NonEmptyReadonlyArray<Uint8Array | string | CloseEvent>,
1156 IE,
1157 unknown,
1158 WebSocketConstructor
1159> =>
1160 Channel.unwrap(
1161 Effect.map(makeWebSocket(url, options), toChannelWith<IE>())
1162 )
1163
1164/**
1165 * Layer that provides a `Socket` service backed by a WebSocket URL or URL
1166 * effect.
1167 *
1168 * @stability unstable
1169 * @category layers
1170 * @since 4.0.0
1171 */
1172export const layerWebSocket: (
1173 url: string | Effect.Effect<string>,
1174 options?: {
1175 readonly openTimeout?: Duration.Input | undefined
1176 readonly protocols?: string | Array<string> | undefined
1177 readonly highWaterMark?: number | undefined
1178 } | undefined
1179) => Layer.Layer<Socket, never, WebSocketConstructor> = flow(makeWebSocket, Layer.effect(Socket))
1180
1181/**
1182 * Readable and writable stream pair used to adapt transform-style streams into
1183 * a `Socket`.
1184 *
1185 * @stability unstable
1186 * @category models
1187 * @since 4.0.0
1188 */
1189export interface InputTransformStream {
1190 readonly readable: ReadableStream<Uint8Array> | ReadableStream<string> | ReadableStream<Uint8Array | string>
1191 readonly writable: WritableStream<Uint8Array>
1192}
1193
1194/**
1195 * Builds a `Socket` from a scoped `InputTransformStream`, pulling incoming
1196 * chunks from the readable side and writing outgoing chunks to the writable
1197 * stream, encoding strings as UTF-8.
1198 *
1199 * @stability unstable
1200 * @category constructors
1201 * @since 4.0.0
1202 */
1203export const fromTransformStream = <R>(
1204 acquire: Effect.Effect<InputTransformStream, SocketError, R>
1205): Effect.Effect<Socket, never, Exclude<R, Scope.Scope>> =>
1206 Effect.withFiber((fiber) => {
1207 const latch = Latch.makeUnsafe(false)
1208 let currentStream: {
1209 readonly stream: InputTransformStream
1210 readonly fail: (error: SocketError) => void
1211 } | undefined
1212 const acquireContext = fiber.context as Context.Context<R>
1213
1214 const writers = new WeakMap<InputTransformStream, WritableStreamDefaultWriter<Uint8Array>>()
1215 const getWriter = (stream: InputTransformStream) => {
1216 let writer = writers.get(stream)
1217 if (!writer) {
1218 writer = stream.writable.getWriter()
1219 writers.set(stream, writer)
1220 }
1221 return writer
1222 }
1223
1224 const reader: Socket["reader"] = Effect.gen(function*() {
1225 const scope = yield* Effect.scope
1226 const stream = yield* Scope.provide(acquire, scope)
1227 const readerHandle = (stream.readable as ReadableStream<Uint8Array | string>).getReader()
1228 let error: SocketError | undefined
1229 yield* Scope.addFinalizer(
1230 scope,
1231 Effect.suspend(() => {
1232 latch.closeUnsafe()
1233 currentStream = undefined
1234 return Effect.promise(() => readerHandle.cancel().catch(constVoid))
1235 })
1236 )
1237 currentStream = {
1238 stream,
1239 fail(err) {
1240 if (error === undefined) error = err
1241 readerHandle.cancel().catch(constVoid)
1242 }
1243 }
1244 latch.openUnsafe()
1245
1246 const read = Effect.tryPromise({
1247 try: () => readerHandle.read(),
1248 catch: (cause) =>
1249 error ?? new SocketError({
1250 reason: new SocketReadError({ cause })
1251 })
1252 })
1253 return {
1254 pull: Effect.suspend(() => {
1255 if (error !== undefined) return Effect.fail(error)
1256 return Effect.flatMap(read, ({ done, value }) =>
1257 done
1258 ? Effect.fail(error ?? closeError(1000))
1259 : Effect.succeed([value] as unknown as NonEmptyReadonlyArray<Uint8Array | string>))
1260 }),
1261 upgrade: SocketUpgradeError.unsupported
1262 }
1263 }).pipe(
1264 Effect.updateContext((input: Context.Context<Scope.Scope>) => Context.merge(acquireContext, input))
1265 ) as Socket["reader"]
1266
1267 const write = (chunk: Uint8Array | string | CloseEvent) =>
1268 latch.whenOpen(Effect.suspend(() => {
1269 const current = currentStream!
1270 if (isCloseEvent(chunk)) {
1271 return Effect.sync(() => {
1272 current.fail(closeError(chunk.code, chunk.reason))
1273 })
1274 }
1275 return Effect.tryPromise({
1276 try: () => getWriter(current.stream).write(typeof chunk === "string" ? encoder.encode(chunk) : chunk),
1277 catch: (cause) => new SocketError({ reason: new SocketWriteError({ cause }) })
1278 })
1279 }))
1280 const writeAll = (chunks: NonEmptyReadonlyArray<Uint8Array | string>) =>
1281 latch.whenOpen(Effect.tryPromise({
1282 try: async () => {
1283 const writer = getWriter(currentStream!.stream)
1284 for (let i = 0; i < chunks.length; i++) {
1285 const chunk = chunks[i]
1286 await writer.write(typeof chunk === "string" ? encoder.encode(chunk) : chunk)
1287 }
1288 },
1289 catch: (cause) => new SocketError({ reason: new SocketWriteError({ cause }) })
1290 }))
1291 const writer: Socket["writer"] = Effect.acquireRelease(
1292 Effect.succeed({ write, writeAll }),
1293 () =>
1294 Effect.promise(async () => {
1295 if (!currentStream) return
1296 await getWriter(currentStream.stream).close().catch(constVoid)
1297 })
1298 )
1299
1300 return Effect.succeed(make({ reader, writer }))
1301 })