Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/reset-function-idle-ttl.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@effect-atom/atom": patch
---

Ensure atom ttl is 0 for internal intermediate state atoms.
76 changes: 38 additions & 38 deletions packages/atom/src/Atom.ts
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,28 @@ export interface WriteContext<A> {
set<R, W>(this: WriteContext<A>, atom: Writable<R, W>, value: W): void
}

/**
* @since 1.0.0
* @category combinators
*/
export const setIdleTTL: {
(duration: Duration.DurationInput): <A extends Atom<any>>(self: A) => A
<A extends Atom<any>>(self: A, duration: Duration.DurationInput): A
} = dual<
(duration: Duration.DurationInput) => <A extends Atom<any>>(self: A) => A,
<A extends Atom<any>>(self: A, duration: Duration.DurationInput) => A
>(2, (self, durationInput) => {
const duration = Duration.decode(durationInput)
const isFinite = Duration.isFinite(duration)
return Object.assign(Object.create(Object.getPrototypeOf(self)), {
...self,
keepAlive: !isFinite,
idleTTL: isFinite ? Duration.toMillis(duration) : undefined
})
})

const removeTtl = setIdleTTL(0)

const AtomProto = {
[TypeId]: TypeId,
pipe() {
Expand Down Expand Up @@ -217,7 +239,7 @@ const RuntimeProto = {
readonly disableAccumulation?: boolean
readonly initialValue?: ReadonlyArray<any>
}) {
const pullSignal = state(0)
const pullSignal = removeTtl(state(0))
const pullAtom = readable((get) => {
const previous = get.self<Result.Result<any, any>>()
const runtimeResult = get(this)
Expand All @@ -236,7 +258,7 @@ const RuntimeProto = {

subscriptionRef(this: AtomRuntime<any, any>, ref: any) {
return makeSubRef(
readable((get) => {
removeTtl(readable((get) => {
const previous = get.self<Result.Result<any, any>>()
const runtimeResult = get(this)
if (runtimeResult._tag !== "Success") {
Expand All @@ -246,7 +268,7 @@ const RuntimeProto = {
return SubscriptionRef.SubscriptionRefTypeId in value
? value
: makeEffect(get, value, Result.initial(true), runtimeResult.value)
}),
})),
(get, ref) => {
const runtime = Result.getOrThrow(get(this))
return readSubscribable(get, ref, runtime)
Expand Down Expand Up @@ -677,11 +699,11 @@ export const context: (options: {
factory.addGlobalLayer = (layer: Layer.Layer<any, any, AtomRegistry | Reactivity.Reactivity>) => {
globalLayer = Layer.provideMerge(globalLayer, Layer.provide(layer, Reactivity.layer))
}
const reactivityAtom = make(
const reactivityAtom = removeTtl(make(
Effect.scopeWith((scope) => Layer.buildWithMemoMap(Reactivity.layer, options.memoMap, scope)).pipe(
Effect.map(EffectContext.get(Reactivity.Reactivity))
)
)
))
factory.withReactivity =
(keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>) =>
<A extends Atom<any>>(atom: A): A =>
Expand Down Expand Up @@ -1007,7 +1029,7 @@ export const fnSync: {
const makeFnSync = <Arg, A>(f: (arg: Arg, get: FnContext) => A, options?: {
readonly initialValue?: A
}): Writable<Option.Option<A> | A, Arg> => {
const argAtom = state<[number, Arg]>([0, undefined as any])
const argAtom = removeTtl(state<[number, Arg]>([0, undefined as any]))
const hasInitialValue = options?.initialValue !== undefined
return writable(function(get) {
;(get as any).isFn = true
Expand Down Expand Up @@ -1100,16 +1122,16 @@ function makeResultFn<Arg, E, A>(
readonly concurrent?: boolean | undefined
}
) {
const argAtom = state<[number, Arg | Interrupt]>([0, undefined as any])
const argAtom = removeTtl(state<[number, Arg | Interrupt]>([0, undefined as any]))
const initialValue = options?.initialValue !== undefined
? Result.success<A, E>(options.initialValue)
: Result.initial<A, E>()
const fibersAtom = options?.concurrent
? make((get) => {
? removeTtl(readable((get) => {
const fibers = new Set<Fiber.RuntimeFiber<any, any>>()
get.addFinalizer(() => fibers.forEach((f) => f.unsafeInterruptAsFork(FiberId.none)))
return fibers
})
}))
: undefined

function read(get: Context, runtime?: Runtime.Runtime<any>): Result.Result<A, E | NoSuchElementException> {
Expand Down Expand Up @@ -1202,12 +1224,10 @@ export const pull = <A, E>(
readonly disableAccumulation?: boolean | undefined
}
): Writable<PullResult<A, E>, void> => {
const pullSignal = state(0)
const pullAtom = readable(
makeRead(function(get) {
return makeStreamPullEffect(get, pullSignal, create, options)
})
)
const pullSignal = removeTtl(state(0))
const pullAtom = readable(makeRead(function(get) {
return makeStreamPullEffect(get, pullSignal, create, options)
}))
return makeStreamPull(pullSignal, pullAtom)
}

Expand Down Expand Up @@ -1452,26 +1472,6 @@ export const withLabel: {
label: [name, new Error().stack?.split("\n")[5] ?? ""]
}))

/**
* @since 1.0.0
* @category combinators
*/
export const setIdleTTL: {
(duration: Duration.DurationInput): <A extends Atom<any>>(self: A) => A
<A extends Atom<any>>(self: A, duration: Duration.DurationInput): A
} = dual<
(duration: Duration.DurationInput) => <A extends Atom<any>>(self: A) => A,
<A extends Atom<any>>(self: A, duration: Duration.DurationInput) => A
>(2, (self, durationInput) => {
const duration = Duration.decode(durationInput)
const isFinite = Duration.isFinite(duration)
return Object.assign(Object.create(Object.getPrototypeOf(self)), {
...self,
keepAlive: !isFinite,
idleTTL: isFinite ? Duration.toMillis(duration) : undefined
})
})

/**
* @since 1.0.0
* @category combinators
Expand Down Expand Up @@ -1593,12 +1593,12 @@ export const debounce: {
*/
export const optimistic = <A>(self: Atom<A>): Writable<A, Atom<Result.Result<A, unknown>>> => {
let counter = 0
const writeAtom = state(
const writeAtom = removeTtl(state(
[
counter,
undefined as any as Atom<Result.Result<A, unknown>>
] as const
)
))
return writable(
(get) => {
let lastValue = get.once(self)
Expand Down Expand Up @@ -1714,7 +1714,7 @@ export const optimisticFn: {
| ((set: (result: NoInfer<W>) => void) => AtomResultFn<OW, XA, XE>)
}
): AtomResultFn<OW, XA, XE> => {
const transition = state<Result.Result<W, unknown>>(Result.initial())
const transition = removeTtl(state<Result.Result<W, unknown>>(Result.initial()))
return fn((arg: OW, get) => {
let value = options.reducer(get(self), arg)
if (Result.isResult(value)) {
Expand Down
40 changes: 40 additions & 0 deletions packages/atom/test/Atom.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -748,6 +748,46 @@ describe("Atom", () => {
expect(r.get(state3)).toEqual(0)
})

it("idleTTL fn", async () => {
const fn = Atom.fn((n: number) => Effect.succeed(n + 1)).pipe(
Atom.setIdleTTL(0)
)
const r = Registry.make({ defaultIdleTTL: 2000 })

let result = r.get(fn)
assert(Result.isInitial(result))

r.set(fn, 1)
result = r.get(fn)
assert(Result.isSuccess(result))
expect(result.value).toEqual(2)

await new Promise((resolve) => resolve(null))

result = r.get(fn)
assert(Result.isInitial(result))
})

it("idleTTL fnSync", async () => {
const fn = Atom.fnSync((n: number) => n + 1).pipe(
Atom.setIdleTTL(0)
)
const r = Registry.make({ defaultIdleTTL: 2000 })

let result = r.get(fn)
assert(Option.isNone(result))

r.set(fn, 1)
result = r.get(fn)
assert(Option.isSome(result))
expect(result.value).toEqual(2)

await new Promise((resolve) => resolve(null))

result = r.get(fn)
assert(Option.isNone(result))
})

it("fn", async () => {
const count = Atom.fnSync((n: number) => n).pipe(Atom.keepAlive)
const r = Registry.make()
Expand Down