1import * as Context from "../Context.ts"
2import * as Duration from "../Duration.ts"
3import * as Effect from "../Effect.ts"
4import * as Exit from "../Exit.ts"
5import * as Fiber from "../Fiber.ts"
6import { identity } from "../Function.ts"
7import { pipeArguments } from "../Pipeable.ts"
8import type * as RcRef from "../RcRef.ts"
9import * as Scope from "../Scope.ts"
10
11const TypeId = "~effect/RcRef"
12
13type State<A> = State.Empty | State.Acquiring<A> | State.Acquired<A> | State.Closed
14
15declare namespace State {
16 interface Empty {
17 readonly _tag: "Empty"
18 }
19
20 interface Acquiring<A> {
21 readonly _tag: "Acquiring"
22 // Assigned as soon as the acquisition is forked.
23 fiber: Fiber.Fiber<Acquired<A>, unknown>
24 awaiters: number
25 // Set once references are allocated to the waiters, which can happen
26 // before the acquisition fiber exits.
27 acquired: Acquired<A> | undefined
28 }
29
30 interface Acquired<A> {
31 readonly _tag: "Acquired"
32 readonly value: A
33 readonly scope: Scope.Closeable
34 fiber: Fiber.Fiber<void, never> | undefined
35 refCount: number
36 invalidated: boolean
37 }
38
39 interface Closed {
40 readonly _tag: "Closed"
41 }
42}
43
44const stateEmpty: State<never> = { _tag: "Empty" }
45const stateClosed: State<never> = { _tag: "Closed" }
46
47const variance: RcRef.RcRef.Variance<any, any> = {
48 _A: identity,
49 _E: identity
50}
51
52class RcRefImpl<A, E> implements RcRef.RcRef<A, E> {
53 readonly [TypeId]: RcRef.RcRef.Variance<A, E> = variance
54
55 pipe() {
56 return pipeArguments(this, arguments)
57 }
58
59 state: State<A> = stateEmpty
60 readonly acquire: Effect.Effect<A, E>
61 readonly context: Context.Context<never>
62 readonly scope: Scope.Scope
63 readonly idleTimeToLive: Duration.Duration | undefined
64
65 constructor(
66 acquire: Effect.Effect<A, E>,
67 context: Context.Context<never>,
68 scope: Scope.Scope,
69 idleTimeToLive: Duration.Duration | undefined
70 ) {
71 this.acquire = acquire
72 this.context = context
73 this.scope = scope
74 this.idleTimeToLive = idleTimeToLive
75 }
76}
77
78/** @internal */
79export const make = <A, E, R>(options: {
80 readonly acquire: Effect.Effect<A, E, R>
81 readonly idleTimeToLive?: Duration.Input | undefined
82}) =>
83 Effect.withFiber<RcRef.RcRef<A, E>, never, R | Scope.Scope>((fiber) => {
84 const context = fiber.context as Context.Context<R | Scope.Scope>
85 const scope = Context.get(context, Scope.Scope)
86 const ref = new RcRefImpl<A, E>(
87 options.acquire as Effect.Effect<A, E>,
88 context,
89 scope,
90 options.idleTimeToLive !== undefined
91 ? Duration.fromInputUnsafe(options.idleTimeToLive)
92 : undefined
93 )
94 return Effect.as(
95 Scope.addFinalizerExit(scope, () => {
96 const state = ref.state
97 ref.state = stateClosed
98 if (state._tag === "Acquired") return Scope.close(state.scope, Exit.void)
99 if (state._tag === "Acquiring") return Fiber.interrupt(state.fiber)
100 return Effect.void
101 }),
102 ref
103 )
104 })
105
106const getState = <A, E>(
107 self: RcRefImpl<A, E>,
108 restore: <AX, EX, RX>(effect: Effect.Effect<AX, EX, RX>) => Effect.Effect<AX, EX, RX>
109): Effect.Effect<State.Acquired<A>, E> => {
110 switch (self.state._tag) {
111 case "Closed": {
112 return Effect.interrupt
113 }
114 case "Acquired": {
115 const state = self.state
116 state.refCount++
117 return state.fiber
118 ? Effect.as(Fiber.interrupt(state.fiber), state)
119 : Effect.succeed(state)
120 }
121 case "Acquiring": {
122 const state = self.state
123 state.awaiters++
124 return awaitAcquiring(self, state, restore)
125 }
126 case "Empty": {
127 // The caller that starts the acquisition counts as its first awaiter.
128 const acquiring: State.Acquiring<A> = {
129 _tag: "Acquiring",
130 fiber: undefined as any,
131 awaiters: 1,
132 acquired: undefined
133 }
134 self.state = acquiring
135 const scope = Scope.makeUnsafe()
136 // Acquire on a detached fiber so the acquisition is shared by every
137 // caller and only interrupted once all of them have been interrupted.
138 const acquire = Effect.provideContext(
139 self.acquire as Effect.Effect<A, E>,
140 Context.add(self.context, Scope.Scope, scope)
141 ).pipe(
142 Effect.flatMap((value) => {
143 // The ref was closed, or the acquisition was abandoned.
144 if (self.state !== acquiring) {
145 return Effect.interrupt
146 }
147 const state: State.Acquired<A> = {
148 _tag: "Acquired",
149 value,
150 scope,
151 fiber: undefined,
152 // Every caller still waiting holds a reference.
153 refCount: acquiring.awaiters,
154 invalidated: false
155 }
156 self.state = state
157 acquiring.acquired = state
158 return Effect.succeed(state)
159 }),
160 Effect.onExit((exit) => {
161 if (Exit.isSuccess(exit)) return Effect.void
162 if (self.state === acquiring) self.state = stateEmpty
163 return Scope.close(scope, exit)
164 })
165 )
166 acquiring.fiber = Effect.runForkWith(Fiber.getCurrent()!.context)(acquire)
167 return awaitAcquiring(self, acquiring, restore)
168 }
169 }
170}
171
172const awaitAcquiring = <A, E>(
173 self: RcRefImpl<A, E>,
174 acquiring: State.Acquiring<A>,
175 restore: <AX, EX, RX>(effect: Effect.Effect<AX, EX, RX>) => Effect.Effect<AX, EX, RX>
176): Effect.Effect<State.Acquired<A>, E> =>
177 Effect.onExit(
178 restore(Fiber.join(acquiring.fiber)) as Effect.Effect<State.Acquired<A>, E>,
179 (exit) => {
180 const acquired = acquiring.acquired
181 if (acquired !== undefined) {
182 // The acquisition counted this caller as a reference, possibly before
183 // its fiber exited. Release it if the caller was interrupted before it
184 // could register its finalizer.
185 return Exit.isFailure(exit) ? release(self, acquired) : Effect.void
186 }
187 // Abandon the acquisition once every caller has left, unless it already
188 // failed (a failed acquisition allocated no references).
189 if (--acquiring.awaiters > 0 || self.state !== acquiring) return Effect.void
190 // Detach it first so new callers start a fresh acquisition instead of
191 // joining one that is being interrupted.
192 self.state = stateEmpty
193 return Fiber.interrupt(acquiring.fiber)
194 }
195 )
196
197/** @internal */
198export const get = <A, E>(self_: RcRef.RcRef<A, E>): Effect.Effect<A, E, Scope.Scope> => {
199 const self = self_ as RcRefImpl<A, E>
200 return Effect.uninterruptibleMask((restore) =>
201 Effect.flatMap(
202 Effect.scope,
203 (scope) =>
204 Effect.flatMap(getState(self, restore), (state) =>
205 Effect.as(Scope.addFinalizerExit(scope, () => release(self, state)), state.value))
206 )
207 )
208}
209
210const release = <A, E>(self: RcRefImpl<A, E>, state: State.Acquired<A>): Effect.Effect<void> => {
211 state.refCount--
212 if (state.refCount > 0) {
213 return Effect.void
214 }
215 const idleTimeToLive = self.idleTimeToLive
216 if (idleTimeToLive === undefined || state.invalidated) {
217 if (self.state === state) {
218 self.state = stateEmpty
219 }
220 return Scope.close(state.scope, Exit.void)
221 } else if (!Duration.isFinite(idleTimeToLive)) {
222 return Effect.void
223 }
224 state.fiber = Effect.uninterruptibleMask((restore) =>
225 Effect.flatMap(restore(Effect.sleep(idleTimeToLive)), () => {
226 if (self.state === state && state.refCount === 0) {
227 self.state = stateEmpty
228 return Scope.close(state.scope, Exit.void)
229 }
230 return Effect.void
231 })
232 ).pipe(
233 Effect.ensuring(Effect.sync(() => {
234 state.fiber = undefined
235 })),
236 Effect.runForkWith(self.context),
237 Fiber.runIn(self.scope)
238 )
239 return Effect.void
240}
241
242/** @internal */
243export const invalidate = <A, E>(
244 self_: RcRef.RcRef<A, E>
245): Effect.Effect<void> => {
246 const self = self_ as RcRefImpl<A, E>
247 return Effect.uninterruptible(Effect.suspend(() => {
248 if (self.state._tag !== "Acquired") {
249 return Effect.void
250 }
251 const state = self.state
252 self.state = stateEmpty
253 state.invalidated = true
254 if (state.refCount > 0) {
255 return Effect.void
256 }
257 state.fiber?.interruptUnsafe()
258 return Scope.close(state.scope, Exit.void)
259 }))
260}