1/**
2 * Runs typed RPC calls from the client side.
3 *
4 * This module turns RPC definitions from an `RpcGroup` into callable client
5 * methods. Each call encodes its payload, sends a message through the active
6 * `Protocol`, decodes exits or stream chunks from the server, and routes the
7 * response back to the waiting `Effect`, `Stream`, or queue. It also defines
8 * the protocol service and includes protocol layers for HTTP, sockets, and
9 * workers.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import type { NonEmptyReadonlyArray } from "../Array.ts"
15import * as Cause from "../Cause.ts"
16import * as Context from "../Context.ts"
17import type * as Duration from "../Duration.ts"
18import * as Effect from "../Effect.ts"
19import * as Exit from "../Exit.ts"
20import * as Fiber from "../Fiber.ts"
21import { constVoid, dual, flow, identity } from "../Function.ts"
22import * as Headers from "../http/Headers.ts"
23import * as HttpBody from "../http/HttpBody.ts"
24import * as HttpClient from "../http/HttpClient.ts"
25import { HttpClientErrorSchema } from "../http/HttpClientError.ts"
26import * as HttpClientRequest from "../http/HttpClientRequest.ts"
27import * as InternalRecord from "../internal/record.ts"
28import * as Latch from "../Latch.ts"
29import * as Layer from "../Layer.ts"
30import * as Option from "../Option.ts"
31import * as Pool from "../Pool.ts"
32import * as Predicate from "../Predicate.ts"
33import * as Queue from "../Queue.ts"
34import * as Result from "../Result.ts"
35import * as Schedule from "../Schedule.ts"
36import * as Schema from "../Schema.ts"
37import * as Scope from "../Scope.ts"
38import * as Socket from "../socket/Socket.ts"
39import * as Stream from "../Stream.ts"
40import type * as Struct from "../Struct.ts"
41import type { Span } from "../Tracer.ts"
42import * as Transferable from "../workers/Transferable.ts"
43import * as Worker from "../workers/Worker.ts"
44import type { WorkerError } from "../workers/WorkerError.ts"
45import * as Rpc from "./Rpc.ts"
46import { RpcClientDefect, RpcClientError } from "./RpcClientError.ts"
47import type * as RpcGroup from "./RpcGroup.ts"
48import type { FromClient, FromClientEncoded, FromServer, FromServerEncoded, Request } from "./RpcMessage.ts"
49import { constPing, isTerminalResponse, RequestId } from "./RpcMessage.ts"
50import type * as RpcMiddleware from "./RpcMiddleware.ts"
51import * as RpcSchema from "./RpcSchema.ts"
52import * as RpcSerialization from "./RpcSerialization.ts"
53import * as RpcWorker from "./RpcWorker.ts"
54import { withRunClient } from "./Utils.ts"
55
56const isRpcClientError = (u: unknown): u is RpcClientError => Predicate.isTagged(u, "RpcClientError")
57
58/**
59 * The object-shaped client generated from a union of RPC definitions, with one
60 * method per RPC tag.
61 *
62 * @stability unstable
63 * @category utility types
64 * @since 4.0.0
65 */
66export type RpcClient<Rpcs extends Rpc.Any, E = never> = Struct.Simplify<RpcClient.From<Rpcs, E>>
67
68/**
69 * Type-level helpers for deriving RPC client call signatures from RPC
70 * definitions.
71 *
72 * @stability unstable
73 * @since 4.0.0
74 */
75export declare namespace RpcClient {
76 /**
77 * Builds an object client type from an RPC union, mapping each RPC tag to a
78 * method that accepts the RPC payload and returns either an `Effect` or
79 * `Stream` based on the RPC success schema.
80 *
81 * @stability unstable
82 * @category utility types
83 * @since 4.0.0
84 */
85 export type From<Rpcs extends Rpc.Any, E = never> = {
86 readonly [Current in Rpcs as Current["_tag"]]: <
87 const AsQueue extends boolean = false,
88 const Discard = false
89 >(
90 input: Rpc.PayloadConstructor<Current>,
91 options?: Rpc.Success<Current> extends Stream.Stream<infer _A, infer _E, infer _R> ? {
92 readonly asQueue?: AsQueue | undefined
93 readonly streamBufferSize?: number | undefined
94 readonly headers?: Headers.Input | undefined
95 readonly context?: Context.Context<never> | undefined
96 } :
97 {
98 readonly headers?: Headers.Input | undefined
99 readonly context?: Context.Context<never> | undefined
100 readonly discard?: Discard | undefined
101 }
102 ) => Current extends Rpc.Rpc<
103 infer _Tag,
104 infer _Payload,
105 infer _Success,
106 infer _Error,
107 infer _Middleware,
108 infer _Requires
109 > ? [_Success] extends [RpcSchema.Stream<infer _A, infer _E>] ? AsQueue extends true ? Effect.Effect<
110 Queue.Dequeue<
111 _A["Type"],
112 _E["Type"] | _Error["Type"] | E | _Middleware["error"]["Type"] | _Middleware["~ClientError"] | Cause.Done
113 >,
114 never,
115 | Scope.Scope
116 | _Payload["EncodingServices"]
117 | _Success["DecodingServices"]
118 | _Error["DecodingServices"]
119 | _Middleware["error"]["DecodingServices"]
120 >
121 : Stream.Stream<
122 _A["Type"],
123 _E["Type"] | _Error["Type"] | E | _Middleware["error"]["Type"] | _Middleware["~ClientError"],
124 | _Payload["EncodingServices"]
125 | _Success["DecodingServices"]
126 | _Error["DecodingServices"]
127 | _Middleware["error"]["DecodingServices"]
128 >
129 : Effect.Effect<
130 Discard extends true ? void : _Success["Type"],
131 | (Discard extends true ? never : _Error["Type"])
132 | E
133 | _Middleware["error"]["Type"]
134 | _Middleware["~ClientError"],
135 | _Payload["EncodingServices"]
136 | _Success["DecodingServices"]
137 | _Error["DecodingServices"]
138 | _Middleware["error"]["DecodingServices"]
139 > :
140 never
141 }
142
143 /**
144 * Builds a flattened RPC client function that accepts an RPC tag and payload,
145 * returning the corresponding `Effect` or `Stream` for that RPC.
146 *
147 * @stability unstable
148 * @category utility types
149 * @since 4.0.0
150 */
151 export type Flat<Rpcs extends Rpc.Any, E = never> = <
152 const Tag extends Rpcs["_tag"],
153 const AsQueue extends boolean = false,
154 const Discard = false
155 >(
156 tag: Tag,
157 payload: Rpc.PayloadConstructor<Rpc.ExtractTag<Rpcs, Tag>>,
158 options?: Rpc.Success<Rpc.ExtractTag<Rpcs, Tag>> extends Stream.Stream<infer _A, infer _E, infer _R> ? {
159 readonly asQueue?: AsQueue | undefined
160 readonly streamBufferSize?: number | undefined
161 readonly headers?: Headers.Input | undefined
162 readonly context?: Context.Context<never> | undefined
163 } :
164 {
165 readonly headers?: Headers.Input | undefined
166 readonly context?: Context.Context<never> | undefined
167 readonly discard?: Discard | undefined
168 }
169 ) => Rpcs extends Rpc.Rpc<
170 infer _Tag extends Tag,
171 infer _Payload,
172 infer _Success,
173 infer _Error,
174 infer _Middleware,
175 infer _Requires
176 > ? [_Success] extends [RpcSchema.Stream<infer _A, infer _E>] ? AsQueue extends true ? Effect.Effect<
177 Queue.Dequeue<
178 _A["Type"],
179 _E["Type"] | _Error["Type"] | E | _Middleware["error"]["Type"] | _Middleware["~ClientError"]
180 >,
181 never,
182 | Scope.Scope
183 | _Payload["EncodingServices"]
184 | _Success["DecodingServices"]
185 | _Error["DecodingServices"]
186 | _Middleware["error"]["DecodingServices"]
187 >
188 : Stream.Stream<
189 _A["Type"],
190 _E["Type"] | _Error["Type"] | E | _Middleware["error"]["Type"] | _Middleware["~ClientError"],
191 | _Payload["EncodingServices"]
192 | _Success["DecodingServices"]
193 | _Error["DecodingServices"]
194 | _Middleware["error"]["DecodingServices"]
195 >
196 : Effect.Effect<
197 Discard extends true ? void : _Success["Type"],
198 (Discard extends true ? never : _Error["Type"]) | E | _Middleware["error"]["Type"] | _Middleware["~ClientError"],
199 | _Payload["EncodingServices"]
200 | _Success["DecodingServices"]
201 | _Error["DecodingServices"]
202 | _Middleware["error"]["DecodingServices"]
203 > :
204 never
205}
206
207/**
208 * Derives the object-shaped RPC client type for all RPCs contained in an
209 * `RpcGroup`.
210 *
211 * @stability unstable
212 * @category utility types
213 * @since 4.0.0
214 */
215export type FromGroup<Group, E = never> = RpcClient<RpcGroup.Rpcs<Group>, E>
216
217let requestIdCounter = 0
218
219/**
220 * Creates an RPC client for an already-decoded message channel, returning the
221 * client API together with a `write` function for delivering server messages
222 * back to the client.
223 *
224 * @stability unstable
225 * @category constructors
226 * @since 4.0.0
227 */
228export const makeNoSerialization: <Rpcs extends Rpc.Any, E, const Flatten extends boolean = false>(
229 group: RpcGroup.RpcGroup<Rpcs>,
230 options: {
231 readonly onFromClient: (
232 options: {
233 readonly message: FromClient<Rpcs>
234 readonly context: Context.Context<never>
235 readonly discard: boolean
236 }
237 ) => Effect.Effect<void, E>
238 readonly supportsAck?: boolean | undefined
239 readonly spanPrefix?: string | undefined
240 readonly spanAttributes?: Record<string, unknown> | undefined
241 readonly generateRequestId?: (() => RequestId) | undefined
242 readonly disableTracing?: boolean | undefined
243 readonly flatten?: Flatten | undefined
244 }
245) => Effect.Effect<
246 {
247 readonly client: Flatten extends true ? RpcClient.Flat<Rpcs, E> : RpcClient<Rpcs, E>
248 readonly write: (message: FromServer<Rpcs>) => Effect.Effect<void>
249 },
250 never,
251 Scope.Scope | Rpc.MiddlewareClient<Rpcs>
252> = Effect.fnUntraced(function*<Rpcs extends Rpc.Any, E, const Flatten extends boolean = false>(
253 group: RpcGroup.RpcGroup<Rpcs>,
254 options: {
255 readonly onFromClient: (
256 options: {
257 readonly message: FromClient<Rpcs>
258 readonly context: Context.Context<never>
259 readonly discard: boolean
260 }
261 ) => Effect.Effect<void, E>
262 readonly supportsAck?: boolean | undefined
263 readonly spanPrefix?: string | undefined
264 readonly spanAttributes?: Record<string, unknown> | undefined
265 readonly generateRequestId?: (() => RequestId) | undefined
266 readonly disableTracing?: boolean | undefined
267 readonly flatten?: Flatten | undefined
268 }
269) {
270 const spanPrefix = options?.spanPrefix ?? "RpcClient"
271 const supportsAck = options?.supportsAck ?? true
272 const disableTracing = options?.disableTracing ?? false
273 const generateRequestId = options?.generateRequestId ?? (() => requestIdCounter++ as RequestId)
274
275 const services = yield* Effect.context<Rpc.MiddlewareClient<Rpcs> | Scope.Scope>()
276 const scope = Context.get(services, Scope.Scope)
277
278 type ClientEntry = {
279 readonly _tag: "Effect"
280 readonly rpc: Rpc.AnyWithProps
281 readonly context: Context.Context<never>
282 resume: (_: Exit.Exit<any, any>) => void
283 } | {
284 readonly _tag: "Queue"
285 readonly rpc: Rpc.AnyWithProps
286 readonly queue: Queue.Queue<any, any>
287 readonly scope: Scope.Scope
288 readonly context: Context.Context<never>
289 }
290 const entries = new Map<RequestId, ClientEntry>()
291
292 let isShutdown = false
293 yield* Scope.addFinalizer(
294 scope,
295 Effect.withFiber((parent) => {
296 isShutdown = true
297 return clearEntries(Exit.interrupt(parent.id))
298 })
299 )
300
301 const clearEntries = Effect.fnUntraced(function*(exit: Exit.Exit<never>) {
302 for (const [id, entry] of entries) {
303 entries.delete(id)
304 if (entry._tag === "Queue") {
305 yield* (exit._tag === "Success"
306 ? Queue.end(entry.queue as any)
307 : Queue.failCause(entry.queue as any, exit.cause)) as Effect.Effect<void>
308 } else {
309 entry.resume(exit)
310 }
311 }
312 })
313
314 const onRequest = (rpc: Rpc.AnyWithProps) => {
315 const isStream = RpcSchema.isStreamSchema(rpc.successSchema)
316 const middleware = getRpcClientMiddleware(rpc)
317 return (payload: any, opts?: {
318 readonly asQueue?: boolean | undefined
319 readonly streamBufferSize?: number | undefined
320 readonly headers?: Headers.Input | undefined
321 readonly context?: Context.Context<never> | undefined
322 readonly discard?: boolean | undefined
323 }) => {
324 const headers = opts?.headers ? Headers.fromInput(opts.headers) : Headers.empty
325 const context = opts?.context ?? Context.empty()
326 if (!isStream) {
327 const onRequest = (span: Span | undefined) =>
328 onEffectRequest(
329 rpc,
330 middleware,
331 span,
332 rpc.payloadSchema.make(payload),
333 headers,
334 context,
335 opts?.discard ?? false
336 )
337 return disableTracing ? onRequest(undefined) : Effect.useSpan(
338 `${spanPrefix}.${rpc._tag}`,
339 { attributes: options.spanAttributes },
340 onRequest
341 )
342 }
343 const queue = onStreamRequest(
344 rpc,
345 middleware,
346 rpc.payloadSchema.make(payload),
347 headers,
348 opts?.streamBufferSize ?? 16,
349 context
350 )
351 if (opts?.asQueue) return queue
352 return Stream.unwrap(Effect.map(queue, Stream.fromQueue))
353 }
354 }
355
356 const onEffectRequest = (
357 rpc: Rpc.AnyWithProps,
358 middleware: (
359 send: (request: Request<Rpcs>) => Effect.Effect<any, E>,
360 request: Request<Rpcs>
361 ) => Effect.Effect<any, E>,
362 span: Span | undefined,
363 payload: any,
364 headers: Headers.Headers,
365 context: Context.Context<never>,
366 discard: boolean
367 ) =>
368 Effect.withFiber<any, any, any>((parentFiber) => {
369 if (isShutdown) {
370 return Effect.interrupt
371 }
372 const id = generateRequestId()
373 const send = middleware(
374 (message) =>
375 options.onFromClient({
376 message,
377 context,
378 discard
379 }),
380 {
381 _tag: "Request",
382 id,
383 tag: rpc._tag as Rpc.Tag<Rpcs>,
384 payload,
385 ...(span ?
386 {
387 traceId: span.traceId,
388 spanId: span.spanId,
389 sampled: span.sampled
390 } :
391 {}),
392 headers: Headers.merge(parentFiber.getRef(CurrentHeaders), headers)
393 }
394 )
395 if (discard) {
396 return send
397 }
398 let fiber: Fiber.Fiber<any, any>
399 return Effect.onInterrupt(
400 Effect.callback<any, any>((resume) => {
401 const entry: ClientEntry = {
402 _tag: "Effect",
403 rpc,
404 context,
405 resume(exit) {
406 resume(exit)
407 if (fiber && !fiber.pollUnsafe()) {
408 parentFiber.currentDispatcher.scheduleTask(() => {
409 fiber.interruptUnsafe(parentFiber.id)
410 }, 0)
411 }
412 }
413 }
414 entries.set(id, entry)
415 fiber = send.pipe(
416 span ? Effect.withParentSpan(span, { captureStackTrace: false }) : identity,
417 Effect.runForkWith(parentFiber.context)
418 )
419 fiber.addObserver((exit) => {
420 if (exit._tag === "Failure") {
421 return resume(exit)
422 }
423 })
424 }),
425 (interruptors) => {
426 entries.delete(id)
427 return Effect.andThen(
428 Fiber.interrupt(fiber),
429 sendInterrupt(id, Array.from(interruptors), context)
430 )
431 }
432 )
433 })
434
435 const onStreamRequest = Effect.fnUntraced(function*(
436 rpc: Rpc.AnyWithProps,
437 middleware: (
438 send: (request: Request<Rpcs>) => Effect.Effect<any, E>,
439 request: Request<Rpcs>
440 ) => Effect.Effect<any, E>,
441 payload: any,
442 headers: Headers.Headers,
443 streamBufferSize: number,
444 context: Context.Context<never>
445 ) {
446 if (isShutdown) {
447 return yield* Effect.interrupt
448 }
449
450 const span = disableTracing ? undefined : yield* Effect.makeSpanScoped(`${spanPrefix}.${rpc._tag}`, {
451 attributes: options.spanAttributes
452 })
453 const fiber = Fiber.getCurrent()!
454 const id = generateRequestId()
455
456 const scope = Context.getUnsafe(fiber.context, Scope.Scope)
457 const queue = yield* Queue.bounded<any, any>(streamBufferSize)
458 yield* Scope.addFinalizerExit(
459 scope,
460 (exit) => {
461 Queue.shutdownUnsafe(queue)
462 if (!entries.has(id)) return Effect.void
463 entries.delete(id)
464 return sendInterrupt(
465 id,
466 Exit.isFailure(exit) ?
467 Array.from(Cause.interruptors(exit.cause))
468 : [],
469 context
470 )
471 }
472 )
473
474 entries.set(id, {
475 _tag: "Queue",
476 rpc,
477 queue,
478 scope,
479 context
480 })
481
482 yield* middleware(
483 (message) =>
484 options.onFromClient({
485 message,
486 context,
487 discard: false
488 }),
489 {
490 _tag: "Request",
491 id,
492 tag: rpc._tag as Rpc.Tag<Rpcs>,
493 payload,
494 ...(span ?
495 {
496 traceId: span.traceId,
497 spanId: span.spanId,
498 sampled: span.sampled
499 } :
500 {}),
501 headers: Headers.merge(fiber.getRef(CurrentHeaders), headers)
502 }
503 ).pipe(
504 span ? Effect.withParentSpan(span, { captureStackTrace: false }) : identity,
505 Effect.onError((cause) => Queue.failCause(queue, cause)),
506 Effect.interruptible,
507 Effect.forkIn(scope, { startImmediately: true })
508 )
509
510 return queue
511 })
512
513 const getRpcClientMiddleware = (
514 rpc: Rpc.AnyWithProps
515 ): (
516 send: (request: Request<Rpcs>) => Effect.Effect<any, E>,
517 request: Request<Rpcs>
518 ) => Effect.Effect<any, E> => {
519 const middlewares: Array<RpcMiddleware.RpcMiddlewareClient<any, any, any>> = []
520 for (const tag of rpc.middlewares.values()) {
521 const middleware = Context.getOrUndefinedUnsafe(services, `${tag.key}/Client`) as any
522 if (!middleware) continue
523 middlewares.push(middleware)
524 }
525
526 if (middlewares.length === 0) {
527 return (send, request) => send(request)
528 }
529
530 return function loop(
531 send: (request: Request<Rpcs>) => Effect.Effect<any, E>,
532 request: Request<Rpcs>,
533 index = middlewares.length - 1
534 ): Effect.Effect<any, E> {
535 if (index === -1) {
536 return send(request)
537 }
538 return middlewares[index]({
539 rpc,
540 request,
541 next(request) {
542 return loop(send, request, index - 1) as any
543 }
544 }) as Effect.Effect<any, E>
545 }
546 }
547
548 const sendInterrupt = (
549 requestId: RequestId,
550 interruptors: ReadonlyArray<number>,
551 context: Context.Context<never>
552 ): Effect.Effect<void> =>
553 Effect.callback<void>((resume) => {
554 const parentFiber = Fiber.getCurrent()!
555 const fiber = options.onFromClient({
556 message: { _tag: "Interrupt", requestId, interruptors },
557 context,
558 discard: false
559 }).pipe(
560 Effect.timeout(1000),
561 Effect.runForkWith(parentFiber.context)
562 )
563 fiber.addObserver(() => {
564 resume(Effect.void)
565 })
566 })
567
568 const write = (message: FromServer<Rpcs>): Effect.Effect<void> => {
569 switch (message._tag) {
570 case "Chunk": {
571 const requestId = message.requestId
572 const entry = entries.get(requestId)
573 if (!entry || entry._tag !== "Queue") return Effect.void
574 return Queue.offerAll(entry.queue, message.values).pipe(
575 supportsAck
576 ? Effect.flatMap(() =>
577 options.onFromClient({
578 message: { _tag: "Ack", requestId: message.requestId },
579 context: entry.context,
580 discard: false
581 })
582 )
583 : identity,
584 Effect.catchCause((cause) => Queue.failCause(entry.queue, cause))
585 )
586 }
587 case "Exit": {
588 const requestId = message.requestId
589 const entry = entries.get(requestId)
590 if (!entry) return Effect.void
591 entries.delete(requestId)
592 if (entry._tag === "Effect") {
593 entry.resume(message.exit)
594 return Effect.void
595 }
596 return message.exit._tag === "Success"
597 ? Queue.end(entry.queue)
598 : Queue.failCause(entry.queue, message.exit.cause)
599 }
600 case "Defect": {
601 return clearEntries(Exit.die(message.defect))
602 }
603 case "ClientEnd": {
604 return Effect.void
605 }
606 }
607 }
608
609 let client: any
610 if (options.flatten) {
611 const fns = new Map<string, any>()
612 client = function client(tag: string, payload: any, options?: {}) {
613 let fn = fns.get(tag)
614 if (!fn) {
615 fn = onRequest(group.requests.get(tag)! as any)
616 fns.set(tag, fn)
617 }
618 return fn(payload, options)
619 }
620 } else {
621 client = {}
622 group.requests.forEach((rpc) => {
623 InternalRecord.assignProperty(client, rpc._tag, onRequest(rpc as any))
624 })
625 }
626
627 return { client, write } as const
628})
629
630let clientIdCounter = 0
631
632/**
633 * Creates a schema-aware RPC client for a group using the current client
634 * `Protocol`, encoding requests and decoding server responses.
635 *
636 * @stability unstable
637 * @category constructors
638 * @since 4.0.0
639 */
640export const make: <Rpcs extends Rpc.Any, const Flatten extends boolean = false>(
641 group: RpcGroup.RpcGroup<Rpcs>,
642 options?: {
643 readonly spanPrefix?: string | undefined
644 readonly spanAttributes?: Record<string, unknown> | undefined
645 readonly generateRequestId?: (() => RequestId) | undefined
646 readonly disableTracing?: boolean | undefined
647 readonly flatten?: Flatten | undefined
648 } | undefined
649) => Effect.Effect<
650 Flatten extends true ? RpcClient.Flat<Rpcs, RpcClientError> : RpcClient<Rpcs, RpcClientError>,
651 never,
652 Protocol | Rpc.MiddlewareClient<Rpcs> | Scope.Scope
653> = Effect.fnUntraced(function*<Rpcs extends Rpc.Any, const Flatten extends boolean = false>(
654 group: RpcGroup.RpcGroup<Rpcs>,
655 options?: {
656 readonly spanPrefix?: string | undefined
657 readonly spanAttributes?: Record<string, unknown> | undefined
658 readonly generateRequestId?: (() => RequestId) | undefined
659 readonly disableTracing?: boolean | undefined
660 readonly flatten?: Flatten | undefined
661 } | undefined
662) {
663 const clientId = clientIdCounter++
664 const { codecFor, run, send, supportsAck, supportsTransferables } = yield* Protocol
665 const rpcSchemas = makeRpcSchemas(codecFor)
666 const decodeDefect = Schema.decodeSync(codecFor(Schema.Defect()))
667
668 type ClientEntry = {
669 readonly rpc: Rpc.AnyWithProps
670 readonly context: Context.Context<never>
671 readonly schemas: RpcSchemas
672 }
673 const entries = new Map<RequestId, ClientEntry>()
674
675 const interruptRequest = (requestId: RequestId): Effect.Effect<void, RpcClientError> => {
676 if (!entries.has(requestId)) return Effect.void
677 entries.delete(requestId)
678 return send(clientId, { _tag: "Interrupt", requestId }) as Effect.Effect<void, RpcClientError>
679 }
680
681 const { client, write } = yield* makeNoSerialization(group, {
682 ...options,
683 supportsAck,
684 onFromClient({ message }) {
685 switch (message._tag) {
686 case "Request": {
687 const rpc = group.requests.get(message.tag)! as any as Rpc.AnyWithProps
688 const collector = supportsTransferables ? Transferable.makeCollectorUnsafe() : undefined
689
690 const fiber = Fiber.getCurrent()!
691
692 const entry: ClientEntry = {
693 rpc,
694 context: collector ? Context.add(fiber.context, Transferable.Collector, collector) : fiber.context,
695 schemas: rpcSchemas(rpc)
696 }
697 entries.set(message.id, entry)
698
699 return entry.schemas.encodePayload(message.payload).pipe(
700 Effect.provideContext(entry.context),
701 Effect.orDie,
702 Effect.flatMap((payload) =>
703 send(clientId, {
704 ...message,
705 id: message.id,
706 payload,
707 headers: Object.entries(message.headers)
708 }, collector && collector.readUnsafe())
709 )
710 ) as Effect.Effect<void, RpcClientError>
711 }
712 case "Ack": {
713 const entry = entries.get(message.requestId)
714 if (!entry) return Effect.void
715 return send(clientId, {
716 _tag: "Ack",
717 requestId: message.requestId
718 }) as Effect.Effect<void, RpcClientError>
719 }
720 case "Interrupt": {
721 return interruptRequest(message.requestId)
722 }
723 case "Eof": {
724 return Effect.void
725 }
726 }
727 }
728 })
729
730 yield* run(clientId, (message) => {
731 switch (message._tag) {
732 case "Chunk": {
733 const requestId = RequestId(message.requestId)
734 const entry = entries.get(requestId)
735 if (!entry || Option.isNone(entry.schemas.decodeChunk)) return Effect.void
736 const decodeChunk = entry.schemas.decodeChunk.value
737 return Effect.uninterruptibleMask((restore) =>
738 restore(
739 decodeChunk(message.values).pipe(
740 Effect.provideContext(entry.context),
741 Effect.orDie,
742 Effect.flatMap((chunk) => write({ _tag: "Chunk", clientId: 0, requestId, values: chunk }))
743 )
744 ).pipe(
745 Effect.onError((cause) =>
746 write({ _tag: "Exit", clientId: 0, requestId, exit: Exit.failCause(cause) }).pipe(
747 Effect.andThen(Cause.hasInterrupts(cause) ? Effect.void : Effect.ignore(interruptRequest(requestId)))
748 )
749 ),
750 // a decode failure only ends its own request; interruption still propagates
751 Effect.catchCauseIf((cause) => !Cause.hasInterrupts(cause), () => Effect.void)
752 )
753 ) as Effect.Effect<void>
754 }
755 case "Exit": {
756 const requestId = RequestId(message.requestId)
757 const entry = entries.get(requestId)
758 if (!entry) return Effect.void
759 entries.delete(requestId)
760 return entry.schemas.decodeExit(message.exit).pipe(
761 Effect.provideContext(entry.context),
762 Effect.orDie,
763 Effect.matchCauseEffect({
764 onSuccess: (exit) => write({ _tag: "Exit", clientId: 0, requestId, exit }),
765 onFailure: (cause) => write({ _tag: "Exit", clientId: 0, requestId, exit: Exit.failCause(cause) })
766 })
767 ) as Effect.Effect<void>
768 }
769 case "Defect": {
770 return write({ _tag: "Defect", clientId: 0, defect: decodeDefect(message.defect as any) })
771 }
772 case "ClientProtocolError": {
773 const exit = Exit.fail(message.error)
774 return Effect.forEach(
775 Array.from(entries.keys()),
776 (requestId) => write({ _tag: "Exit", clientId: 0, requestId, exit: exit as any })
777 )
778 }
779 default: {
780 return Effect.void
781 }
782 }
783 }).pipe(
784 Effect.catchCause(Effect.logError),
785 Effect.interruptible,
786 Effect.forkScoped
787 )
788
789 return client
790})
791
792interface RpcSchemas {
793 readonly decodeChunk: Option.Option<
794 (chunk: ReadonlyArray<unknown>) => Effect.Effect<NonEmptyReadonlyArray<any>, Schema.SchemaError, unknown>
795 >
796 readonly encodePayload: (payload: any) => Effect.Effect<any, Schema.SchemaError, unknown>
797 readonly decodeExit: (encoded: unknown) => Effect.Effect<Exit.Exit<any, any>, Schema.SchemaError, unknown>
798}
799// Codecs are compiled per client, because two protocols can fill the message
800// holes with different codecs.
801const makeRpcSchemas = (codecFor: RpcSerialization.CodecFor) => {
802 const cache = new WeakMap<Rpc.AnyWithProps, RpcSchemas>()
803 return (rpc: Rpc.AnyWithProps): RpcSchemas => {
804 let entry = cache.get(rpc)
805 if (entry !== undefined) {
806 return entry
807 }
808 const streamSchemas = RpcSchema.getStreamSchemas(rpc.successSchema)
809 entry = {
810 decodeChunk: Option.map(
811 streamSchemas,
812 (streamSchemas) => Schema.decodeUnknownEffect(codecFor(Schema.NonEmptyArray(streamSchemas.success)))
813 ),
814 encodePayload: Schema.encodeEffect(codecFor(rpc.payloadSchema)),
815 decodeExit: Schema.decodeUnknownEffect(codecFor(Rpc.exitSchema(rpc as any)))
816 }
817 cache.set(rpc, entry)
818 return entry
819 }
820}
821
822/**
823 * Fiber reference containing headers that are merged into outgoing RPC
824 * client requests.
825 *
826 * **When to use**
827 *
828 * Use to set request headers that should be automatically merged into outgoing
829 * RPC client messages.
830 *
831 * @stability unstable
832 * @category services
833 * @since 4.0.0
834 */
835export const CurrentHeaders = Context.Reference<Headers.Headers>("effect/rpc/RpcClient/CurrentHeaders", {
836 defaultValue: () => Headers.empty
837})
838
839/**
840 * Runs an effect with additional RPC client headers, merging them with the
841 * current `CurrentHeaders` value for outgoing requests.
842 *
843 * @stability unstable
844 * @category headers
845 * @since 4.0.0
846 */
847export const withHeaders: {
848 (headers: Headers.Input): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
849 <A, E, R>(effect: Effect.Effect<A, E, R>, headers: Headers.Input): Effect.Effect<A, E, R>
850} = dual(
851 2,
852 <A, E, R>(effect: Effect.Effect<A, E, R>, headers: Headers.Input): Effect.Effect<A, E, R> =>
853 Effect.updateService(effect, CurrentHeaders, Headers.merge(Headers.fromInput(headers)))
854)
855
856/**
857 * Defines the service interface for an RPC client transport, responsible for running the
858 * receive loop and sending encoded client messages.
859 *
860 * **When to use**
861 *
862 * Use to provide the transport boundary for RPC clients over HTTP, WebSocket,
863 * workers, sockets, or custom protocols.
864 *
865 * @stability unstable
866 * @category services
867 * @since 4.0.0
868 */
869export class Protocol extends Context.Service<Protocol, {
870 readonly run: (
871 clientId: number,
872 f: (data: FromServerEncoded) => Effect.Effect<void>
873 ) => Effect.Effect<never>
874 readonly send: (
875 clientId: number,
876 request: FromClientEncoded,
877 transferables?: ReadonlyArray<globalThis.Transferable>
878 ) => Effect.Effect<void, RpcClientError>
879 readonly supportsAck: boolean
880 readonly supportsTransferables: boolean
881 /**
882 * Builds the codec that fills the `unknown` holes of the protocol messages,
883 * re-passed from the `RpcSerialization` backing this transport.
884 */
885 readonly codecFor: RpcSerialization.CodecFor
886}>()("effect/rpc/RpcClient/Protocol") {
887 /**
888 * Creates a client protocol service from the supplied RPC request runner.
889 *
890 * @stability unstable
891 * @since 4.0.0
892 */
893 static make = withRunClient
894}
895
896/**
897 * Creates a client `Protocol` that sends each RPC request through the supplied
898 * `HttpClient` and decodes responses with the current `RpcSerialization`.
899 *
900 * @stability unstable
901 * @category protocols
902 * @since 4.0.0
903 */
904export const makeProtocolHttp = (client: HttpClient.HttpClient): Effect.Effect<
905 Protocol["Service"],
906 never,
907 RpcSerialization.RpcSerialization
908> =>
909 Protocol.make(Effect.fnUntraced(function*(writeResponse) {
910 const serialization = yield* RpcSerialization.RpcSerialization
911 const isFramed = serialization.includesFraming
912 const httpClientError = (cause: any) =>
913 new RpcClientError({
914 reason: HttpClientErrorSchema.fromHttpClientError(cause)
915 })
916 const protocolDefect = (message: string, cause: unknown) =>
917 new RpcClientError({
918 reason: new RpcClientDefect({ message, cause })
919 })
920 const emptyResponseError = (request: FromClientEncoded) =>
921 protocolDefect("Received empty HTTP response from RPC server", request)
922 const incompleteResponseError = (request: FromClientEncoded) =>
923 protocolDefect("HTTP response ended before RPC request completed", request)
924
925 const send = Effect.fnUntraced(function*(clientId: number, request: FromClientEncoded) {
926 if (request._tag !== "Request") {
927 return
928 }
929
930 const parser = serialization.makeUnsafe()
931
932 const encoded = parser.encode(request)!
933 const body = typeof encoded === "string" ?
934 HttpBody.text(encoded, serialization.contentType) :
935 HttpBody.uint8Array(encoded, serialization.contentType)
936
937 const response = yield* client.post("", { body }).pipe(Effect.mapError(httpClientError))
938
939 if (!isFramed) {
940 const text = yield* response.text.pipe(Effect.mapError(httpClientError))
941 const responses = yield* Effect.try({
942 try: () => parser.decode(text),
943 catch: (cause) => protocolDefect("Error decoding HTTP response", cause)
944 })
945 if (!Array.isArray(responses)) {
946 return yield* protocolDefect("Expected an array of responses", responses)
947 }
948 if (responses.length === 0) {
949 return yield* emptyResponseError(request)
950 }
951 let completed = false
952 for (let i = 0; i < responses.length; i++) {
953 const response = responses[i]
954 if (isTerminalResponse(response)) {
955 completed = true
956 }
957 yield* writeResponse(clientId, response)
958 }
959 if (!completed) {
960 return yield* incompleteResponseError(request)
961 }
962 return
963 }
964
965 let hasResponse = false
966 let completed = false
967 yield* Stream.runForEachArray(response.stream, (chunk) =>
968 Effect.try({
969 try: () => chunk.flatMap(parser.decode) as Array<FromServerEncoded>,
970 catch: (cause) => protocolDefect("Error decoding HTTP response", cause)
971 }).pipe(
972 Effect.flatMap((responses) => {
973 if (responses.length === 0) return Effect.void
974 hasResponse = true
975 let i = 0
976 return Effect.whileLoop({
977 while: () => i < responses.length,
978 body: () => {
979 const response = responses[i++]
980 if (isTerminalResponse(response)) {
981 completed = true
982 }
983 return writeResponse(clientId, response)
984 },
985 step: constVoid
986 })
987 })
988 )).pipe(
989 Effect.mapError((cause) => isRpcClientError(cause) ? cause : httpClientError(cause))
990 )
991 if (!hasResponse) {
992 return yield* emptyResponseError(request)
993 } else if (!completed) {
994 return yield* incompleteResponseError(request)
995 }
996 })
997
998 return {
999 send,
1000 supportsAck: false,
1001 supportsTransferables: false,
1002 codecFor: serialization.codecFor
1003 }
1004 }))
1005
1006/**
1007 * Provides a client `Protocol` backed by `HttpClient`, targeting the configured
1008 * URL and optionally transforming the client before use.
1009 *
1010 * @stability unstable
1011 * @category layers
1012 * @since 4.0.0
1013 */
1014export const layerProtocolHttp = (options: {
1015 readonly url: string
1016 readonly transformClient?: <E, R>(client: HttpClient.HttpClient.With<E, R>) => HttpClient.HttpClient.With<E, R>
1017}): Layer.Layer<Protocol, never, RpcSerialization.RpcSerialization | HttpClient.HttpClient> =>
1018 Layer.effect(Protocol)(
1019 Effect.flatMap(
1020 HttpClient.HttpClient,
1021 (client) => {
1022 client = HttpClient.mapRequest(client, HttpClientRequest.prependUrl(options.url))
1023 return makeProtocolHttp(options.transformClient ? options.transformClient(client) : client)
1024 }
1025 )
1026 )
1027
1028/**
1029 * Creates a client `Protocol` over the current `Socket`, using the current
1030 * `RpcSerialization`, connection hooks, ping timeouts, and the configured retry
1031 * policy.
1032 *
1033 * @stability unstable
1034 * @category protocols
1035 * @since 4.0.0
1036 */
1037export const makeProtocolSocket = (options?: {
1038 readonly retryTransientErrors?: boolean | undefined
1039 readonly retryPolicy?: Schedule.Schedule<any, Socket.SocketError> | undefined
1040 /**
1041 * Runs for each retried `SocketOpenError` when `retryTransientErrors` is enabled.
1042 * A missed pong fails in-flight calls and is not reported through this hook.
1043 * The returned `Effect<void>` cannot fail with a typed error or require
1044 * services; defects are logged and ignored so retries can continue.
1045 */
1046 readonly onTransientError?: ((error: RpcClientError) => Effect.Effect<void>) | undefined
1047}): Effect.Effect<
1048 Protocol["Service"],
1049 never,
1050 Scope.Scope | RpcSerialization.RpcSerialization | Socket.Socket
1051> =>
1052 Protocol.make(Effect.fnUntraced(function*(writeResponse, clientIds) {
1053 const socket = yield* Socket.Socket
1054 const serialization = yield* RpcSerialization.RpcSerialization
1055 const hooks = yield* Effect.serviceOption(ConnectionHooks)
1056 const requestClientMap = new Map<string | number, number>()
1057
1058 const writer = yield* socket.writer
1059
1060 let parser = serialization.makeUnsafe()
1061
1062 // `parser` is replaced on every connect, and a stateful serialization
1063 // encodes against the connection it is writing to, so the ping is encoded
1064 // when it is sent rather than once up front.
1065 const pinger = yield* makePinger(Effect.suspend(() => writer.write(parser.encode(constPing)!)))
1066 let currentError: RpcClientError | undefined
1067
1068 const broadcast = (response: FromServerEncoded) =>
1069 Effect.forEach(clientIds, (clientId) => writeResponse(clientId, response))
1070 const broadcastError = (error: RpcClientError) => {
1071 currentError = error
1072 return broadcast({
1073 _tag: "ClientProtocolError",
1074 error
1075 })
1076 }
1077
1078 const processData = (data: Uint8Array | string): Effect.Effect<void> => {
1079 try {
1080 const responses = parser.decode(data) as Array<FromServerEncoded>
1081 if (responses.length === 0) return Effect.void
1082 let i = 0
1083 return Effect.whileLoop({
1084 while: () => i < responses.length,
1085 body: () => {
1086 const response = responses[i++]
1087 if (response._tag === "Pong") {
1088 pinger.onPong()
1089 return Effect.void
1090 }
1091 if (Object.hasOwn(response, "requestId")) {
1092 const requestId = (response as FromServerEncoded & { readonly requestId: string | number }).requestId
1093 const clientId = requestClientMap.get(requestId)
1094 if (clientId !== undefined) {
1095 if (response._tag === "Exit") {
1096 requestClientMap.delete(requestId)
1097 }
1098 return writeResponse(clientId, response)
1099 }
1100 }
1101 return broadcast(response)
1102 },
1103 step: constVoid
1104 })
1105 } catch (defect) {
1106 return broadcast({
1107 _tag: "ClientProtocolError",
1108 error: new RpcClientError({
1109 reason: new RpcClientDefect({
1110 message: "Error decoding message",
1111 cause: defect
1112 })
1113 })
1114 })
1115 }
1116 }
1117
1118 yield* Effect.suspend(() => {
1119 parser = serialization.makeUnsafe()
1120 pinger.reset()
1121 return Effect.gen(function*() {
1122 const { pull } = yield* socket.reader
1123 currentError = undefined
1124 if (Option.isSome(hooks)) {
1125 yield* hooks.value.onConnect
1126 }
1127 while (true) {
1128 const frames = yield* pull
1129 for (let i = 0; i < frames.length; i++) {
1130 yield* processData(frames[i])
1131 }
1132 }
1133 }).pipe(
1134 Effect.scoped,
1135 Effect.raceFirst(Effect.flatMap(
1136 pinger.timeout,
1137 () =>
1138 Effect.fail(
1139 new Socket.SocketError({
1140 reason: new Socket.SocketReadError({
1141 cause: new Error("ping timeout")
1142 })
1143 })
1144 )
1145 ))
1146 )
1147 }).pipe(
1148 Option.isSome(hooks) ? Effect.ensuring(hooks.value.onDisconnect) : identity,
1149 Effect.tapCause((cause) => {
1150 const error = Cause.findError(cause)
1151 const hasError = Result.isSuccess(error)
1152 const rpcError = new RpcClientError({
1153 reason: hasError ? error.success.reason : new RpcClientDefect({
1154 message: "Unknown socket error",
1155 cause: Cause.squash(cause)
1156 })
1157 })
1158 if (
1159 options?.retryTransientErrors && hasError &&
1160 error.success.reason._tag === "SocketOpenError"
1161 ) {
1162 return (options.onTransientError?.(rpcError) ?? Effect.void).pipe(
1163 Effect.ignoreCause({
1164 log: true,
1165 message: "RpcClient onTransientError hook failed"
1166 })
1167 )
1168 }
1169 return broadcastError(rpcError)
1170 }),
1171 Effect.retryOrElse(
1172 options?.retryPolicy ?? defaultRetryPolicy,
1173 (error) => broadcastError(new RpcClientError({ reason: error.reason }))
1174 ),
1175 Effect.annotateLogs({
1176 module: "RpcClient",
1177 method: "makeProtocolSocket"
1178 }),
1179 Effect.forkScoped
1180 )
1181
1182 return {
1183 send(clientId, request) {
1184 if (currentError) {
1185 return Effect.fail(currentError)
1186 }
1187 if (request._tag === "Request") {
1188 requestClientMap.set(request.id, clientId)
1189 }
1190 const encoded = parser.encode(request)
1191 if (encoded === undefined) return Effect.void
1192 return Effect.orDie(writer.write(encoded))
1193 },
1194 supportsAck: true,
1195 supportsTransferables: false,
1196 codecFor: serialization.codecFor
1197 }
1198 }))
1199
1200const defaultRetryPolicy = Schedule.min([
1201 Schedule.exponential(500, 1.5),
1202 Schedule.spaced(5000)
1203])
1204
1205const makePinger = Effect.fnUntraced(function*<A, E, R>(writePing: Effect.Effect<A, E, R>) {
1206 let recievedPong = true
1207 const latch = Latch.makeUnsafe()
1208 const reset = () => {
1209 recievedPong = true
1210 latch.closeUnsafe()
1211 }
1212 const onPong = () => {
1213 recievedPong = true
1214 }
1215 yield* Effect.suspend((): Effect.Effect<void, E, R> => {
1216 if (!recievedPong) return latch.open
1217 recievedPong = false
1218 return writePing
1219 }).pipe(
1220 Effect.delay("5 seconds"),
1221 Effect.ignore,
1222 Effect.forever,
1223 Effect.interruptible,
1224 Effect.forkScoped
1225 )
1226 return { timeout: latch.await, reset, onPong } as const
1227})
1228
1229/**
1230 * Provides a client `Protocol` backed by the current `Socket` and
1231 * `RpcSerialization` services.
1232 *
1233 * @stability unstable
1234 * @category layers
1235 * @since 4.0.0
1236 */
1237export const layerProtocolSocket = (options?: {
1238 readonly retryTransientErrors?: boolean | undefined
1239 /**
1240 * Runs for each retried `SocketOpenError` when `retryTransientErrors` is enabled.
1241 * A missed pong fails in-flight calls and is not reported through this hook.
1242 * The returned `Effect<void>` cannot fail with a typed error or require
1243 * services; defects are logged and ignored so retries can continue.
1244 */
1245 readonly onTransientError?: ((error: RpcClientError) => Effect.Effect<void>) | undefined
1246}): Layer.Layer<
1247 Protocol,
1248 never,
1249 Socket.Socket | RpcSerialization.RpcSerialization
1250> => Layer.effect(Protocol)(makeProtocolSocket(options))
1251
1252/**
1253 * Creates a client `Protocol` backed by a pool of workers, routing RPC requests
1254 * to workers and supporting transferable values when the platform does.
1255 *
1256 * @stability unstable
1257 * @category protocols
1258 * @since 4.0.0
1259 */
1260export const makeProtocolWorker = (
1261 options: {
1262 readonly size: number
1263 readonly concurrency?: number | undefined
1264 readonly targetUtilization?: number | undefined
1265 } | {
1266 readonly minSize: number
1267 readonly maxSize: number
1268 readonly concurrency?: number | undefined
1269 readonly targetUtilization?: number | undefined
1270 readonly timeToLive: Duration.Input
1271 }
1272): Effect.Effect<
1273 Protocol["Service"],
1274 WorkerError,
1275 Scope.Scope | Worker.WorkerPlatform | Worker.Spawner
1276> =>
1277 Protocol.make(Effect.fnUntraced(function*(writeResponse, clientIds) {
1278 const worker = yield* Worker.WorkerPlatform
1279 const scope = yield* Effect.scope
1280 let workerId = 0
1281 const initialMessage = yield* Effect.serviceOption(RpcWorker.InitialMessage)
1282 const hooks = yield* Effect.serviceOption(ConnectionHooks)
1283
1284 const entries = new Map<string | number, {
1285 readonly clientId: number
1286 readonly worker: Worker.Worker<FromServerEncoded, FromClientEncoded | RpcWorker.InitialMessage.Encoded>
1287 readonly latch: Latch.Latch
1288 }>()
1289
1290 const broadcast = (response: FromServerEncoded) =>
1291 Effect.forEach(clientIds, (clientId) => writeResponse(clientId, response))
1292
1293 const acquire = Effect.gen(function*() {
1294 const id = workerId++
1295 const backing = yield* worker.spawn<FromServerEncoded, FromClientEncoded | RpcWorker.InitialMessage.Encoded>(id)
1296
1297 yield* backing.run((response) => {
1298 if (response._tag === "Exit") {
1299 const entry = entries.get(response.requestId)
1300 if (entry) {
1301 entries.delete(response.requestId)
1302 entry.latch.openUnsafe()
1303 return writeResponse(entry.clientId, response)
1304 }
1305 } else if (response._tag === "Defect") {
1306 for (const [requestId, entry] of entries) {
1307 entries.delete(requestId)
1308 entry.latch.openUnsafe()
1309 }
1310 return broadcast(response)
1311 } else if ("requestId" in response) {
1312 const entry = entries.get(response.requestId)
1313 if (entry) {
1314 return writeResponse(entry.clientId, response)
1315 }
1316 }
1317 return broadcast(response)
1318 }, {
1319 onSpawn: Option.isSome(initialMessage) ?
1320 Effect.flatMap(
1321 initialMessage.value,
1322 ([value, transfers]) => Effect.orDie(backing.send({ _tag: "InitialMessage", value }, transfers))
1323 ) :
1324 undefined
1325 }).pipe(
1326 Effect.tapCause((cause) => {
1327 for (const [requestId, entry] of entries) {
1328 if (entry.worker !== backing) continue
1329 entries.delete(requestId)
1330 entry.latch.openUnsafe()
1331 }
1332 const error = Cause.findError(cause)
1333 return broadcast({
1334 _tag: "ClientProtocolError",
1335 error: new RpcClientError({
1336 reason: Result.isSuccess(error) ? error.success.reason : new RpcClientDefect({
1337 message: "Error in worker",
1338 cause: Cause.squash(cause)
1339 })
1340 })
1341 })
1342 }),
1343 Effect.retry(Schedule.spaced(1000)),
1344 Effect.annotateLogs({
1345 module: "RpcClient",
1346 method: "makeProtocolWorker"
1347 }),
1348 Effect.interruptible,
1349 Effect.forkScoped
1350 )
1351
1352 return backing
1353 })
1354
1355 const pool = "minSize" in options ?
1356 yield* Pool.makeWithTTL({
1357 acquire,
1358 min: options.minSize,
1359 max: options.maxSize,
1360 concurrency: options.concurrency,
1361 targetUtilization: options.targetUtilization,
1362 timeToLive: options.timeToLive
1363 }) :
1364 yield* Pool.make({
1365 acquire,
1366 size: options.size,
1367 concurrency: options.concurrency,
1368 targetUtilization: options.targetUtilization
1369 })
1370
1371 yield* Scope.addFinalizer(
1372 scope,
1373 Effect.sync(() => {
1374 for (const entry of entries.values()) {
1375 entry.latch.openUnsafe()
1376 }
1377 entries.clear()
1378 })
1379 )
1380
1381 const send = (
1382 clientId: number,
1383 request: FromClientEncoded,
1384 transferables?: ReadonlyArray<globalThis.Transferable>
1385 ) => {
1386 switch (request._tag) {
1387 case "Request": {
1388 return Pool.get(pool).pipe(
1389 Effect.flatMap((worker) => {
1390 const latch = Latch.makeUnsafe(false)
1391 entries.set(request.id, { clientId, worker, latch })
1392 return Effect.flatMap(worker.send(request, transferables), () => latch.await)
1393 }),
1394 Effect.scoped,
1395 Effect.orDie
1396 )
1397 }
1398 case "Interrupt": {
1399 const entry = entries.get(request.requestId)
1400 if (!entry) return Effect.void
1401 entries.delete(request.requestId)
1402 entry.latch.openUnsafe()
1403 return Effect.orDie(entry.worker.send(request))
1404 }
1405 case "Ack": {
1406 const entry = entries.get(request.requestId)
1407 if (!entry) return Effect.void
1408 return Effect.orDie(entry.worker.send(request))
1409 }
1410 }
1411 return Effect.void
1412 }
1413
1414 yield* Effect.scoped(Pool.get(pool))
1415 if (Option.isSome(hooks)) yield* hooks.value.onConnect
1416
1417 return {
1418 send,
1419 supportsAck: true,
1420 supportsTransferables: true,
1421 // Worker protocols use structured clone, so they do not depend on
1422 // `RpcSerialization`. A binary worker protocol is a separate protocol.
1423 codecFor: Schema.toCodecJson as RpcSerialization.CodecFor
1424 }
1425 }))
1426
1427/**
1428 * Provides a client `Protocol` backed by a worker pool using the current worker
1429 * platform and spawner services.
1430 *
1431 * @stability unstable
1432 * @category layers
1433 * @since 4.0.0
1434 */
1435export const layerProtocolWorker: (
1436 options: {
1437 readonly size: number
1438 readonly concurrency?: number | undefined
1439 readonly targetUtilization?: number | undefined
1440 } | {
1441 readonly minSize: number
1442 readonly maxSize: number
1443 readonly concurrency?: number | undefined
1444 readonly targetUtilization?: number | undefined
1445 readonly timeToLive: Duration.Input
1446 }
1447) => Layer.Layer<
1448 Protocol,
1449 WorkerError,
1450 Worker.WorkerPlatform | Worker.Spawner
1451> = flow(makeProtocolWorker, Layer.effect(Protocol))
1452
1453/**
1454 * Represents optional client protocol hooks that run when a transport connects
1455 * and disconnects.
1456 *
1457 * **When to use**
1458 *
1459 * Use to run setup or cleanup effects when an RPC client transport opens or
1460 * closes.
1461 *
1462 * @stability unstable
1463 * @category services
1464 * @since 4.0.0
1465 */
1466export class ConnectionHooks extends Context.Service<ConnectionHooks, {
1467 readonly onConnect: Effect.Effect<void>
1468 readonly onDisconnect: Effect.Effect<void>
1469}>()("effect/rpc/RpcClient/ConnectionHooks") {}