feat: experimental interoperability implementation

Author: divdavem

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;
+};

src/internal/storeTrackingUsage.ts

@@ -1,7 +1,8 @@
+import { getActiveConsumer } from '../interop';
 import { type Flushable, inFlushUnused, planFlush } from './asyncFlush';
-import { RawStoreFlags } from './store';
+import { RawStoreFlags, type TansuInteropConsumer } from './store';
 import { checkNotInNotificationPhase, RawStoreWritable } from './storeWritable';
-import { activeConsumer, untrack } from './untrack';
+import { untrack } from './untrack';
 
 export abstract class RawStoreTrackingUsage<T> extends RawStoreWritable<T> implements Flushable {
   private extraUsages = 0;
@@ -36,24 +37,30 @@ export abstract class RawStoreTrackingUsage<T> extends RawStoreWritable<T> imple
 
   override get(): T {
     checkNotInNotificationPhase();
-    if (activeConsumer) {
-      return activeConsumer.addProducer(this);
-    } else {
-      this.extraUsages++;
-      try {
+    this.extraUsages++;
+    try {
+      const activeConsumer = getActiveConsumer();
+      const addTansuProducer = (activeConsumer as TansuInteropConsumer)?.addTansuProducer;
+      if (!addTansuProducer) {
+        // tansu calls updateValue in addTansuProducer, so we don't need to call it here
         this.updateValue();
         // 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 (this.flags & RawStoreFlags.DIRTY) {
           throw new Error('assert failed: store still dirty after updating it');
         }
-        return this.readValue();
-      } finally {
-        const extraUsages = --this.extraUsages;
-        if (extraUsages === 0) {
-          this.checkUnused();
-        }
+      }
+      if (addTansuProducer) {
+        addTansuProducer.call(activeConsumer, this);
+      } else if (activeConsumer) {
+        activeConsumer.addProducer(this);
+      }
+      return this.readValue();
+    } finally {
+      const extraUsages = --this.extraUsages;
+      if (extraUsages === 0) {
+        this.checkUnused();
       }
     }
   }

src/internal/storeWritable.ts

@@ -1,10 +1,10 @@
+import { beginBatch, getActiveConsumer, type Signal, type Watcher } from '../interop';
 import type { Subscriber, UnsubscribeFunction, UnsubscribeObject, Updater } from '../types';
-import { batch } from './batch';
 import { equal } from './equal';
 import type { Consumer, RawStore } from './store';
 import { RawStoreFlags } from './store';
 import { SubscribeConsumer } from './subscribeConsumer';
-import { activeConsumer } from './untrack';
+import { watchRawStore } from './watch';
 
 export let notificationPhase = false;
 
@@ -25,7 +25,7 @@ export interface ProducerConsumerLink<T> {
   skipMarkDirty: boolean;
 }
 
-export class RawStoreWritable<T> implements RawStore<T, ProducerConsumerLink<T>> {
+export class RawStoreWritable<T> implements RawStore<T, ProducerConsumerLink<T>>, Signal {
   constructor(protected value: T) {}
   flags = RawStoreFlags.NONE;
   private version = 0;
@@ -64,10 +64,9 @@ export class RawStoreWritable<T> implements RawStore<T, ProducerConsumerLink<T>>
     return res;
   }
 
-  updateLink(link: ProducerConsumerLink<T>): T {
+  updateLink(link: ProducerConsumerLink<T>): void {
     link.value = this.value;
     link.version = this.version;
-    return this.readValue();
   }
 
   registerConsumer(link: ProducerConsumerLink<T>): ProducerConsumerLink<T> {
@@ -115,12 +114,19 @@ export class RawStoreWritable<T> implements RawStore<T, ProducerConsumerLink<T>>
     checkNotInNotificationPhase();
     const same = this.equal(this.value, newValue);
     if (!same) {
-      batch(() => {
+      const endBatch = beginBatch();
+      let queueError;
+      try {
         this.value = newValue;
         this.version++;
         this.equalCache = null;
         this.increaseEpoch();
-      });
+      } finally {
+        queueError = endBatch();
+      }
+      if (queueError) {
+        throw queueError.error;
+      }
     }
   }
 
@@ -145,7 +151,8 @@ export class RawStoreWritable<T> implements RawStore<T, ProducerConsumerLink<T>>
 
   get(): T {
     checkNotInNotificationPhase();
-    return activeConsumer ? activeConsumer.addProducer(this) : this.readValue();
+    getActiveConsumer()?.addProducer(this);
+    return this.readValue();
   }
 
   readValue(): T {
@@ -159,4 +166,8 @@ export class RawStoreWritable<T> implements RawStore<T, ProducerConsumerLink<T>>
     unsubscriber.unsubscribe = unsubscriber;
     return unsubscriber;
   }
+
+  watchSignal(notify: () => void): Watcher {
+    return watchRawStore(this, notify);
+  }
 }