feat(core) atom.whileWatched (useEffect-like primitive) and createExternalStoreAtom

.changeset/ninety-actors-hide.md

@@ -0,0 +1,11 @@
+---
+'@tanstack/angular-store': minor
+'@tanstack/preact-store': minor
+'@tanstack/react-store': minor
+'@tanstack/solid-store': minor
+'@tanstack/vue-store': minor
+'@tanstack/store': minor
+'@tanstack/svelte-store': minor
+---
+
+Added atom.whileWatched(cb), store.whileWatched(cb), createExternalStoreAtom

docs/config.json

@@ -225,6 +225,10 @@
           "label": "StoreActionsFactory",
           "to": "reference/type-aliases/StoreActionsFactory"
         },
+        {
+          "label": "WatchedEffect",
+          "to": "reference/type-aliases/WatchedEffect"
+        },
         {
           "label": "batch",
           "to": "reference/functions/batch"
@@ -237,6 +241,10 @@
           "label": "createAtom",
           "to": "reference/functions/createAtom"
         },
+        {
+          "label": "createExternalStoreAtom",
+          "to": "reference/functions/createExternalStoreAtom"
+        },
         {
           "label": "createStore",
           "to": "reference/functions/createStore"

packages/store/src/alien.ts

@@ -38,10 +38,12 @@ export const enum ReactiveFlags {
 export function createReactiveSystem({
   update,
   notify,
+  watched,
   unwatched,
 }: {
   update(sub: ReactiveNode): boolean
   notify(sub: ReactiveNode): void
+  watched(sub: ReactiveNode): void
   unwatched(sub: ReactiveNode): void
 }) {
   return {
@@ -95,6 +97,7 @@ export function createReactiveSystem({
       prevSub.nextSub = newLink
     } else {
       dep.subs = newLink
+      watched(dep)
     }
   }
 

packages/store/src/atom.ts

@@ -26,11 +26,29 @@ export function toObserver<T>(
   }
 }
 
+/**
+ * Called when the atom is watched.
+ * Returns a cleanup function that will be called when the atom is unwatched.
+ */
+export type WatchedEffect = () => (() => void) | void | undefined
+
 interface InternalAtom<T> extends ReactiveNode {
   _snapshot: T
   _update: (getValue?: T | ((snapshot: T) => T)) => boolean
+
   get: () => T
   subscribe: (observerOrFn: Observer<T> | ((value: T) => void)) => Subscription
+  /**
+   * `effect` will be called while the atom is watched. `effect` may return a
+   * cleanup function, which will be called when the atom is unwatched.
+   *
+   * Returns a `stop` function which cancels the listener.
+   */
+  whileWatched: (effect: WatchedEffect) => () => void
+
+  _watched: boolean
+  _watchedSubs?: Array<WatchedEffect>
+  _watchedCleanups?: Array<(() => void) | void | undefined>
 }
 
 const queuedEffects: Array<Effect | undefined> = []
@@ -45,7 +63,21 @@ const { link, unlink, propagate, checkDirty, shallowPropagate } =
       queuedEffects[queuedEffectsLength++] = effect
       effect.flags &= ~ReactiveFlags.Watching
     },
+    watched(atom: InternalAtom<any>): void {
+      atom._watched = true
+      if (atom._watchedSubs?.length) {
+        atom._watchedCleanups = atom._watchedSubs.map((sub) => sub())
+      }
+    },
     unwatched(atom: InternalAtom<any>): void {
+      atom._watched = false
+      // not sure if this should go above or below the deps purge
+      // preorder vs postorder
+      if (atom._watchedCleanups?.length) {
+        atom._watchedCleanups.forEach((cleanup) => cleanup?.())
+        atom._watchedCleanups.length = 0
+      }
+
       if (atom.depsTail !== undefined) {
         atom.depsTail = undefined
         atom.flags = ReactiveFlags.Mutable | ReactiveFlags.Dirty
@@ -135,6 +167,43 @@ export function createAsyncAtom<T>(
   return atom
 }
 
+/**
+ * Like React.useSyncExternalStore: pulls external state into an atom.
+ * This can be used for interoperating with other state management libraries.
+ *
+ * ```ts
+ * import * as redux from "redux"
+ *
+ * const reduxStore = redux.createStore((state: number, action: number) => state + action, 0)
+ * const atom = createExternalStoreAtom(reduxStore.getState, reduxStore.subscribe)
+ *
+ * const timesTwo = createAtom(() => atom.get() * 2)
+ * timesTwo.subscribe((value) => {
+ *   console.log('timesTwo: ', value)
+ * })
+ *
+ * reduxStore.dispatch(1)
+ * // timesTwo: 2
+ * reduxStore.dispatch(1)
+ * // timesTwo: 4
+ * ```
+ */
+export function createExternalStoreAtom<T>(
+  getSnapshot: () => T,
+  subscribe: (onStoreChange: () => void) => () => void,
+  options?: AtomOptions<T>,
+): ReadonlyAtom<T> {
+  const trigger = createAtom(0)
+  const invalidate = () => trigger.set((n) => n + 1)
+  trigger.whileWatched(() => subscribe(invalidate))
+
+  return createAtom(() => {
+    // Return latest snapshot when `trigger` changes
+    trigger.get()
+    return getSnapshot()
+  }, options)
+}
+
 export function createAtom<T>(
   getValue: (prev?: NoInfer<T>) => T,
   options?: AtomOptions<T>,
@@ -153,6 +222,7 @@ export function createAtom<T>(
   // Create plain object atom
   const atom: InternalAtom<T> = {
     _snapshot: isComputed ? undefined! : valueOrFn,
+    _watched: false,
 
     subs: undefined,
     subsTail: undefined,
@@ -185,6 +255,25 @@ export function createAtom<T>(
         },
       }
     },
+
+    whileWatched(listener: WatchedEffect): () => void {
+      atom._watchedSubs ??= []
+      atom._watchedSubs.push(listener)
+      if (atom._watched) {
+        atom._watchedCleanups ??= []
+        atom._watchedCleanups.push(listener())
+      }
+      return () => {
+        if (!atom._watchedSubs) {
+          return
+        }
+        const index = atom._watchedSubs.indexOf(listener)
+        if (index !== -1) {
+          atom._watchedSubs.splice(index, 1)
+        }
+      }
+    },
+
     _update(getValue?: T | ((snapshot: T) => T)): boolean {
       const prevSub = activeSub
       const compare = options?.compare ?? Object.is

packages/store/src/signal.ts

@@ -60,6 +60,9 @@ const { link, unlink, propagate, checkDirty, shallowPropagate } =
         queued[insertIndex] = left
       }
     },
+    watched(_node) {
+      // whileWatched not implemented for signal.ts
+    },
     unwatched(node) {
       if (!(node.flags & ReactiveFlags.Mutable)) {
         effectScopeOper.call(node)

packages/store/src/store.ts

@@ -1,4 +1,5 @@
 import { createAtom, toObserver } from './atom'
+import type { WatchedEffect } from './atom'
 import type { Atom, Observer, Subscription } from './types'
 
 export type StoreAction = (...args: Array<any>) => any
@@ -54,6 +55,15 @@ export class Store<T, TActions extends StoreActionMap = never> {
   ): Subscription {
     return this.atom.subscribe(toObserver(observerOrFn))
   }
+  /**
+   * `effect` will be called while the atom is watched. `effect` may return a
+   * cleanup function, which will be called when the atom is unwatched.
+   *
+   * Returns a `stop` function which cancels the listener.
+   */
+  public whileWatched(effect: WatchedEffect): () => void {
+    return this.atom.whileWatched(effect)
+  }
 }
 
 export class ReadonlyStore<T> implements Omit<
@@ -81,6 +91,15 @@ export class ReadonlyStore<T> implements Omit<
   ): Subscription {
     return this.atom.subscribe(toObserver(observerOrFn))
   }
+  /**
+   * `effect` will be called while the atom is watched. `effect` may return a
+   * cleanup function, which will be called when the atom is unwatched.
+   *
+   * Returns a `stop` function which cancels the listener.
+   */
+  public whileWatched(effect: WatchedEffect): () => void {
+    return this.atom.whileWatched(effect)
+  }
 }
 
 export function createStore<T>(

packages/store/src/types.ts

@@ -1,3 +1,4 @@
+import type { WatchedEffect } from './atom'
 import type { ReactiveNode } from './alien'
 
 export type Selection<TSelected> = Readable<TSelected>
@@ -30,7 +31,36 @@ export interface Readable<T> extends Subscribable<T> {
   get: () => T
 }
 
-export interface BaseAtom<T> extends Subscribable<T>, Readable<T> {}
+export interface BaseAtom<T> extends Subscribable<T>, Readable<T> {
+  /**
+   * `effect` will be called when the first subscriber is added. `effect`
+   * may return a cleanup function, which will be called when the atom is
+   * unwatched.
+   *
+   * Returns a `cleanup` function which removes the `whileWatched` listener.
+   *
+   * This can be used to sync external resources into the atom, similar to
+   * `useLayoutEffect` in React.
+   *
+   * ```ts
+   * function createTicker(ms: number) {
+   *   const now = createAtom(Date.now())
+   *   const refresh = () => now.set(Date.now())
+   *   now.whileWatched(() => {
+   *     refresh()
+   *     const interval = setInterval(refresh, ms)
+   *     return () => clearInterval(interval)
+   *   })
+   *   return createAtom(() => now.get())
+   * }
+   * const ticker = createTicker(1000)
+   * ticker.subscribe(() => {
+   *   console.log('current time: ', ticker.get())
+   * })
+   * ```
+   */
+  whileWatched: (effect: WatchedEffect) => () => void
+}
 
 export interface InternalBaseAtom<T> extends Subscribable<T>, Readable<T> {
   /** @internal */