Interoperability with other signal libraries

package.json

@@ -75,7 +75,7 @@
     "tdd:ui": "vitest --ui",
     "benchmark": "vitest bench --run",
     "clean": "rm -rf dist temp",
-    "lint": "eslint {src,benchmarks}/{,**/}*.ts",
+    "lint": "eslint {src,test,benchmarks}/{,**/}*.ts",
     "build:rollup": "rollup --failAfterWarnings -c",
     "build:dts": "tsc -p tsconfig.d.json",
     "build": "npm run clean && npm run build:rollup && npm run build:dts",

src/internal/asyncFlush.ts

@@ -0,0 +1,43 @@
+import { RawStoreFlags } from './store';
+
+export interface Flushable {
+  flags: RawStoreFlags;
+  checkUnused(): void;
+}
+
+let flushUnusedQueue: Flushable[] | null = null;
+export let inFlushUnused = false;
+
+export const planFlush = (object: Flushable): void => {
+  if (!(object.flags & RawStoreFlags.FLUSH_PLANNED)) {
+    object.flags |= RawStoreFlags.FLUSH_PLANNED;
+    if (!flushUnusedQueue) {
+      flushUnusedQueue = [];
+      queueMicrotask(flushUnused);
+    }
+    flushUnusedQueue.push(object);
+  }
+};
+
+export const flushUnused = (): void => {
+  // Ignoring coverage for the following lines because, unless there is a bug in tansu (which would have to be fixed!)
+  // there should be no way to trigger this error.
+  /* v8 ignore next 3 */
+  if (inFlushUnused) {
+    throw new Error('assert failed: recursive flushUnused call');
+  }
+  inFlushUnused = true;
+  try {
+    const queue = flushUnusedQueue;
+    if (queue) {
+      flushUnusedQueue = null;
+      for (let i = 0, l = queue.length; i < l; i++) {
+        const producer = queue[i];
+        producer.flags &= ~RawStoreFlags.FLUSH_PLANNED;
+        producer.checkUnused();
+      }
+    }
+  } finally {
+    inFlushUnused = false;
+  }
+};

src/internal/batch.ts

