1/**
2 * Runs server-side handlers for RPC groups.
3 *
4 * This module connects typed handlers for an `RpcGroup` to a server `Protocol`.
5 * It receives client messages, decodes request payloads, runs matching handlers
6 * and middleware, tracks in-flight requests, handles acknowledgements and
7 * interrupts, and sends responses back to clients. It also provides constructors
8 * and layers for decoded messages, HTTP, WebSocket, sockets, stdio, and worker
9 * runner protocols.
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 * as Deferred from "../Deferred.ts"
18import * as Effect from "../Effect.ts"
19import * as Exit from "../Exit.ts"
20import * as Fiber from "../Fiber.ts"
21import { constant, constTrue, constVoid, identity } from "../Function.ts"
22import * as Headers from "../http/Headers.ts"
23import * as HttpRouter from "../http/HttpRouter.ts"
24import * as HttpServerRequest from "../http/HttpServerRequest.ts"
25import * as HttpServerResponse from "../http/HttpServerResponse.ts"
26import { reportCauseUnsafe } from "../internal/effect.ts"
27import * as Latch from "../Latch.ts"
28import * as Layer from "../Layer.ts"
29import * as Option from "../Option.ts"
30import * as Predicate from "../Predicate.ts"
31import * as Pull from "../Pull.ts"
32import * as Queue from "../Queue.ts"
33import * as Schedule from "../Schedule.ts"
34import * as Schema from "../Schema.ts"
35import * as SchemaIssue from "../SchemaIssue.ts"
36import * as Scope from "../Scope.ts"
37import * as Semaphore from "../Semaphore.ts"
38import * as Socket from "../socket/Socket.ts"
39import * as SocketServer from "../socket/SocketServer.ts"
40import { Stdio } from "../Stdio.ts"
41import * as Stream from "../Stream.ts"
42import * as Tracer from "../Tracer.ts"
43import type * as Types from "../Types.ts"
44import * as Transferable from "../workers/Transferable.ts"
45import type { WorkerError } from "../workers/WorkerError.ts"
46import * as WorkerRunner from "../workers/WorkerRunner.ts"
47import * as Rpc from "./Rpc.ts"
48import type * as RpcGroup from "./RpcGroup.ts"
49import type {
50 FromClient,
51 FromClientEncoded,
52 FromServer,
53 FromServerEncoded,
54 Request,
55 RequestEncoded,
56 ResponseExitEncoded
57} from "./RpcMessage.ts"
58import { constEof, constPong, RequestId, ResponseDefectEncoded } from "./RpcMessage.ts"
59import * as RpcSchema from "./RpcSchema.ts"
60import * as RpcSerialization from "./RpcSerialization.ts"
61import type { InitialMessage } from "./RpcWorker.ts"
62import { withRun } from "./Utils.ts"
63
64/**
65 * The decoded RPC server boundary, accepting client messages for a client id
66 * and allowing that client to be disconnected.
67 *
68 * @stability unstable
69 * @category models
70 * @since 4.0.0
71 */
72export interface RpcServer<A extends Rpc.Any> {
73 readonly write: (clientId: number, message: FromClient<A>, options?: {
74 readonly onRequest?: (<A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>) | undefined
75 }) => Effect.Effect<void>
76 readonly disconnect: (clientId: number) => Effect.Effect<void>
77}
78
79/**
80 * Creates an RPC server for an already-decoded message channel, running
81 * handlers for a group and sending decoded server responses through
82 * `onFromServer`.
83 *
84 * @stability unstable
85 * @category constructors
86 * @since 4.0.0
87 */
88export const makeNoSerialization: <Rpcs extends Rpc.Any>(
89 group: RpcGroup.RpcGroup<Rpcs>,
90 options: {
91 readonly onFromServer: (response: FromServer<Rpcs>) => Effect.Effect<void>
92 readonly disableTracing?: boolean | undefined
93 readonly disableSpanPropagation?: boolean | undefined
94 readonly spanPrefix?: string | undefined
95 readonly spanAttributes?: Record<string, unknown> | undefined
96 readonly disableClientAcks?: boolean | undefined
97 readonly concurrency?: number | "unbounded" | undefined
98 readonly disableFatalDefects?: boolean | undefined
99 }
100) => Effect.Effect<
101 RpcServer<Rpcs>,
102 never,
103 Rpc.ToHandler<Rpcs> | Rpc.Middleware<Rpcs> | Scope.Scope
104> = Effect.fnUntraced(function*<Rpcs extends Rpc.Any>(
105 group: RpcGroup.RpcGroup<Rpcs>,
106 options: {
107 readonly onFromServer: (response: FromServer<Rpcs>) => Effect.Effect<void>
108 readonly disableTracing?: boolean | undefined
109 readonly disableSpanPropagation?: boolean | undefined
110 readonly spanPrefix?: string | undefined
111 readonly spanAttributes?: Record<string, unknown> | undefined
112 readonly disableClientAcks?: boolean | undefined
113 readonly concurrency?: number | "unbounded" | undefined
114 readonly disableFatalDefects?: boolean | undefined
115 }
116) {
117 const enableTracing = options.disableTracing !== true
118 const enableSpanPropagation = options.disableSpanPropagation !== true
119 const supportsAck = options.disableClientAcks !== true
120 const spanPrefix = options.spanPrefix ?? "RpcServer"
121 const concurrency = options.concurrency ?? "unbounded"
122 const disableFatalDefects = options.disableFatalDefects ?? false
123 const services = yield* Effect.context<Rpc.ToHandler<Rpcs> | Scope.Scope>()
124 const scope = Context.get(services, Scope.Scope)
125 const trackFiber = Fiber.runIn(Scope.forkUnsafe(scope, "parallel"))
126 const concurrencySemaphore = concurrency === "unbounded"
127 ? undefined
128 : yield* Semaphore.make(concurrency)
129
130 type Client = {
131 readonly id: number
132 // lazily created: only streaming requests with client acks use latches
133 latches: Map<RequestId, Latch.Latch> | undefined
134 readonly fibers: Map<RequestId, Fiber.Fiber<unknown, any>>
135 readonly serverClient: Rpc.ServerClient
136 ended: boolean
137 }
138
139 const clients = new Map<number, Client>()
140 let isShutdown = false
141 const shutdownLatch = Latch.makeUnsafe(false)
142 yield* Scope.addFinalizer(
143 scope,
144 Effect.withFiber((parent) => {
145 isShutdown = true
146 for (const client of clients.values()) {
147 client.ended = true
148 if (client.fibers.size === 0) {
149 trackFiber(Effect.runForkWith(services)(endClient(client)))
150 continue
151 }
152 for (const fiber of client.fibers.values()) {
153 fiber.interruptUnsafe(parent.id)
154 }
155 }
156 if (clients.size === 0) {
157 return Effect.void
158 }
159 return shutdownLatch.await
160 })
161 )
162
163 const disconnect = (clientId: number) =>
164 Effect.withFiber((parent) => {
165 const client = clients.get(clientId)
166 if (!client) return Effect.void
167 for (const fiber of client.fibers.values()) {
168 fiber.interruptUnsafe(parent.id)
169 }
170 clients.delete(clientId)
171 return Effect.void
172 })
173
174 const write: RpcServer<Rpcs>["write"] = (clientId, message, opts) =>
175 Effect.catchDefect(
176 Effect.withFiber((requestFiber) => {
177 if (isShutdown) return Effect.interrupt
178 let client = clients.get(clientId)
179 if (!client) {
180 client = {
181 id: clientId,
182 latches: undefined,
183 fibers: new Map(),
184 ended: false,
185 serverClient: new Rpc.ServerClient(clientId)
186 }
187 clients.set(clientId, client)
188 } else if (
189 client.ended &&
190 (message._tag !== "Interrupt" || !client.fibers.has(message.requestId))
191 ) {
192 return Effect.interrupt
193 }
194
195 switch (message._tag) {
196 case "Request": {
197 return handleRequest(requestFiber, client, message, opts)
198 }
199 case "Ack": {
200 const latch = client.latches?.get(message.requestId)
201 return latch ? latch.open : Effect.void
202 }
203 case "Interrupt": {
204 const fiber = client.fibers.get(message.requestId)
205 if (fiber) {
206 fiber.interruptUnsafe(requestFiber.id, RpcSchema.ClientAbort.annotation)
207 return Effect.void
208 }
209 return options.onFromServer({
210 _tag: "Exit",
211 clientId,
212 requestId: message.requestId,
213 exit: Exit.interrupt()
214 })
215 }
216 case "Eof": {
217 client.ended = true
218 if (client.fibers.size > 0) return Effect.void
219 return endClient(client)
220 }
221 default: {
222 return sendDefect(client, `Unknown request tag: ${(message as any)._tag}`)
223 }
224 }
225 }),
226 (defect) => sendDefect(clients.get(clientId)!, defect)
227 )
228
229 const endClient = (client: Client) => {
230 clients.delete(client.id)
231 const write = options.onFromServer({
232 _tag: "ClientEnd",
233 clientId: client.id
234 })
235 if (isShutdown && clients.size === 0) {
236 return Effect.andThen(write, shutdownLatch.open)
237 }
238 return write
239 }
240
241 const handleRequest = (
242 requestFiber: Fiber.Fiber<any, any>,
243 client: Client,
244 request: Request<Rpcs>,
245 opts: Parameters<RpcServer<Rpcs>["write"]>[2]
246 ): Effect.Effect<void> => {
247 if (client.fibers.has(request.id)) {
248 return Effect.interrupt
249 }
250 const rpc = group.requests.get(request.tag) as any as Rpc.AnyWithProps
251 const entry = Context.getOrUndefinedUnsafe(services, rpc?.key) as Rpc.Handler<Rpcs["_tag"]>
252 if (!rpc || !entry) {
253 const write = Effect.catchDefect(
254 options.onFromServer({
255 _tag: "Exit",
256 clientId: client.id,
257 requestId: request.id,
258 exit: Exit.die(`Unknown request tag: ${request.tag}`)
259 }),
260 (defect) => sendDefect(client, defect)
261 )
262 if (!client.ended || client.fibers.size > 0) return write
263 return Effect.ensuring(write, endClient(client))
264 }
265 const isStream = RpcSchema.isStreamSchema(rpc.successSchema)
266 const metadata = {
267 rpc,
268 client: client.serverClient,
269 requestId: request.id,
270 headers: request.headers,
271 payload: request.payload
272 }
273 const result = entry.handler(request.payload, metadata)
274
275 // if the handler requested forking, then we skip the concurrency control
276 const isWrapper = Rpc.isWrapper(result)
277 const isFork = isWrapper && result.fork
278 const isUninterruptible = isWrapper && result.uninterruptible
279 // unwrap the fork data type
280 const streamOrEffect = isWrapper ? result.value : result
281 const handler = isStream
282 ? (streamEffect(client, request, streamOrEffect) as Effect.Effect<{} | Deferred.Deferred<any, any>>)
283 : (streamOrEffect as Effect.Effect<{} | Deferred.Deferred<any, any>>)
284
285 const withMiddleware = rpc.middlewares.size > 0
286 ? applyMiddleware(services, handler, metadata)
287 : handler
288 let responded = false
289 const scope = Scope.makeUnsafe()
290 let deferred: Deferred.Deferred<unknown, unknown> | undefined = undefined
291 let effect = Effect.onExit(withMiddleware, (exit) => {
292 responded = true
293 let write: Effect.Effect<void>
294 if (exit._tag === "Success") {
295 if (Deferred.isDeferred(exit.value)) {
296 deferred = exit.value
297 write = Effect.void
298 } else {
299 write = options.onFromServer({
300 _tag: "Exit",
301 clientId: client.id,
302 requestId: request.id,
303 exit: exit as any
304 })
305 }
306 } else if (
307 !disableFatalDefects &&
308 Cause.hasDies(exit.cause) &&
309 !Cause.hasInterrupts(exit.cause)
310 ) {
311 write = sendDefect(client, Cause.squash(exit.cause))
312 } else {
313 write = options.onFromServer({
314 _tag: "Exit",
315 clientId: client.id,
316 requestId: request.id,
317 exit: exit as any
318 })
319 }
320 const close = Scope.closeUnsafe(scope, exit)
321 if (exit._tag === "Failure") {
322 reportCauseUnsafe(Fiber.getCurrent()!, exit.cause)
323 }
324 return close ? Effect.ensuring(write, close) : write
325 })
326 if (opts?.onRequest) {
327 effect = opts.onRequest(effect)
328 }
329 if (enableTracing) {
330 const parentSpan = Context.getOrUndefined(
331 requestFiber.context,
332 Tracer.ParentSpan
333 )
334 effect = Effect.withSpan(effect, `${spanPrefix}.${request.tag}`, {
335 captureStackTrace: false,
336 attributes: options.spanAttributes,
337 parent: enableSpanPropagation && request.spanId
338 ? Tracer.externalSpan({
339 traceId: request.traceId!,
340 spanId: request.spanId,
341 sampled: request.sampled!
342 })
343 : undefined,
344 links: enableSpanPropagation && parentSpan
345 ? [
346 {
347 span: parentSpan,
348 attributes: {}
349 }
350 ]
351 : undefined
352 })
353 }
354 if (!isFork && concurrencySemaphore) {
355 effect = concurrencySemaphore.withPermit(effect)
356 }
357 const runFork = Effect.runForkWith(
358 Context.add(handlerContext(entry, requestFiber.context), Scope.Scope, scope)
359 )
360 const fiber = trackFiber(
361 runFork(
362 effect,
363 isUninterruptible ? { uninterruptible: true } : undefined
364 )
365 )
366 client.fibers.set(request.id, fiber)
367 fiber.addObserver(function onExit(exit: Exit.Exit<any, any>): void {
368 if (deferred) {
369 const fiber = trackFiber(runFork(Effect.onExit(Deferred.await(deferred), (exit) =>
370 options.onFromServer({
371 _tag: "Exit",
372 clientId: client.id,
373 requestId: request.id,
374 exit: exit as any
375 }))))
376 client.fibers.set(request.id, fiber)
377 deferred = undefined
378 fiber.addObserver(onExit)
379 return
380 }
381 if (!responded && exit._tag === "Failure") {
382 trackFiber(
383 runFork(
384 options.onFromServer({
385 _tag: "Exit",
386 clientId: client.id,
387 requestId: request.id,
388 exit: Exit.interrupt()
389 })
390 )
391 )
392 }
393 client.fibers.delete(request.id)
394 client.latches?.delete(request.id)
395 if (client.ended && client.fibers.size === 0) {
396 trackFiber(runFork(endClient(client)))
397 }
398 })
399 return Effect.void
400 }
401
402 const streamEffect = (
403 client: Client,
404 request: Request<Rpcs>,
405 stream:
406 | Stream.Stream<any, any>
407 | Effect.Effect<Queue.Dequeue<any, any>, any, Scope.Scope>
408 ) => {
409 let latch: Latch.Latch | undefined
410 if (supportsAck) {
411 client.latches ??= new Map()
412 latch = client.latches.get(request.id)
413 if (!latch) {
414 latch = Latch.makeUnsafe(false)
415 client.latches.set(request.id, latch)
416 }
417 }
418 if (Effect.isEffect(stream)) {
419 return stream.pipe(
420 Effect.flatMap((queue) =>
421 Effect.whileLoop({
422 while: constTrue,
423 body: constant(
424 Effect.flatMap(Queue.takeAll(queue), (values) => {
425 const write = options.onFromServer({
426 _tag: "Chunk",
427 clientId: client.id,
428 requestId: request.id,
429 values
430 })
431 if (!latch) return write
432 latch.closeUnsafe()
433 return Effect.flatMap(write, () => latch.await)
434 })
435 ),
436 step: constVoid
437 })
438 ),
439 Pull.catchDone(() => Effect.void),
440 Effect.scoped
441 )
442 }
443 return Stream.runForEachArray(stream, (values) => {
444 const write = options.onFromServer({
445 _tag: "Chunk",
446 clientId: client.id,
447 requestId: request.id,
448 values
449 })
450 if (!latch) return write
451 latch.closeUnsafe()
452 return Effect.andThen(write, latch.await)
453 })
454 }
455
456 const sendDefect = (client: Client, defect: unknown) =>
457 Effect.suspend(() => {
458 const shouldEnd = client.ended && client.fibers.size === 0
459 const write = options.onFromServer({
460 _tag: "Defect",
461 clientId: client.id,
462 defect
463 })
464 if (!shouldEnd) return write
465 return Effect.andThen(write, endClient(client))
466 })
467
468 return identity<RpcServer<Rpcs>>({
469 write,
470 disconnect
471 })
472})
473
474// Merging the handler entry services with the request fiber services costs
475// O(services), so the merged context is cached per (entry, fiber context)
476// pair. Request fibers on persistent connections keep one context across
477// requests, leaving a single O(1) overlay per request for the request scope.
478const handlerContextCache = new WeakMap<
479 Rpc.Handler<any>,
480 WeakMap<Context.Context<never>, Context.Context<never>>
481>()
482
483const handlerContext = (
484 entry: Rpc.Handler<any>,
485 fiberContext: Context.Context<never>
486): Context.Context<never> => {
487 let cache = handlerContextCache.get(entry)
488 if (cache === undefined) {
489 cache = new WeakMap()
490 handlerContextCache.set(entry, cache)
491 }
492 let merged = cache.get(fiberContext)
493 if (merged === undefined) {
494 merged = Context.merge(entry.context, fiberContext)
495 cache.set(fiberContext, merged)
496 }
497 return merged
498}
499
500const applyMiddleware = <A, E, R>(
501 context: Context.Context<never>,
502 handler: Effect.Effect<A, E, R>,
503 options: {
504 readonly rpc: Rpc.AnyWithProps
505 readonly client: Rpc.ServerClient
506 readonly requestId: RequestId
507 readonly headers: Headers.Headers
508 readonly payload: unknown
509 }
510) => {
511 for (const service of options.rpc.middlewares) {
512 const middleware = Context.getUnsafe(context, service)
513 handler = middleware(handler as any, options) as any
514 }
515
516 return handler
517}
518
519/**
520 * Runs an RPC server for a group using the current server `Protocol`, decoding
521 * requests, invoking handlers, encoding responses, and managing in-flight
522 * request lifetime.
523 *
524 * @stability unstable
525 * @category running
526 * @since 4.0.0
527 */
528export const make: <Rpcs extends Rpc.Any>(
529 group: RpcGroup.RpcGroup<Rpcs>,
530 options?:
531 | {
532 readonly disableTracing?: boolean | undefined
533 readonly spanPrefix?: string | undefined
534 readonly spanAttributes?: Record<string, unknown> | undefined
535 readonly concurrency?: number | "unbounded" | undefined
536 readonly disableFatalDefects?: boolean | undefined
537 }
538 | undefined
539) => Effect.Effect<
540 never,
541 never,
542 | Protocol
543 | Rpc.ToHandler<Rpcs>
544 | Rpc.Middleware<Rpcs>
545 | Rpc.ServicesServer<Rpcs>
546> = Effect.fnUntraced(function*<Rpcs extends Rpc.Any>(
547 group: RpcGroup.RpcGroup<Rpcs>,
548 options?: {
549 readonly disableTracing?: boolean | undefined
550 readonly spanPrefix?: string | undefined
551 readonly spanAttributes?: Record<string, unknown> | undefined
552 readonly concurrency?: number | "unbounded" | undefined
553 readonly disableFatalDefects?: boolean | undefined
554 }
555) {
556 const {
557 codecFor,
558 disconnects,
559 end,
560 run,
561 send,
562 supportsAck,
563 supportsSpanPropagation,
564 supportsTransferables
565 } = yield* Protocol
566 const encodeDefectUnsafe = Schema.encodeSync(codecFor(Schema.Defect()))
567 const services = yield* Effect.context<Rpc.ToHandler<Rpcs> | Rpc.Middleware<Rpcs>>()
568 const scope = yield* Scope.make()
569
570 const server = yield* makeNoSerialization(group, {
571 ...options,
572 disableClientAcks: !supportsAck,
573 disableSpanPropagation: !supportsSpanPropagation,
574 onFromServer(response): Effect.Effect<void> {
575 const client = clients.get(response.clientId)
576 if (!client) return Effect.void
577 switch (response._tag) {
578 case "Chunk": {
579 const schemas = client.schemas.get(response.requestId)
580 if (!schemas) return Effect.void
581 return handleEncode(
582 client,
583 response.requestId,
584 schemas,
585 schemas.encodeChunk(response.values),
586 "Chunk"
587 )
588 }
589 case "Exit": {
590 const schemas = client.schemas.get(response.requestId)
591 if (!schemas) return Effect.void
592 client.schemas.delete(response.requestId)
593 return handleEncode(
594 client,
595 response.requestId,
596 schemas,
597 schemas.encodeExit(response.exit),
598 "Exit"
599 )
600 }
601 case "Defect": {
602 return sendDefect(client, response.defect)
603 }
604 case "ClientEnd": {
605 clients.delete(response.clientId)
606 return end(response.clientId)
607 }
608 }
609 }
610 }).pipe(Scope.provide(scope))
611
612 // handle disconnects
613 yield* Effect.forkChild(
614 Effect.whileLoop({
615 while: constTrue,
616 body: constant(
617 Effect.flatMap(Queue.take(disconnects), (clientId) => {
618 clients.delete(clientId)
619 return server.disconnect(clientId)
620 })
621 ),
622 step: constVoid
623 })
624 )
625
626 type Schemas = {
627 readonly decode: (u: unknown) => Effect.Effect<Rpc.Payload<Rpcs>, Schema.SchemaError>
628 readonly encodeChunk: (
629 u: NonEmptyReadonlyArray<unknown>
630 ) => Effect.Effect<NonEmptyReadonlyArray<unknown>, Schema.SchemaError>
631 readonly encodeExit: (u: unknown) => Effect.Effect<ResponseExitEncoded["exit"], Schema.SchemaError>
632 readonly encodeDefect: (u: unknown) => Effect.Effect<unknown, Schema.SchemaError>
633 readonly context: Context.Context<never>
634 readonly collector?: Transferable.Collector["Service"] | undefined
635 }
636
637 const schemasCache = new WeakMap<any, Schemas>()
638 const getSchemas = (rpc: Rpc.AnyWithProps) => {
639 let schemas = schemasCache.get(rpc)
640 if (!schemas) {
641 const entry = Context.getOrUndefinedUnsafe(services, rpc.key) as Rpc.Handler<Rpcs["_tag"]>
642 const streamSchemas = RpcSchema.getStreamSchemas(rpc.successSchema)
643 schemas = {
644 decode: Schema.decodeUnknownEffect(codecFor(rpc.payloadSchema)) as any,
645 encodeChunk: Schema.encodeUnknownEffect(
646 codecFor(
647 Schema.NonEmptyArray(Option.isSome(streamSchemas) ? streamSchemas.value.success : Schema.Any)
648 )
649 ) as any,
650 encodeExit: Schema.encodeUnknownEffect(codecFor(Rpc.exitSchema(rpc as any))) as any,
651 encodeDefect: Schema.encodeUnknownEffect(codecFor(rpc.defectSchema)) as any,
652 context: entry.context
653 }
654 schemasCache.set(rpc, schemas)
655 }
656 return schemas
657 }
658
659 type Client = {
660 readonly id: number
661 readonly schemas: Map<RequestId, Schemas>
662 }
663 const clients = new Map<number, Client>()
664
665 const responseEnvelope = (
666 requestId: RequestId,
667 tag: "Chunk" | "Exit",
668 value: unknown
669 ): FromServerEncoded =>
670 tag === "Chunk"
671 ? { _tag: "Chunk", requestId, values: value as NonEmptyReadonlyArray<unknown> }
672 : { _tag: "Exit", requestId, exit: value as ResponseExitEncoded["exit"] }
673
674 const handleEncode = (
675 client: Client,
676 requestId: RequestId,
677 schemas: Schemas,
678 effect: Effect.Effect<unknown, Schema.SchemaError>,
679 tag: "Chunk" | "Exit"
680 ) => {
681 const collector = schemas.collector
682 // The schema encoders evaluate eagerly, so an encode that needs no
683 // services is already a resolved success and can skip the effect wrappers.
684 const write = Exit.isExit(effect) && Exit.isSuccess(effect)
685 ? send(client.id, responseEnvelope(requestId, tag, effect.value), collector && collector.clearUnsafe())
686 : Effect.flatMap(
687 Effect.provideContext(
688 collector ? Effect.provideService(effect, Transferable.Collector, collector) : effect,
689 schemas.context
690 ),
691 (value) => send(client.id, responseEnvelope(requestId, tag, value), collector && collector.clearUnsafe())
692 )
693 return Effect.catchCause(write, (cause) => {
694 client.schemas.delete(requestId)
695 const defect = Cause.squash(Cause.map(cause, (e) => SchemaIssue.defaultFormatter(e.issue)))
696 return Effect.andThen(
697 sendRequestDefect(client, requestId, schemas.encodeDefect, defect),
698 server.write(client.id, { _tag: "Interrupt", requestId, interruptors: [] })
699 )
700 })
701 }
702
703 const sendRequestDefect = (
704 client: Client,
705 requestId: RequestId,
706 encodeDefect: (u: unknown) => Effect.Effect<unknown, Schema.SchemaError>,
707 defect: unknown
708 ) =>
709 Effect.catchCause(
710 Effect.flatMap(encodeDefect(defect), (encodedDefect) =>
711 send(client.id, {
712 _tag: "Exit",
713 requestId,
714 exit: {
715 _tag: "Failure",
716 cause: [{
717 _tag: "Die",
718 defect: encodedDefect
719 }]
720 }
721 })),
722 (cause) => sendDefect(client, Cause.squash(cause))
723 )
724
725 const sendDefect = (client: Client, defect: unknown) =>
726 Effect.catchCause(
727 send(client.id, ResponseDefectEncoded(encodeDefectUnsafe(defect))),
728 (cause) =>
729 Effect.annotateLogs(Effect.logDebug(cause), {
730 module: "RpcServer",
731 method: "sendDefect"
732 })
733 )
734
735 const writeDecodedRequest = (
736 client: Client,
737 requestId: RequestId,
738 schemas: Schemas,
739 request: RequestEncoded,
740 payload: unknown
741 ) => {
742 client.schemas.set(
743 requestId,
744 supportsTransferables
745 ? {
746 ...schemas,
747 collector: Transferable.makeCollectorUnsafe()
748 }
749 : schemas
750 )
751 // the envelope is owned by this protocol loop, so it is decoded in place
752 // instead of copied for every request
753 const decoded = request as Types.Mutable<RequestEncoded>
754 decoded.id = requestId
755 decoded.payload = payload
756 decoded.headers = Headers.fromInput(request.headers) as any
757 return server.write(client.id, decoded as any)
758 }
759
760 // main server loop
761 return yield* run((clientId, request) => {
762 let client = clients.get(clientId)
763 if (!client) {
764 client = {
765 id: clientId,
766 schemas: new Map()
767 }
768 clients.set(clientId, client)
769 }
770
771 switch (request._tag) {
772 case "Request": {
773 const tag = Object.hasOwn(request, "tag") ? (request.tag as string) : ""
774 let requestId: RequestId
775 switch (typeof request.id) {
776 case "number":
777 case "string": {
778 requestId = RequestId(request.id)
779 break
780 }
781 default: {
782 return sendDefect(client, `Invalid request id: ${request.id}`)
783 }
784 }
785 const rpc = group.requests.get(tag)
786 if (!rpc) {
787 return sendRequestDefect(client, requestId, (defect) => Effect.succeed(defect), `Unknown request tag: ${tag}`)
788 }
789 const schemas = getSchemas(rpc as any)
790 const decoded = schemas.decode(request.payload)
791 // The schema decoders evaluate eagerly, so a decode that needs no
792 // services is already a resolved success and can skip the effect
793 // wrappers.
794 if (Exit.isExit(decoded) && Exit.isSuccess(decoded)) {
795 return writeDecodedRequest(client, requestId, schemas, request, decoded.value)
796 }
797 return Effect.matchEffect(
798 Effect.provideContext(decoded, schemas.context),
799 {
800 onFailure: (error) =>
801 sendRequestDefect(client, requestId, schemas.encodeDefect, SchemaIssue.defaultFormatter(error.issue)),
802 onSuccess: (payload) => writeDecodedRequest(client, requestId, schemas, request, payload)
803 }
804 )
805 }
806 case "Ping": {
807 return Effect.catchCause(send(client.id, constPong), (cause) => sendDefect(client, Cause.squash(cause)))
808 }
809 case "Eof": {
810 return server.write(clientId, request)
811 }
812 case "Ack": {
813 // the branded `RequestId` shares the runtime representation of the
814 // encoded id, so the message can be forwarded as-is
815 return server.write(clientId, request as FromClient<Rpcs>)
816 }
817 case "Interrupt": {
818 return server.write(clientId, {
819 ...request,
820 requestId: RequestId(request.requestId),
821 interruptors: []
822 })
823 }
824 default: {
825 return sendDefect(client, `Unknown request tag: ${(request as any)._tag}`)
826 }
827 }
828 }).pipe(
829 Effect.tapCause((cause) => Effect.logFatal("BUG: RpcServer protocol crashed", cause)),
830 Effect.onExit((exit) => Scope.close(scope, exit))
831 )
832})
833
834/**
835 * Provides a scoped layer that starts an RPC server for a group using the
836 * current server `Protocol`.
837 *
838 * @stability unstable
839 * @category layers
840 * @since 4.0.0
841 */
842export const layer = <Rpcs extends Rpc.Any>(
843 group: RpcGroup.RpcGroup<Rpcs>,
844 options?: {
845 readonly disableTracing?: boolean | undefined
846 readonly spanPrefix?: string | undefined
847 readonly spanAttributes?: Record<string, unknown> | undefined
848 readonly concurrency?: number | "unbounded" | undefined
849 readonly disableFatalDefects?: boolean | undefined
850 }
851): Layer.Layer<
852 never,
853 never,
854 | Protocol
855 | Rpc.ToHandler<Rpcs>
856 | Rpc.Middleware<Rpcs>
857 | Rpc.ServicesServer<Rpcs>
858> => Layer.effectDiscard(Effect.forkScoped(make(group, options)))
859
860/**
861 * Creates a RPC server that registers a HTTP route with a `HttpRouter`.
862 *
863 * **Details**
864 *
865 * Defaults to using websockets for communication, but can be configured to use
866 * HTTP.
867 *
868 * @stability unstable
869 * @category layers
870 * @since 4.0.0
871 */
872export const layerHttp = <Rpcs extends Rpc.Any>(options: {
873 readonly group: RpcGroup.RpcGroup<Rpcs>
874 readonly path: HttpRouter.PathInput
875 readonly protocol?: "http" | "websocket" | undefined
876 readonly disableTracing?: boolean | undefined
877 readonly spanPrefix?: string | undefined
878 readonly spanAttributes?: Record<string, unknown> | undefined
879 readonly concurrency?: number | "unbounded" | undefined
880 readonly disableFatalDefects?: boolean | undefined
881 readonly streamBufferSize?: number | "unbounded" | undefined
882}): Layer.Layer<
883 never,
884 never,
885 | RpcSerialization.RpcSerialization
886 | HttpRouter.HttpRouter
887 | Rpc.ToHandler<Rpcs>
888 | Rpc.Middleware<Rpcs>
889 | Rpc.ServicesServer<Rpcs>
890> =>
891 layer(options.group, options).pipe(
892 Layer.provide(
893 options.protocol === "http"
894 ? layerProtocolHttp(options)
895 : layerProtocolWebsocket(options)
896 )
897 )
898
899/**
900 * Defines the service interface for an RPC server transport, responsible for receiving
901 * encoded client messages, sending encoded responses, tracking clients, and
902 * declaring transport capabilities.
903 *
904 * **When to use**
905 *
906 * Use to provide the transport boundary for RPC servers over HTTP, WebSocket,
907 * workers, sockets, or custom protocols.
908 *
909 * @stability unstable
910 * @category services
911 * @since 4.0.0
912 */
913export class Protocol extends Context.Service<
914 Protocol,
915 {
916 readonly run: (
917 f: (clientId: number, data: FromClientEncoded) => Effect.Effect<void>
918 ) => Effect.Effect<never>
919 readonly disconnects: Queue.Dequeue<number>
920 readonly send: (
921 clientId: number,
922 response: FromServerEncoded,
923 transferables?: ReadonlyArray<globalThis.Transferable>
924 ) => Effect.Effect<void>
925 readonly end: (clientId: number) => Effect.Effect<void>
926 readonly clientIds: Effect.Effect<ReadonlySet<number>>
927 readonly initialMessage: Effect.Effect<Option.Option<unknown>>
928 readonly supportsAck: boolean
929 readonly supportsTransferables: boolean
930 readonly supportsSpanPropagation: boolean
931 readonly supportsNotifications: boolean
932 /**
933 * Builds the codec that fills the `unknown` holes of the protocol messages,
934 * re-passed from the `RpcSerialization` backing this transport.
935 */
936 readonly codecFor: RpcSerialization.CodecFor
937 }
938>()("effect/rpc/RpcServer/Protocol") {
939 /**
940 * Creates a server protocol service from the supplied RPC implementation.
941 *
942 * @stability unstable
943 * @since 4.0.0
944 */
945 static make = withRun<Protocol["Service"]>()
946}
947
948/**
949 * Creates a server `Protocol` backed by the current `SocketServer`, accepting
950 * socket connections and routing decoded RPC messages.
951 *
952 * @stability unstable
953 * @category protocols
954 * @since 4.0.0
955 */
956export const makeProtocolSocketServer = Effect.gen(function*() {
957 const server = yield* SocketServer.SocketServer
958 const { onSocket, protocol } = yield* makeSocketProtocol
959 yield* Effect.forkScoped(
960 server.run((socket) => Effect.scoped(onSocket(socket)))
961 )
962 return protocol
963})
964
965/**
966 * RPC protocol that uses `SocketServer` for communication.
967 *
968 * @stability unstable
969 * @category layers
970 * @since 4.0.0
971 */
972export const layerProtocolSocketServer: Layer.Layer<
973 Protocol,
974 never,
975 RpcSerialization.RpcSerialization | SocketServer.SocketServer
976> = Layer.effect(Protocol)(makeProtocolSocketServer)
977
978/**
979 * Creates a websocket server `Protocol` together with an HTTP effect that
980 * upgrades the current request to a websocket and attaches it to the protocol.
981 *
982 * @stability unstable
983 * @category protocols
984 * @since 4.0.0
985 */
986export const makeProtocolWithHttpEffectWebsocket: Effect.Effect<
987 {
988 readonly protocol: Protocol["Service"]
989 readonly httpEffect: Effect.Effect<
990 HttpServerResponse.HttpServerResponse,
991 never,
992 Scope.Scope | HttpServerRequest.HttpServerRequest
993 >
994 },
995 never,
996 RpcSerialization.RpcSerialization
997> = Effect.gen(function*() {
998 const { onSocket, protocol } = yield* makeSocketProtocol
999
1000 const httpEffect: Effect.Effect<
1001 HttpServerResponse.HttpServerResponse,
1002 never,
1003 Scope.Scope | HttpServerRequest.HttpServerRequest
1004 > = Effect.gen(function*() {
1005 const request = yield* HttpServerRequest.HttpServerRequest
1006 const socket = yield* Effect.orDie(request.upgrade)
1007 yield* onSocket(socket, Object.entries(request.headers))
1008 return HttpServerResponse.empty()
1009 })
1010
1011 return { protocol, httpEffect } as const
1012})
1013
1014/**
1015 * Creates a websocket server `Protocol` and registers its upgrade handler as a
1016 * GET route on the current `HttpRouter`.
1017 *
1018 * @stability unstable
1019 * @category protocols
1020 * @since 4.0.0
1021 */
1022export const makeProtocolWebsocket: (options: {
1023 readonly path: HttpRouter.PathInput
1024}) => Effect.Effect<
1025 Protocol["Service"],
1026 never,
1027 RpcSerialization.RpcSerialization | HttpRouter.HttpRouter
1028> = Effect.fnUntraced(function*(options) {
1029 const { httpEffect, protocol } = yield* makeProtocolWithHttpEffectWebsocket
1030 const router = yield* HttpRouter.HttpRouter
1031 yield* router.add("GET", options.path, httpEffect)
1032 return protocol
1033})
1034
1035/**
1036 * RPC protocol that uses WebSockets for communication.
1037 *
1038 * @stability unstable
1039 * @category layers
1040 * @since 4.0.0
1041 */
1042export const layerProtocolWebsocket = (options: {
1043 readonly path: HttpRouter.PathInput
1044}): Layer.Layer<
1045 Protocol,
1046 never,
1047 RpcSerialization.RpcSerialization | HttpRouter.HttpRouter
1048> => {
1049 return Layer.effect(Protocol)(makeProtocolWebsocket(options))
1050}
1051
1052/**
1053 * Creates an HTTP request/response server `Protocol` together with an HTTP
1054 * effect that decodes the current request and streams or returns encoded RPC
1055 * responses.
1056 *
1057 * @stability unstable
1058 * @category protocols
1059 * @since 4.0.0
1060 */
1061export const makeProtocolWithHttpEffect: (
1062 options?: {
1063 readonly streamBufferSize?: number | "unbounded" | undefined
1064 } | undefined
1065) => Effect.Effect<
1066 {
1067 readonly protocol: Protocol["Service"]
1068 readonly httpEffect: Effect.Effect<
1069 HttpServerResponse.HttpServerResponse,
1070 never,
1071 Scope.Scope | HttpServerRequest.HttpServerRequest
1072 >
1073 },
1074 never,
1075 RpcSerialization.RpcSerialization
1076> = Effect.fnUntraced(function*(options = {}) {
1077 const serialization = yield* RpcSerialization.RpcSerialization
1078 const encodeDefectUnsafe = Schema.encodeSync(serialization.codecFor(Schema.Defect()))
1079 const includesFraming = serialization.includesFraming
1080 const isBinary = !serialization.contentType.includes("json")
1081
1082 const disconnects = yield* Queue.make<number>()
1083 let writeRequest!: (
1084 clientId: number,
1085 message: FromClientEncoded
1086 ) => Effect.Effect<void>
1087
1088 let clientId = 0
1089
1090 type Client = {
1091 readonly write: (bytes: FromServerEncoded) => Effect.Effect<void>
1092 readonly end: Effect.Effect<void>
1093 }
1094 const clients = new Map<number, Client>()
1095 const clientIds = new Set<number>()
1096
1097 const encoder = new TextEncoder()
1098
1099 const httpEffect: Effect.Effect<
1100 HttpServerResponse.HttpServerResponse,
1101 never,
1102 Scope.Scope | HttpServerRequest.HttpServerRequest
1103 > = Effect.gen(function*() {
1104 const fiber = Fiber.getCurrent()!
1105 const request = Context.getUnsafe(fiber.context, HttpServerRequest.HttpServerRequest)
1106 const scope = Context.getUnsafe(fiber.context, Scope.Scope)
1107 const requestHeaders = Object.entries(request.headers)
1108 const data = yield* Effect.orDie<string | Uint8Array, any, never>(
1109 isBinary ? Effect.map(request.arrayBuffer, (buf) => new Uint8Array(buf)) : request.text
1110 )
1111 const id = clientId++
1112 const queue = yield* Queue.make<Uint8Array | FromServerEncoded, Cause.Done>({
1113 capacity: includesFraming && options.streamBufferSize !== "unbounded"
1114 ? options.streamBufferSize ?? 16
1115 : undefined
1116 })
1117 const parser = serialization.makeUnsafe()
1118 const requestIds: Array<RequestId> = []
1119
1120 const offer = (data: Uint8Array | string) =>
1121 typeof data === "string" ? Queue.offer(queue, encoder.encode(data)) : Queue.offer(queue, data)
1122 const client: Client = {
1123 write: !includesFraming
1124 ? (response) =>
1125 // buffered responses cannot carry notifications, so they are dropped
1126 response._tag === "Request" && response.isNotification === true
1127 ? Effect.void
1128 : Queue.offer(queue, response)
1129 : (response) => {
1130 try {
1131 const encoded = parser.encode(response)
1132 if (encoded === undefined) return Effect.void
1133 return offer(encoded)
1134 } catch (cause) {
1135 return offer(parser.encode(ResponseDefectEncoded(encodeDefectUnsafe(cause)))!)
1136 }
1137 },
1138 end: Queue.end(queue)
1139 }
1140
1141 yield* Scope.addFinalizerExit(scope, () => {
1142 clients.delete(id)
1143 clientIds.delete(id)
1144 Queue.offerUnsafe(disconnects, id)
1145 if (queue.state._tag === "Done") return Effect.void
1146 return Effect.andThen(
1147 Queue.shutdown(queue),
1148 Effect.forEach(
1149 requestIds,
1150 (requestId) => writeRequest(id, { _tag: "Interrupt", requestId }),
1151 { discard: true }
1152 )
1153 )
1154 })
1155 clients.set(id, client)
1156 clientIds.add(id)
1157
1158 // @effect-diagnostics-next-line tryCatchInEffectGen:off
1159 try {
1160 const decoded = parser.decode(data) as ReadonlyArray<FromClientEncoded>
1161 for (let i = 0; i < decoded.length; i++) {
1162 const message = decoded[i]
1163 if (message._tag === "Request") {
1164 requestIds.push(RequestId(message.id))
1165 ;(message as Types.Mutable<RequestEncoded>).headers = requestHeaders.concat(message.headers)
1166 }
1167 yield* writeRequest(id, message)
1168 }
1169 } catch (cause) {
1170 yield* client.write(ResponseDefectEncoded(encodeDefectUnsafe(cause)))
1171 }
1172
1173 yield* writeRequest(id, constEof)
1174
1175 if (!includesFraming) {
1176 const responses = yield* Queue.collect(queue)
1177 // Notification-only batches encode without a response body.
1178 const encoded = parser.encode(responses)
1179 return HttpServerResponse.text(encoded === undefined ? "" : encoded as string, {
1180 contentType: serialization.contentType
1181 })
1182 }
1183
1184 const initialChunk = yield* Queue.takeAll(queue) as any as Effect.Effect<NonEmptyReadonlyArray<Uint8Array>>
1185 if (queue.state._tag === "Done") {
1186 return HttpServerResponse.uint8Array(mergeUint8Arrays(initialChunk), {
1187 contentType: serialization.contentType
1188 })
1189 }
1190
1191 return HttpServerResponse.stream(
1192 Stream.fromArray(initialChunk).pipe(
1193 Stream.concat(
1194 Stream.fromQueue(queue as Queue.Dequeue<Uint8Array, Cause.Done>)
1195 )
1196 ),
1197 { contentType: serialization.contentType }
1198 )
1199 })
1200
1201 const protocol = yield* Protocol.make((writeRequest_) => {
1202 writeRequest = writeRequest_
1203 return Effect.succeed({
1204 disconnects,
1205 send(clientId, response) {
1206 const client = clients.get(clientId)
1207 if (!client) return Effect.void
1208 return client.write(response)
1209 },
1210 end(clientId) {
1211 const client = clients.get(clientId)
1212 if (!client) return Effect.void
1213 return client.end
1214 },
1215 clientIds: Effect.sync(() => clientIds),
1216 initialMessage: Effect.succeedNone,
1217 supportsAck: false,
1218 supportsTransferables: false,
1219 supportsSpanPropagation: false,
1220 supportsNotifications: includesFraming,
1221 codecFor: serialization.codecFor
1222 })
1223 })
1224
1225 return { protocol, httpEffect } as const
1226})
1227
1228const mergeUint8Arrays = (arrays: ReadonlyArray<Uint8Array>) => {
1229 if (arrays.length === 0) return new Uint8Array(0)
1230 if (arrays.length === 1) return arrays[0]
1231 const length = arrays.reduce((acc, arr) => acc + arr.length, 0)
1232 const result = new Uint8Array(length)
1233 let offset = 0
1234 for (const arr of arrays) {
1235 result.set(arr, offset)
1236 offset += arr.length
1237 }
1238 return result
1239}
1240
1241/**
1242 * Creates an HTTP server `Protocol` and registers its request handler as a POST
1243 * route on the current `HttpRouter`.
1244 *
1245 * @stability unstable
1246 * @category protocols
1247 * @since 4.0.0
1248 */
1249export const makeProtocolHttp: (options: {
1250 readonly path: HttpRouter.PathInput
1251 readonly streamBufferSize?: number | "unbounded" | undefined
1252}) => Effect.Effect<
1253 Protocol["Service"],
1254 never,
1255 RpcSerialization.RpcSerialization | HttpRouter.HttpRouter
1256> = Effect.fnUntraced(function*(options) {
1257 const { httpEffect, protocol } = yield* makeProtocolWithHttpEffect(options)
1258 const router = yield* HttpRouter.HttpRouter
1259 yield* router.add("POST", options.path, httpEffect)
1260 return protocol
1261})
1262
1263/**
1264 * Provides a server `Protocol` that uses HTTP POST requests for RPC
1265 * communication.
1266 *
1267 * @stability unstable
1268 * @category layers
1269 * @since 4.0.0
1270 */
1271export const layerProtocolHttp = (options: {
1272 readonly path: HttpRouter.PathInput
1273 readonly streamBufferSize?: number | "unbounded" | undefined
1274}): Layer.Layer<Protocol, never, RpcSerialization.RpcSerialization | HttpRouter.HttpRouter> => {
1275 return Layer.effect(Protocol)(makeProtocolHttp(options))
1276}
1277
1278/**
1279 * Starts an RPC server for a group and returns the HTTP request/response effect
1280 * that serves the non-websocket HTTP RPC protocol.
1281 *
1282 * @stability unstable
1283 * @category running
1284 * @since 4.0.0
1285 */
1286export const toHttpEffect: <Rpcs extends Rpc.Any>(
1287 group: RpcGroup.RpcGroup<Rpcs>,
1288 options?: {
1289 readonly disableTracing?: boolean | undefined
1290 readonly spanPrefix?: string | undefined
1291 readonly spanAttributes?: Record<string, unknown> | undefined
1292 readonly disableFatalDefects?: boolean | undefined
1293 readonly streamBufferSize?: number | "unbounded" | undefined
1294 } | undefined
1295) => Effect.Effect<
1296 Effect.Effect<HttpServerResponse.HttpServerResponse, never, Scope.Scope | HttpServerRequest.HttpServerRequest>,
1297 never,
1298 | Scope.Scope
1299 | RpcSerialization.RpcSerialization
1300 | Rpc.ToHandler<Rpcs>
1301 | Rpc.Middleware<Rpcs>
1302 | Rpc.ServicesServer<Rpcs>
1303> = Effect.fnUntraced(function*<Rpcs extends Rpc.Any>(
1304 group: RpcGroup.RpcGroup<Rpcs>,
1305 options?: {
1306 readonly disableTracing?: boolean | undefined
1307 readonly spanPrefix?: string | undefined
1308 readonly spanAttributes?: Record<string, unknown> | undefined
1309 readonly disableFatalDefects?: boolean | undefined
1310 readonly streamBufferSize?: number | "unbounded" | undefined
1311 }
1312) {
1313 const { httpEffect, protocol } = yield* makeProtocolWithHttpEffect(options)
1314 yield* make(group, options).pipe(
1315 Effect.provideService(Protocol, protocol),
1316 Effect.forkScoped
1317 )
1318 // @effect-diagnostics-next-line returnEffectInGen:off
1319 return httpEffect
1320})
1321
1322/**
1323 * Starts an RPC server for a group and returns the HTTP effect that upgrades
1324 * requests to the websocket RPC protocol.
1325 *
1326 * @stability unstable
1327 * @category running
1328 * @since 4.0.0
1329 */
1330export const toHttpEffectWebsocket: <Rpcs extends Rpc.Any>(
1331 group: RpcGroup.RpcGroup<Rpcs>,
1332 options?: {
1333 readonly disableTracing?: boolean | undefined
1334 readonly spanPrefix?: string | undefined
1335 readonly spanAttributes?: Record<string, unknown> | undefined
1336 readonly disableFatalDefects?: boolean | undefined
1337 } | undefined
1338) => Effect.Effect<
1339 Effect.Effect<HttpServerResponse.HttpServerResponse, never, Scope.Scope | HttpServerRequest.HttpServerRequest>,
1340 never,
1341 | Scope.Scope
1342 | RpcSerialization.RpcSerialization
1343 | Rpc.ToHandler<Rpcs>
1344 | Rpc.Middleware<Rpcs>
1345 | Rpc.ServicesServer<Rpcs>
1346> = Effect.fnUntraced(function*<Rpcs extends Rpc.Any>(
1347 group: RpcGroup.RpcGroup<Rpcs>,
1348 options?: {
1349 readonly disableTracing?: boolean | undefined
1350 readonly spanPrefix?: string | undefined
1351 readonly spanAttributes?: Record<string, unknown> | undefined
1352 readonly disableFatalDefects?: boolean | undefined
1353 }
1354) {
1355 const { httpEffect, protocol } = yield* makeProtocolWithHttpEffectWebsocket
1356 yield* make(group, options).pipe(
1357 Effect.provideService(Protocol, protocol),
1358 Effect.forkScoped
1359 )
1360 // @effect-diagnostics-next-line returnEffectInGen:off
1361 return httpEffect
1362})
1363
1364/**
1365 * Creates a server `Protocol` that reads RPC messages from `Stdio.stdin` and
1366 * writes encoded responses to `Stdio.stdout`.
1367 *
1368 * @stability unstable
1369 * @category protocols
1370 * @since 4.0.0
1371 */
1372export const makeProtocolStdio = Effect.gen(function*() {
1373 const stdio = yield* Stdio
1374 const fiber = Fiber.getCurrent()!
1375 const serialization = yield* RpcSerialization.RpcSerialization
1376
1377 return yield* Protocol.make(Effect.fnUntraced(function*(writeRequest) {
1378 const queue = yield* Queue.bounded<Uint8Array | string, Cause.Done>(8)
1379 const parser = serialization.makeUnsafe()
1380
1381 yield* stdio.stdin.pipe(
1382 Stream.runForEach((data) => {
1383 const decoded = parser.decode(data) as ReadonlyArray<FromClientEncoded>
1384 if (decoded.length === 0) return Effect.void
1385 let i = 0
1386 return Effect.whileLoop({
1387 while: () => i < decoded.length,
1388 body: () => writeRequest(0, decoded[i++]),
1389 step: constVoid
1390 })
1391 }),
1392 Effect.sandbox,
1393 Effect.tapError(Effect.logError),
1394 Effect.retry(Schedule.spaced(500)),
1395 Effect.andThen(() => writeRequest(0, constEof)),
1396 Effect.forkScoped
1397 )
1398
1399 yield* Stream.fromQueue(queue).pipe(
1400 Stream.run(stdio.stdout()),
1401 Effect.retry(Schedule.spaced(500)),
1402 Effect.ensuring(Effect.forkDetach(Fiber.interrupt(fiber), { startImmediately: true })),
1403 Effect.forkScoped
1404 )
1405
1406 return {
1407 disconnects: yield* Queue.make<number>(),
1408 send(_clientId, response) {
1409 const responseEncoded = parser.encode(response)
1410 if (responseEncoded === undefined) {
1411 return Effect.void
1412 }
1413 return Queue.offer(queue, responseEncoded)
1414 },
1415 end(_clientId) {
1416 return Queue.end(queue)
1417 },
1418 clientIds: Effect.succeed(new Set([0])),
1419 initialMessage: Effect.succeedNone,
1420 supportsAck: true,
1421 supportsTransferables: false,
1422 supportsSpanPropagation: true,
1423 supportsNotifications: true,
1424 codecFor: serialization.codecFor
1425 }
1426 }))
1427})
1428
1429/**
1430 * Provides a server `Protocol` that reads RPC messages from `Stdio.stdin` and
1431 * writes encoded responses to `Stdio.stdout`.
1432 *
1433 * @stability unstable
1434 * @category layers
1435 * @since 4.0.0
1436 */
1437export const layerProtocolStdio: Layer.Layer<
1438 Protocol,
1439 never,
1440 RpcSerialization.RpcSerialization | Stdio
1441> = Layer.effect(Protocol, makeProtocolStdio)
1442
1443/**
1444 * Creates a server `Protocol` backed by `WorkerRunnerPlatform`, routing worker
1445 * messages to the RPC server and server responses back to workers.
1446 *
1447 * @stability unstable
1448 * @category protocols
1449 * @since 4.0.0
1450 */
1451export const makeProtocolWorkerRunner: Effect.Effect<
1452 Protocol["Service"],
1453 WorkerError,
1454 WorkerRunner.WorkerRunnerPlatform | Scope.Scope
1455> = Protocol.make(Effect.fnUntraced(function*(writeRequest) {
1456 const fiber = Fiber.getCurrent()!
1457 const runner = yield* WorkerRunner.WorkerRunnerPlatform
1458 const backing = yield* runner.start<FromServerEncoded, FromClientEncoded | InitialMessage.Encoded>()
1459 const initialMessage = yield* Deferred.make<unknown>()
1460 const clientIds = new Set<number>()
1461 const disconnects = yield* Queue.make<number>()
1462
1463 yield* backing.run((clientId, message) => {
1464 clientIds.add(clientId)
1465 if (message._tag === "InitialMessage") {
1466 return Deferred.succeed(initialMessage, message.value)
1467 }
1468 return writeRequest(clientId, message)
1469 }).pipe(
1470 Effect.tapCause(Effect.logError),
1471 Effect.onExit(() =>
1472 Effect.sync(() => {
1473 fiber.currentDispatcher.scheduleTask(() => fiber.interruptUnsafe(fiber.id), 0)
1474 })
1475 ),
1476 Effect.forkScoped
1477 )
1478
1479 if (backing.disconnects) {
1480 yield* Queue.take(backing.disconnects).pipe(
1481 Effect.tap((clientId) => {
1482 clientIds.delete(clientId)
1483 return Queue.offer(disconnects, clientId)
1484 }),
1485 Effect.forever,
1486 Effect.forkScoped
1487 )
1488 }
1489
1490 return {
1491 disconnects,
1492 send: backing.send,
1493 end(_clientId) {
1494 return Effect.void
1495 },
1496 clientIds: Effect.sync(() => clientIds),
1497 initialMessage: Effect.asSome(Deferred.await(initialMessage)),
1498 supportsAck: true,
1499 supportsTransferables: true,
1500 supportsSpanPropagation: true,
1501 supportsNotifications: true,
1502 // Worker protocols use structured clone, so they do not depend on
1503 // `RpcSerialization`. A binary worker protocol is a separate protocol.
1504 codecFor: Schema.toCodecJson as RpcSerialization.CodecFor
1505 }
1506}))
1507
1508/**
1509 * Provides a server `Protocol` backed by the current `WorkerRunnerPlatform`.
1510 *
1511 * @stability unstable
1512 * @category layers
1513 * @since 4.0.0
1514 */
1515export const layerProtocolWorkerRunner: Layer.Layer<
1516 Protocol,
1517 WorkerError,
1518 WorkerRunner.WorkerRunnerPlatform
1519> = Layer.effect(Protocol)(makeProtocolWorkerRunner)
1520
1521// internal
1522
1523const makeSocketProtocol: Effect.Effect<
1524 {
1525 readonly protocol: Protocol["Service"]
1526 readonly onSocket: (
1527 socket: Socket.Socket,
1528 headers?: ReadonlyArray<[string, string]>
1529 ) => Effect.Effect<void, never, Scope.Scope>
1530 },
1531 never,
1532 RpcSerialization.RpcSerialization
1533> = Effect.gen(function*() {
1534 const serialization = yield* RpcSerialization.RpcSerialization
1535 const encodeDefectUnsafe = Schema.encodeSync(serialization.codecFor(Schema.Defect()))
1536 const disconnects = yield* Queue.make<number>()
1537
1538 let clientId = 0
1539 const clients = new Map<number, {
1540 readonly write: (bytes: FromServerEncoded) => Effect.Effect<void>
1541 }>()
1542 const clientIds = new Set<number>()
1543
1544 let writeRequest!: (clientId: number, message: FromClientEncoded) => Effect.Effect<void>
1545
1546 const onSocket = Effect.fnUntraced(function*(socket: Socket.Socket, headers?: ReadonlyArray<[string, string]>) {
1547 const scope = yield* Effect.scope
1548 const parser = serialization.makeUnsafe()
1549 const id = clientId++
1550 yield* Scope.addFinalizerExit(scope, () => {
1551 clients.delete(id)
1552 clientIds.delete(id)
1553 return Queue.offer(disconnects, id)
1554 })
1555
1556 const writer = yield* socket.writer
1557 const write = (response: FromServerEncoded) => {
1558 try {
1559 const encoded = parser.encode(response)
1560 if (encoded === undefined) {
1561 return Effect.void
1562 }
1563 return Effect.orDie(writer.write(encoded))
1564 } catch (cause) {
1565 return Effect.orDie(
1566 writer.write(parser.encode(ResponseDefectEncoded(encodeDefectUnsafe(cause)))!)
1567 )
1568 }
1569 }
1570 clients.set(id, { write })
1571 clientIds.add(id)
1572
1573 const processData = (data: Uint8Array | string): Effect.Effect<void, Socket.SocketError> => {
1574 try {
1575 const decoded = parser.decode(data) as ReadonlyArray<FromClientEncoded>
1576 if (decoded.length === 0) return Effect.void
1577 let i = 0
1578 return Effect.whileLoop({
1579 while: () => i < decoded.length,
1580 body() {
1581 const message = decoded[i++]
1582 if (message._tag === "Request" && headers) {
1583 ;(message as Types.Mutable<RequestEncoded>).headers = headers.concat(message.headers)
1584 }
1585 return writeRequest(id, message)
1586 },
1587 step: constVoid
1588 })
1589 } catch (cause) {
1590 if (Predicate.isTagged(cause, "MaxBufferSizeExceeded")) {
1591 return writer.write(new Socket.CloseEvent(1009, String(cause)))
1592 }
1593 return writer.write(parser.encode(ResponseDefectEncoded(encodeDefectUnsafe(cause)))!)
1594 }
1595 }
1596
1597 yield* Effect.gen(function*() {
1598 const { pull } = yield* socket.reader
1599 while (true) {
1600 const frames = yield* pull
1601 for (let i = 0; i < frames.length; i++) {
1602 yield* processData(frames[i])
1603 }
1604 }
1605 }).pipe(
1606 Effect.catchReason("SocketError", "SocketCloseError", (_) => Effect.void),
1607 Effect.orDie
1608 )
1609 })
1610
1611 const protocol = yield* Protocol.make((writeRequest_) => {
1612 writeRequest = writeRequest_
1613 return Effect.succeed({
1614 disconnects,
1615 send: (clientId, response) => {
1616 const client = clients.get(clientId)
1617 if (!client) return Effect.void
1618 return Effect.orDie(client.write(response))
1619 },
1620 end(_clientId) {
1621 return Effect.void
1622 },
1623 clientIds: Effect.sync(() => clientIds),
1624 initialMessage: Effect.succeedNone,
1625 supportsAck: true,
1626 supportsTransferables: false,
1627 supportsSpanPropagation: true,
1628 supportsNotifications: true,
1629 codecFor: serialization.codecFor
1630 })
1631 })
1632
1633 return { protocol, onSocket } as const
1634})