| ... | @@ -74,6 +74,8 @@ interface EnvironmentQueryAtomOptions<Input, A, E, R> extends Omit< | ... | @@ -74,6 +74,8 @@ interface EnvironmentQueryAtomOptions<Input, A, E, R> extends Omit< |
| 74 | readonly environmentId: EnvironmentIdType; | 74 | readonly environmentId: EnvironmentIdType; |
| 75 | readonly input: Input; | 75 | readonly input: Input; |
| 76 | }) => Atom.Atom<unknown> | undefined; | 76 | }) => Atom.Atom<unknown> | undefined; |
| | 77 | /** A value that can no longer change. It stops the interval and `refreshTrigger` re-reads. */ |
| | 78 | readonly isFinal?: (value: A) => boolean; |
| 77 | } | 79 | } |
| 78 | | 80 | |
| 79 | interface EnvironmentSubscriptionAtomOptions<Input, A, E, R> { | 81 | interface EnvironmentSubscriptionAtomOptions<Input, A, E, R> { |
| ... | @@ -485,8 +487,8 @@ function runInEnvironment<A, E, R>( | ... | @@ -485,8 +487,8 @@ function runInEnvironment<A, E, R>( |
| 485 | * refreshes on its next read. | 487 | * refreshes on its next read. |
| 486 | */ | 488 | */ |
| 487 | const refreshOnSignalWhileRead = | 489 | const refreshOnSignalWhileRead = |
| 488 | (signal: Atom.Atom<unknown>) => | 490 | <V>(signal: Atom.Atom<unknown>, isFinal: (value: V) => boolean) => |
| 489 | <A>(self: Atom.Atom<A>): Atom.Atom<A> => { | 491 | <A extends V>(self: Atom.Atom<A>): Atom.Atom<A> => { |
| 490 | // The signal value each registry's data was read under. Held outside the node because the | 492 | // The signal value each registry's data was read under. Held outside the node because the |
| 491 | // registry can sweep this node while `self` keeps the data, and the rebuilt node must still | 493 | // registry can sweep this node while `self` keeps the data, and the rebuilt node must still |
| 492 | // see a change it missed. | 494 | // see a change it missed. |
| ... | @@ -499,13 +501,31 @@ const refreshOnSignalWhileRead = | ... | @@ -499,13 +501,31 @@ const refreshOnSignalWhileRead = |
| 499 | const changed = | 501 | const changed = |
| 500 | readUnder.has(get.registry) && !Object.is(readUnder.get(get.registry), current); | 502 | readUnder.has(get.registry) && !Object.is(readUnder.get(get.registry), current); |
| 501 | readUnder.set(get.registry, current); | 503 | readUnder.set(get.registry, current); |
| 502 | if (changed) get.refresh(self); | 504 | if (changed && !isFinal(get.once(self))) get.refresh(self); |
| 503 | return get.once(self); | 505 | return get.once(self); |
| 504 | }, | 506 | }, |
| 505 | { initialValueTarget: self }, | 507 | { initialValueTarget: self }, |
| 506 | ); | 508 | ); |
| 507 | }; | 509 | }; |
| 508 | | 510 | |
| | 511 | /** `Atom.withRefresh`, except the timer stops once the value is final. */ |
| | 512 | const refreshEveryUntilFinal = |
| | 513 | <V>(intervalMs: number, isFinal: (value: V) => boolean) => |
| | 514 | <A extends V>(self: Atom.Atom<A>): Atom.Atom<A> => |
| | 515 | Atom.transform( |
| | 516 | self, |
| | 517 | (get) => { |
| | 518 | const value = get(self); |
| | 519 | if (!isFinal(value)) { |
| | 520 | // @effect-diagnostics-next-line globalTimers:off - Mirrors `Atom.withRefresh`, outside an Effect runtime. |
| | 521 | const handle = setTimeout(() => get.refresh(self), intervalMs); |
| | 522 | get.addFinalizer(() => clearTimeout(handle)); |
| | 523 | } |
| | 524 | return value; |
| | 525 | }, |
| | 526 | { initialValueTarget: self }, |
| | 527 | ); |
| | 528 | |
| 509 | export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>( | 529 | export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>( |
| 510 | runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>, | 530 | runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>, |
| 511 | options: EnvironmentQueryAtomOptions< | 531 | options: EnvironmentQueryAtomOptions< |
| ... | @@ -626,15 +646,17 @@ export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>( | ... | @@ -626,15 +646,17 @@ export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>( |
| 626 | }), | 646 | }), |
| 627 | Atom.setIdleTTL(idleTtlMs), | 647 | Atom.setIdleTTL(idleTtlMs), |
| 628 | ); | 648 | ); |
| | 649 | const isFinal = (result: AsyncResult.AsyncResult<A, unknown>) => |
| | 650 | options.isFinal !== undefined && Option.exists(AsyncResult.value(result), options.isFinal); |
| 629 | const intervalQuery = | 651 | const intervalQuery = |
| 630 | options.refreshIntervalMs === undefined | 652 | options.refreshIntervalMs === undefined |
| 631 | ? queryAtom | 653 | ? queryAtom |
| 632 | : queryAtom.pipe(Atom.withRefresh(options.refreshIntervalMs)); | 654 | : queryAtom.pipe(refreshEveryUntilFinal(options.refreshIntervalMs, isFinal)); |
| 633 | const refreshTrigger = options.refreshTrigger?.(target); | 655 | const refreshTrigger = options.refreshTrigger?.(target); |
| 634 | return ( | 656 | return ( |
| 635 | refreshTrigger === undefined | 657 | refreshTrigger === undefined |
| 636 | ? intervalQuery | 658 | ? intervalQuery |
| 637 | : intervalQuery.pipe(refreshOnSignalWhileRead(refreshTrigger)) | 659 | : intervalQuery.pipe(refreshOnSignalWhileRead(refreshTrigger, isFinal)) |
| 638 | ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`)); | 660 | ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`)); |
| 639 | }); | 661 | }); |
| 640 | return (target) => family(environmentRpcKey(target)); | 662 | return (target) => family(environmentRpcKey(target)); |
| ... | @@ -716,6 +738,7 @@ export function createEnvironmentRpcQueryAtomFamily< | ... | @@ -716,6 +738,7 @@ export function createEnvironmentRpcQueryAtomFamily< |
| 716 | readonly environmentId: EnvironmentIdType; | 738 | readonly environmentId: EnvironmentIdType; |
| 717 | readonly input: EnvironmentRpcInput<TTag>; | 739 | readonly input: EnvironmentRpcInput<TTag>; |
| 718 | }) => Atom.Atom<unknown> | undefined; | 740 | }) => Atom.Atom<unknown> | undefined; |
| | 741 | readonly isFinal?: (value: EnvironmentRpcSuccess<TTag>) => boolean; |
| 719 | }, | 742 | }, |
| 720 | ) { | 743 | ) { |
| 721 | return createEnvironmentQueryAtomFamily(runtime, { | 744 | return createEnvironmentQueryAtomFamily(runtime, { |
| ... | @@ -727,6 +750,7 @@ export function createEnvironmentRpcQueryAtomFamily< | ... | @@ -727,6 +750,7 @@ export function createEnvironmentRpcQueryAtomFamily< |
| 727 | ? {} | 750 | ? {} |
| 728 | : { refreshIntervalMs: options.refreshIntervalMs }), | 751 | : { refreshIntervalMs: options.refreshIntervalMs }), |
| 729 | ...(options.refreshTrigger === undefined ? {} : { refreshTrigger: options.refreshTrigger }), | 752 | ...(options.refreshTrigger === undefined ? {} : { refreshTrigger: options.refreshTrigger }), |
| | 753 | ...(options.isFinal === undefined ? {} : { isFinal: options.isFinal }), |
| 730 | execute: ( | 754 | execute: ( |
| 731 | input: EnvironmentRpcInput<TTag>, | 755 | input: EnvironmentRpcInput<TTag>, |
| 732 | emit: (value: EnvironmentRpcSuccess<TTag>) => Effect.Effect<void>, | 756 | emit: (value: EnvironmentRpcSuccess<TTag>) => Effect.Effect<void>, |