1/**
2 * Buffers RPC protocol messages until their receive loop starts.
3 *
4 * @stability unstable
5 * @since 4.0.0
6 */
7import type * as Context from "../Context.ts"
8import * as Effect from "../Effect.ts"
9import * as Semaphore from "../Semaphore.ts"
10import type { Protocol } from "./RpcClient.ts"
11import type { FromServerEncoded } from "./RpcMessage.ts"
12
13/**
14 * Buffers writes until `run` replays them with their original contexts and
15 * installs the writer. Restores the previous writer when `run` ends.
16 *
17 * @stability unstable
18 * @category services
19 * @since 4.0.0
20 */
21export const withRun = <
22 A extends {
23 readonly run: (f: (...args: Array<any>) => Effect.Effect<void>) => Effect.Effect<never>
24 }
25>() =>
26<EX, RX>(f: (write: Parameters<A["run"]>[0]) => Effect.Effect<Omit<A, "run">, EX, RX>): Effect.Effect<A, EX, RX> =>
27 Effect.suspend(() => {
28 const semaphore = Semaphore.makeUnsafe(1)
29 let buffer: Array<[Array<any>, Context.Context<never>]> = []
30 let write = (...args: Array<any>): Effect.Effect<void> =>
31 Effect.contextWith((context) => {
32 buffer.push([args, context])
33 return Effect.void
34 })
35 return Effect.map(f((...args) => write(...args)), (a) => ({
36 ...a,
37 run(f) {
38 return semaphore.withPermits(1)(Effect.gen(function*() {
39 const prev = write
40
41 for (const [args, context] of buffer) {
42 yield* Effect.provideContext(Effect.suspend(() => f(...args)), context)
43 }
44 buffer = []
45 write = f
46
47 return yield* Effect.onExit(Effect.never, () => {
48 write = prev
49 return Effect.void
50 })
51 }))
52 }
53 } as A))
54 })
55
56/**
57 * Builds an RPC client protocol service that tracks active client IDs and
58 * buffers server responses per client until that client's `run` handler is
59 * installed.
60 *
61 * @stability unstable
62 * @category services
63 * @since 4.0.0
64 */
65export const withRunClient = <EX, RX>(
66 f: (
67 write: (clientId: number, response: FromServerEncoded) => Effect.Effect<void>,
68 clientIds: ReadonlySet<number>
69 ) => Effect.Effect<Omit<Protocol["Service"], "run">, EX, RX>
70): Effect.Effect<Protocol["Service"], EX, RX> =>
71 Effect.suspend(() => {
72 const clientIds = new Set<number>()
73 const clientBuffers = new Map<number, Array<[FromServerEncoded, Context.Context<never>]>>()
74 const clientWrites = new Map<number, (data: FromServerEncoded) => Effect.Effect<void>>()
75 let write = (clientId: number, data: FromServerEncoded): Effect.Effect<void> =>
76 Effect.contextWith((context) => {
77 let buffer = clientBuffers.get(clientId)
78 if (!buffer) {
79 buffer = []
80 clientBuffers.set(clientId, buffer)
81 }
82 buffer.push([data, context])
83 return Effect.void
84 })
85 return Effect.map(
86 f((clientId, data) => {
87 const clientWrite = clientWrites.get(clientId)
88 if (clientWrite) {
89 return clientWrite(data)
90 }
91 return write(clientId, data)
92 }, clientIds),
93 (a) => ({
94 ...a,
95 run(clientId, f) {
96 return Effect.gen(function*() {
97 clientIds.add(clientId)
98 clientWrites.set(clientId, f)
99
100 const buffer = clientBuffers.get(clientId)
101 if (buffer) {
102 clientBuffers.delete(clientId)
103 for (const [args, context] of buffer) {
104 yield* Effect.provideContext(Effect.suspend(() => f(args)), context)
105 }
106 }
107
108 return yield* Effect.onExit(Effect.never, () => {
109 clientIds.delete(clientId)
110 clientWrites.delete(clientId)
111 return Effect.void
112 })
113 })
114 }
115 } satisfies Protocol["Service"])
116 )
117 })