feat: watch and fromWatch functions

Author: divdavem

src/index.spec.ts

@@ -5,6 +5,8 @@ import { BehaviorSubject, from } from 'rxjs';
 import { writable as svelteWritable } from 'svelte/store';
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest';
 import type {
+  InteropWatcher,
+  InteropWatcherFactory,
   OnUseArgument,
   Readable,
   ReadableSignal,
@@ -24,10 +26,12 @@ import {
   computed,
   derived,
   equal,
+  fromWatch,
   get,
   readable,
   symbolObservable,
   untrack,
+  watch,
   writable,
 } from './index';
 import { rawStoreSymbol } from './internal/exposeRawStores';
@@ -3565,4 +3569,198 @@ describe('stores', () => {
       expectCorrectlyCleanedUp(doubleDoubleDoubleStore);
     });
   });
+
+  describe('watch', () => {
+    it('should work', () => {
+      const onUseCalls: { onUnused: number }[] = [];
+      const store = writable(0, {
+        onUse: () => {
+          const call = { onUnused: 0 };
+          onUseCalls.push(call);
+          return () => {
+            call.onUnused++;
+          };
+        },
+      });
+      const notify = vi.fn();
+      const watcher = watch(store, notify);
+      expect(watcher.isUpToDate()).toBe(false);
+      expect(onUseCalls.length).toBe(0);
+      expect(watcher.update()).toBe(true);
+      expect(onUseCalls.length).toBe(1);
+      expect(onUseCalls[0].onUnused).toBe(0);
+      expect(watcher.isUpToDate()).toBe(true);
+      expect(watcher.get()).toBe(0);
+      expect(notify).not.toHaveBeenCalled();
+      store.set(1);
+      expect(notify).toHaveBeenCalledOnce();
+      notify.mockClear();
+      expect(watcher.isUpToDate()).toBe(false);
+      store.set(2);
+      expect(notify).not.toHaveBeenCalled();
+      expect(watcher.update()).toBe(true);
+      expect(watcher.isUpToDate()).toBe(true);
+      expect(watcher.get()).toBe(2);
+      expect(notify).not.toHaveBeenCalled();
+      store.set(3);
+      expect(notify).toHaveBeenCalledOnce();
+      expect(watcher.isUpToDate()).toBe(false);
+      notify.mockClear();
+      store.set(4);
+      store.set(2);
+      expect(notify).not.toHaveBeenCalled();
+      expect(watcher.update()).toBe(false);
+      expect(watcher.isUpToDate()).toBe(true);
+      expect(watcher.update()).toBe(false);
+      expect(watcher.get()).toBe(2);
+      watcher.destroy();
+      expect(() => watcher.get()).toThrowError('invalid watcher state');
+    });
+  });
+
+  describe('fromWatch', () => {
+    it('should work', () => {
+      let notify: () => void;
+      let isUpdated = true;
+      let value = 0;
+      const watcher = {
+        update: vi.fn(() => isUpdated),
+        get: vi.fn(() => value),
+        destroy: vi.fn(),
+      } satisfies InteropWatcher<number>;
+      const watchFn = vi.fn((notifyFn: () => void) => {
+        notify = notifyFn;
+        return watcher;
+      }) satisfies InteropWatcherFactory<number>;
+      const clearMocks = () => {
+        watchFn.mockClear();
+        watcher.update.mockClear();
+        watcher.get.mockClear();
+        watcher.destroy.mockClear();
+      };
+      const store = fromWatch(watchFn);
+      expect(watchFn).not.toHaveBeenCalled();
+      expect(store.get()).toBe(0);
+      expect(watchFn).toHaveBeenCalledOnce();
+      expect(watcher.update).toHaveBeenCalledOnce();
+      expect(watcher.get).toHaveBeenCalledOnce();
+      expect(watcher.destroy).toHaveBeenCalledOnce();
+      clearMocks();
+      const values: number[] = [];
+      const unsubscribe = store.subscribe((value) => {
+        values.push(value);
+      });
+      expect(values).toEqual([0]);
+      expect(watchFn).toHaveBeenCalledOnce();
+      expect(watcher.update).toHaveBeenCalledOnce();
+      expect(watcher.get).toHaveBeenCalledOnce();
+      expect(watcher.destroy).not.toHaveBeenCalled();
+      clearMocks();
+      isUpdated = false;
+      batch(() => {
+        notify!();
+      });
+      expect(watcher.update).toHaveBeenCalledOnce();
+      expect(watchFn).not.toHaveBeenCalled();
+      expect(watcher.get).not.toHaveBeenCalled();
+      expect(watcher.destroy).not.toHaveBeenCalled();
+      clearMocks();
+      expect(values).toEqual([0]);
+      isUpdated = true;
+      value = 2;
+      batch(() => {
+        notify!();
+      });
+      expect(watcher.update).toHaveBeenCalledOnce();
+      expect(watcher.get).toHaveBeenCalledOnce();
+      expect(watcher.destroy).not.toHaveBeenCalled();
+      expect(watchFn).not.toHaveBeenCalled();
+      clearMocks();
+      expect(values).toEqual([0, 2]);
+      isUpdated = true;
+      watcher.get.mockImplementation(() => {
+        throw new Error('myerror');
+      });
+      expect(() => {
+        batch(() => {
+          notify!();
+        });
+      }).toThrowError('myerror');
+      expect(watcher.update).toHaveBeenCalledOnce();
+      expect(watcher.get).toHaveBeenCalledOnce();
+      expect(watcher.destroy).not.toHaveBeenCalled();
+      expect(watchFn).not.toHaveBeenCalled();
+      clearMocks();
+      expect(() => {
+        store.get();
+      }).toThrowError('myerror');
+      expect(values).toEqual([0, 2]);
+      expect(watcher.update).not.toHaveBeenCalled();
+      expect(watcher.get).not.toHaveBeenCalled();
+      expect(watcher.destroy).not.toHaveBeenCalled();
+      expect(watchFn).not.toHaveBeenCalled();
+      unsubscribe();
+      expect(watcher.destroy).toHaveBeenCalledOnce();
+      expect(watcher.update).not.toHaveBeenCalled();
+      expect(watcher.get).not.toHaveBeenCalled();
+      expect(watchFn).not.toHaveBeenCalled();
+    });
+  });
+
+  describe('watch / fromWatch', () => {
+    it('should work to convert back and forth a basic writable', () => {
+      const store = writable(0);
+      const otherStore = fromWatch((notify) => watch(store, notify));
+      expect(otherStore()).toBe(0);
+      store.set(1);
+      expect(otherStore()).toBe(1);
+      store.set(2);
+      expect(otherStore()).toBe(2);
+
+      const values: number[] = [];
+      const unsubscribe = otherStore.subscribe((value) => {
+        values.push(value);
+      });
+      expect(values).toEqual([2]);
+      store.set(3);
+      expect(values).toEqual([2, 3]);
+      expect(otherStore()).toBe(3);
+      batch(() => {
+        store.set(4);
+        expect(otherStore()).toBe(4);
+        store.set(5);
+        store.set(3);
+      });
+      expect(values).toEqual([2, 3]);
+      unsubscribe();
+    });
+
+    it('should work to convert back and forth a computed', () => {
+      const store = writable(0);
+      const doubleStore = computed(() => store() * 2);
+      const otherStore = fromWatch((notify) => watch(doubleStore, notify));
+      expect(otherStore()).toBe(0);
+      store.set(1);
+      expect(otherStore()).toBe(2);
+      store.set(2);
+      expect(otherStore()).toBe(4);
+
+      const values: number[] = [];
+      const unsubscribe = otherStore.subscribe((value) => {
+        values.push(value);
+      });
+      expect(values).toEqual([4]);
+      store.set(3);
+      expect(values).toEqual([4, 6]);
+      expect(otherStore()).toBe(6);
+      batch(() => {
+        store.set(4);
+        expect(otherStore()).toBe(8);
+        store.set(5);
+        store.set(3);
+      });
+      expect(values).toEqual([4, 6]);
+      unsubscribe();
+    });
+  });
 });

