fix: fixing asymmetric diamond dependency problem

Author: divdavem

src/index.spec.ts

@@ -664,6 +664,22 @@ describe('stores', () => {
       unsubscribe();
     });
 
+    it('should prevent the asymmetric diamond dependency problem', () => {
+      const a = writable(0);
+      const b = derived(a, (a) => `b${a}`);
+      const c = derived([a, b], ([a, b]) => `${a}${b}`);
+
+      const values: string[] = [];
+
+      const unsubscribe = c.subscribe((value) => {
+        values.push(value);
+      });
+      expect(values).toEqual(['0b0']);
+      a.set(1);
+      expect(values).toEqual(['0b0', '1b1']);
+      unsubscribe();
+    });
+
     it('should be able to use destructuring', () => {
       const store = writable(0);
       const derivedStore = derived(store, (value) => {
@@ -835,6 +851,22 @@ describe('stores', () => {
       unsubscribe();
     });
 
+    it('should not call again listeners when the value finally did not change', () => {
+      const a = writable(0);
+      const values: number[] = [];
+      const unsubscribe = a.subscribe((value) => {
+        values.push(value);
+      });
+      expect(values).toEqual([0]);
+      batch(() => {
+        a.set(0);
+        a.set(1);
+        a.set(0);
+      });
+      expect(values).toEqual([0]);
+      unsubscribe();
+    });
+
     it('should not call an unregistered listener', () => {
       const store = writable(0);
       const calls: number[] = [];

src/index.ts

@@ -20,6 +20,7 @@ export interface SubscriberObject<T> {
   error?: any;
   complete?: any;
   invalidate: () => void;
+  revalidate: () => void;
 }
 
 /**
@@ -110,8 +111,22 @@ const bind = <T>(object: T | null | undefined, fnName: keyof T) => {
 
 const toSubscriberObject = <T>(subscriber: Subscriber<T>): SubscriberObject<T> =>
   typeof subscriber === 'function'
-    ? { next: subscriber.bind(null), invalidate: noop }
-    : { next: bind(subscriber, 'next'), invalidate: bind(subscriber, 'invalidate') };
+    ? { next: subscriber.bind(null), invalidate: noop, revalidate: noop }
+    : {
+        next: bind(subscriber, 'next'),
+        invalidate: bind(subscriber, 'invalidate'),
+        revalidate: bind(subscriber, 'revalidate'),
+      };
+
+const callAllSubscribers = <T, U extends keyof SubscriberObject<T>>(
+  subscribers: Iterable<SubscriberObject<T>>,
+  fn: U,
+  ...args: Parameters<SubscriberObject<T>[U]>
+) => {
+  for (const subscriber of [...subscribers]) {
+    subscriber[fn](...args);
+  }
+};
 
 const returnThis = function <T>(this: T): T {
   return this;
@@ -123,8 +138,9 @@ const asReadable = <T>(store: Store<T>): Readable<T> => ({
   [symbolObservable]: returnThis,
 });
 
+const queueProcess = Symbol();
 let willProcessQueue = false;
-const queue = new Map<SubscriberObject<any>, any>();
+const queue = new Set<{ [queueProcess](): void }>();
 
 const callUnsubscribe = (unsubscribe: Unsubscriber) =>
   typeof unsubscribe === 'function' ? unsubscribe() : unsubscribe.unsubscribe();
@@ -172,9 +188,9 @@ export const batch = <T>(fn: () => T): T => {
     return fn();
   } finally {
     if (needsProcessQueue) {
-      for (const [subscriberObject, value] of queue) {
-        queue.delete(subscriberObject);
-        subscriberObject.next(value);
+      for (const store of queue) {
+        queue.delete(store);
+        store[queueProcess]();
       }
       willProcessQueue = false;
     }
@@ -233,12 +249,16 @@ export function get<T>(store: SubscribableStore<T>): T {
 export abstract class Store<T> implements Readable<T> {
   private _subscribers = new Set<SubscriberObject<T>>();
   private _cleanupFn: null | Unsubscriber = null;
+  private _notifiedValue: T;
+  private _invalidated = false;
 
   /**
    *
    * @param _value - Initial value of the store
    */
-  constructor(private _value: T) {}
+  constructor(private _value: T) {
+    this._notifiedValue = this._value;
+  }
 
   private _start() {
     this._cleanupFn = this.onUse() || noop;
@@ -252,28 +272,50 @@ export abstract class Store<T> implements Readable<T> {
     }
   }
 
+  private [queueProcess](): void {
+    const value = this._value;
+    this._invalidated = false;
+    if (notEqual(this._notifiedValue, value)) {
+      this._notifiedValue = value;
+      callAllSubscribers(this._subscribers, 'next', value);
+    } else {
+      callAllSubscribers(this._subscribers, 'revalidate');
+    }
+  }
+
+  protected invalidate(): void {
+    if (!this._invalidated) {
+      this._invalidated = true;
+      callAllSubscribers(this._subscribers, 'invalidate');
+    }
+  }
+
+  protected revalidate(): void {
+    if (this._invalidated) {
+      batch(() => {
+        queue.add(this as any);
+      });
+    }
+  }
+
   /**
    * Replaces store's state with the provided value.
    * Equivalent of {@link Writable.set}, but internal to the store.
    *
    * @param value - value to be used as the new state of a store.
    */
   protected set(value: T): void {
-    if (notEqual(this._value, value)) {
+    const different = notEqual(this._value, value);
+    if (different) {
       this._value = value;
       if (!this._cleanupFn) {
-        // subscriber not yet initialized
+        this._notifiedValue = this._value;
         return;
       }
-      batch(() => {
-        for (const subscriber of this._subscribers) {
-          const needInvalidate = !queue.has(subscriber);
-          queue.set(subscriber, value);
-          if (needInvalidate) {
-            subscriber.invalidate();
-          }
-        }
-      });
+      this.invalidate();
+    }
+    if (this._invalidated) {
+      this.revalidate();
     }
   }
 
@@ -323,7 +365,10 @@ export abstract class Store<T> implements Readable<T> {
     if (this._subscribers.size == 1) {
       this._start();
     }
-    subscriberObject.next(this._value);
+    subscriberObject.next(this._notifiedValue);
+    if (this._invalidated) {
+      subscriberObject.invalidate();
+    }
 
     const unsubscribe = () => {
       const removed = this._subscribers.delete(subscriberObject);
@@ -491,6 +536,7 @@ export abstract class DerivedStore<
   protected onUse(): Unsubscriber | void {
     let initDone = false;
     let pending = 0;
+    let changed = 0;
 
     const stores = this._stores;
     const isArray = Array.isArray(stores);
@@ -510,7 +556,8 @@ export abstract class DerivedStore<
     };
 
     const callDerive = () => {
-      if (initDone && !pending) {
+      if (initDone && !pending && changed) {
+        changed = 0;
         callCleanup();
         cleanupFn = this.derive(isArray ? dependantValues : dependantValues[0]) || noop;
       }
@@ -520,11 +567,17 @@ export abstract class DerivedStore<
       store.subscribe({
         next: (v) => {
           dependantValues[idx] = v;
+          changed |= 1 << idx;
           pending &= ~(1 << idx);
           callDerive();
         },
         invalidate: () => {
           pending |= 1 << idx;
+          this.invalidate();
+        },
+        revalidate: () => {
+          pending &= ~(1 << idx);
+          callDerive();
         },
       })
     );