@@ -1,7 +1,4 @@
-import type { SubscribeConsumer } from './subscribeConsumer';
-
-export const subscribersQueue: SubscribeConsumer<any, any>[] = [];
-let willProcessQueue = false;
+import { beginBatch } from '../interop';
 
 /**
  * Batches multiple changes to stores while calling the provided function,
@@ -44,33 +41,16 @@ let willProcessQueue = false;
  * ```
  */
 export const batch = <T>(fn: () => T): T => {
-  const needsProcessQueue = !willProcessQueue;
-  willProcessQueue = true;
-  let success = true;
   let res;
-  let error;
+  let queueError;
+  const endBatch = beginBatch();
   try {
     res = fn();
   } finally {
-    if (needsProcessQueue) {
-      while (subscribersQueue.length > 0) {
-        const consumer = subscribersQueue.shift()!;
-        try {
-          consumer.notify();
-        } catch (e) {
-          // an error in one consumer should not impact others
-          if (success) {
-            // will throw the first error
-            success = false;
-            error = e;
-          }
-        }
-      }
-      willProcessQueue = false;
-    }
+    queueError = endBatch();
   }
-  if (success) {
-    return res;
+  if (queueError) {
+    throw queueError.error;
   }
-  throw error;
+  return res;
 };

src/internal/store.ts

@@ -1,4 +1,9 @@
 import type { SignalStore, SubscribableStore } from '../types';
+import type { Consumer as InteropConsumer } from '../interop';
+
+export interface TansuInteropConsumer extends InteropConsumer {
+  addTansuProducer?<T>(producer: RawStore<T>): void;
+}
 
 export interface Consumer {
   markDirty(): void;
@@ -13,6 +18,8 @@ export const enum RawStoreFlags {
   // the following flags are used in RawStoreComputedOrDerived and derived classes
   COMPUTING = 1 << 3,
   DIRTY = 1 << 4,
+  // used in Watcher
+  START_CALLED = 1 << 5,
 }
 
 export interface BaseLink<T> {
@@ -29,7 +36,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

@@ -1,16 +1,17 @@
-import type { BaseLink, Consumer, RawStore } from './store';
+import { getActiveConsumer, setActiveConsumer, type Signal } from '../interop';
+import type { BaseLink, Consumer, RawStore, TansuInteropConsumer } from './store';
 import { RawStoreFlags, updateLinkProducerValue } from './store';
 import {
   COMPUTED_ERRORED,
   COMPUTED_UNSET,
   RawStoreComputedOrDerived,
 } from './storeComputedOrDerived';
+import { fromInteropSignal } from './storeFromWatch';
 import { epoch, notificationPhase } from './storeWritable';
-import { activeConsumer, setActiveConsumer, type ActiveConsumer } from './untrack';
 
 export class RawStoreComputed<T>
   extends RawStoreComputedOrDerived<T>
-  implements Consumer, ActiveConsumer
+  implements Consumer, TansuInteropConsumer
 {
   private producerIndex = 0;
   private producerLinks: BaseLink<any>[] = [];
@@ -26,30 +27,41 @@ export class RawStoreComputed<T>
 
   override updateValue(): void {
     const flags = this.flags;
-    if (flags & RawStoreFlags.START_USE_CALLED && this.epoch === epoch) {
+    if (
+      flags & RawStoreFlags.START_USE_CALLED &&
+      !(flags & RawStoreFlags.DIRTY) &&
+      !(flags & RawStoreFlags.COMPUTING)
+    ) {
       return;
     }
     super.updateValue();
     this.epoch = epoch;
   }
 
   override get(): T {
+    // FIXME: better test all cases of this optimization:
+    const flags = this.flags;
     if (
-      !activeConsumer &&
+      !getActiveConsumer() &&
       !notificationPhase &&
-      this.epoch === epoch &&
-      (!(this.flags & RawStoreFlags.HAS_VISIBLE_ONUSE) ||
-        this.flags & RawStoreFlags.START_USE_CALLED)
+      !(flags & RawStoreFlags.COMPUTING) &&
+      !(flags & RawStoreFlags.DIRTY) &&
+      (flags & RawStoreFlags.START_USE_CALLED ||
+        (this.epoch === epoch && !(flags & RawStoreFlags.HAS_VISIBLE_ONUSE)))
     ) {
       return this.readValue();
     }
     return super.get();
   }
 
-  addProducer<U, L extends BaseLink<U>>(producer: RawStore<U, L>): U {
+  addProducer(signal: Signal): void {
+    this.addTansuProducer(fromInteropSignal(signal));
+  }
+
+  addTansuProducer<U>(producer: RawStore<U>): void {
     const producerLinks = this.producerLinks;
     const producerIndex = this.producerIndex;
-    let link = producerLinks[producerIndex] as L | undefined;
+    let link = producerLinks[producerIndex] as BaseLink<U> | undefined;
     if (link?.producer !== producer) {
       if (link) {
         producerLinks.push(link); // push the existing link at the end (to be removed later)
@@ -62,7 +74,7 @@ export class RawStoreComputed<T>
     if (producer.flags & RawStoreFlags.HAS_VISIBLE_ONUSE) {
       this.flags |= RawStoreFlags.HAS_VISIBLE_ONUSE;
     }
-    return producer.updateLink(link);
+    producer.updateLink(link);
   }
 
   override startUse(): void {

src/internal/storeComputedOrDerived.ts

@@ -1,7 +1,7 @@
+import { setActiveConsumer } from '../interop';
 import type { Consumer } from './store';
 import { RawStoreFlags } from './store';
 import { RawStoreTrackingUsage } from './storeTrackingUsage';
-import { setActiveConsumer } from './untrack';
 
 const MAX_CHANGE_RECOMPUTES = 1000;
 
@@ -61,7 +61,7 @@ export abstract class RawStoreComputedOrDerived<T>
         do {
           iterations++;
           this.flags &= ~RawStoreFlags.DIRTY;
-          if (this.areProducersUpToDate()) {
+          if (this.areProducersUpToDate() && !(this.flags & RawStoreFlags.DIRTY)) {
             return;
           }
         } while (this.flags & RawStoreFlags.DIRTY && iterations < MAX_CHANGE_RECOMPUTES);

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;

src/internal/storeFromWatch.ts

@@ -0,0 +1,54 @@
+import type { Signal, Watcher } from '../interop';
+import { type RawStore, RawStoreFlags } from './store';
+import { RawStoreComputedOrDerived } from './storeComputedOrDerived';
+import { RawStoreWritable } from './storeWritable';
+
+export class RawStoreFromWatch extends RawStoreComputedOrDerived<undefined> {
+  // FIXME: remove HAS_VISIBLE_ONUSE and add a new HAS_INTEROP_DEPENDENCIES ? (so that watcher.stop is called asynchronously)
+  override flags = RawStoreFlags.HAS_VISIBLE_ONUSE | RawStoreFlags.DIRTY;
+  private watcher: Watcher;
+
+  constructor(interopSignal: Signal) {
+    super(undefined as any);
+    this.watcher = interopSignal.watchSignal(this.markDirty.bind(this));
+  }
+
+  override equal(): boolean {
+    return false;
+  }
+
+  override startUse(): void {
+    this.watcher.start();
+    this.flags |= RawStoreFlags.DIRTY;
+  }
+
+  override areProducersUpToDate(): boolean {
+    return !this.watcher!.update();
+  }
+
+  override recompute(): void {
+    this.set(undefined);
+  }
+
+  override endUse(): void {
+    this.watcher.stop();
+  }
+
+  protected override increaseEpoch(): void {
+    // do nothing
+  }
+}
+
+const interopSignals = new WeakMap<Signal, RawStoreFromWatch>();
+const tansuWatchSignal = RawStoreWritable.prototype.watchSignal;
+export const fromInteropSignal = (signal: Signal): RawStore<any> => {
+  if (signal.watchSignal === tansuWatchSignal) {
+    return signal as RawStoreWritable<any>;
+  }
+  let store = interopSignals.get(signal);
+  if (!store) {
+    store = new RawStoreFromWatch(signal);
+    interopSignals.set(signal, store);
+  }
+  return store;
+};