src/index.ts

@@ -24,9 +24,11 @@ import { RawStoreWithOnUse } from './internal/storeWithOnUse';
 import { RawStoreWritable } from './internal/storeWritable';
 import { noop } from './internal/subscribeConsumer';
 import { untrack } from './internal/untrack';
+import { WatcherConsumer } from './internal/watch';
 import type {
   AsyncDeriveFn,
   AsyncDeriveOptions,
+  InteropWatcherFactory,
   OnUseFn,
   Readable,
   ReadableSignal,
@@ -41,15 +43,18 @@ import type {
   UnsubscribeObject,
   Unsubscriber,
   Updater,
+  Watcher,
   Writable,
   WritableSignal,
 } from './types';
 
-export { batch } from './internal/batch';
+import { batch } from './internal/batch';
+import { RawStoreFromWatch } from './internal/storeFromWatch';
 export { equal } from './internal/equal';
 export { symbolObservable } from './internal/exposeRawStores';
 export { untrack } from './internal/untrack';
 export type * from './types';
+export { batch };
 
 /**
  * Returns a wrapper (for the given store) which only exposes the {@link ReadableSignal} interface.
@@ -509,3 +514,49 @@ export function computed<T>(
 ): ReadableSignal<T> {
   return exposeRawStore(applyStoreOptions(new RawStoreComputed(fn), options));
 }
+
+/**
+ * Creates a watcher on a store and returns it.
+ *
+ * @remarks
+ *
+ * A watcher calls synchronously its notify function when any store in the set of transitive dependencies of the watched store is changing.
+ *
+ * Note that the notify function must not read or write any store. It should not do any heavy task, it usually only schedules some work to be done later.
+ * It is called even inside {@link batch}.
+ *
+ * A watcher is initially created in the dirty state.
+ *
+ * When a watcher is in the dirty state, the notify function is not called until the {@link Watcher.update|update} method is called.
+ *
+ * The {@link Watcher.update|update} method clears the dirty state and updates the watched store, allowing the notify function to be called the next
+ * time any store in the set of transitive dependencies of the watched store changes.
+ *
+ * When a watcher is no longer needed, it should be destroyed by calling its {@link Watcher.destroy|destroy} method.
+ *
+ * @param store - store to watch
+ * @param notify - function that will be called synchronously when any store in the set of transitive dependencies of the watched store is changing
+ * @returns watcher object
+ */
+export function watch<T>(store: StoreInput<T>, notify: () => void): Watcher<T> {
+  const watcherConsumer = new WatcherConsumer(getRawStore(store), notify);
+  return {
+    isUpToDate: watcherConsumer.isUpToDate.bind(watcherConsumer),
+    update: watcherConsumer.update.bind(watcherConsumer),
+    get: watcherConsumer.get.bind(watcherConsumer),
+    destroy: watcherConsumer.destroy.bind(watcherConsumer),
+  };
+}
+
+/**
+ * Creates a tansu store from a watcher factory.
+ * @param watch - watcher factory
+ * @param options - store option. Allows to define the {@link StoreOptions.equal|equal} function, if needed
+ * @returns store containing the value returned by the computation function
+ */
+export function fromWatch<T>(
+  watch: InteropWatcherFactory<T>,
+  options?: Omit<StoreOptions<T>, 'onUse'>
+): ReadableSignal<T> {
+  return exposeRawStore(applyStoreOptions(new RawStoreFromWatch(watch), options));
+}

src/internal/batch.ts

@@ -1,6 +1,4 @@
-import type { SubscribeConsumer } from './subscribeConsumer';
-
-export const subscribersQueue: SubscribeConsumer<any, any>[] = [];
+export const subscribersQueue: { process(): void }[] = [];
 let willProcessQueue = false;
 
 /**
@@ -56,7 +54,7 @@ export const batch = <T>(fn: () => T): T => {
       while (subscribersQueue.length > 0) {
         const consumer = subscribersQueue.shift()!;
         try {
-          consumer.notify();
+          consumer.process();
         } catch (e) {
           // an error in one consumer should not impact others
           if (success) {

src/internal/store.ts

@@ -29,7 +29,8 @@ export interface RawStore<T, Link extends BaseLink<T> = BaseLink<T>>
   unregisterConsumer(link: Link): void;
   updateValue(): void;
   isLinkUpToDate(link: Link): boolean;
-  updateLink(link: Link): T;
+  updateLink(link: Link): void;
+  readValue(): T;
 }
 
 export const updateLinkProducerValue = <T>(link: BaseLink<T>): void => {

src/internal/storeComputed.ts

@@ -62,7 +62,8 @@ export class RawStoreComputed<T>
     if (producer.flags & RawStoreFlags.HAS_VISIBLE_ONUSE) {
       this.flags |= RawStoreFlags.HAS_VISIBLE_ONUSE;
     }
-    return producer.updateLink(link);
+    producer.updateLink(link);
+    return producer.readValue();
   }
 
   override startUse(): void {

src/internal/storeConst.ts

@@ -21,7 +21,8 @@ export class RawStoreConst<T> implements RawStore<T, BaseLink<T>> {
   isLinkUpToDate(_link: BaseLink<T>): boolean {
     return true;
   }
-  updateLink(_link: BaseLink<T>): T {
+  updateLink(_link: BaseLink<T>): void {}
+  readValue(): T {
     return this.value;
   }
   get(): T {

src/internal/storeDerived.ts

@@ -82,7 +82,11 @@ abstract class RawStoreDerived<T, S extends StoresInput>
   override recompute(): void {
     try {
       this.callCleanUpFn();
-      const values = this.producerLinks!.map((link) => link.producer.updateLink(link));
+      const values = this.producerLinks!.map((link) => {
+        const producer = link.producer;
+        producer.updateLink(link);
+        return producer.readValue();
+      });
       this.cleanUpFn = normalizeUnsubscribe(this.derive(this.arrayMode ? values : values[0]));
     } catch (error) {
       this.error = error;