refactor

Author: ENvironmentSet

core/src/aggregators/SyncAggregator.ts

@@ -1,36 +1,37 @@
-import { produceEffects } from "produceEffects";
 import { aggregate } from "../aggregate";
 import type { Effect } from "../Effect";
 import type { DomainEvent } from "../event-types";
+import { produceEffects } from "../produceEffects";
 import type { Stack } from "../Stack";
+import { delay } from "../utils/delay";
 import type { Publisher } from "../utils/publishers/Publisher";
 import type { Scheduler } from "../utils/schedulers/Scheduler";
 import type { Aggregator } from "./Aggregator";
 
 export class SyncAggregator implements Aggregator {
   private events: DomainEvent[];
+  private latestStackSnapshot: Stack;
   private changePublisher: Publisher<{ effects: Effect[]; stack: Stack }>;
   private updateScheduler: Scheduler;
-  private previousStack: Stack;
 
-  constructor(
-    events: DomainEvent[],
-    changePublisher: Publisher<{ effects: Effect[]; stack: Stack }>,
-    updateScheduler: Scheduler,
-  ) {
-    this.events = events;
-    this.changePublisher = changePublisher;
-    this.updateScheduler = updateScheduler;
-    this.previousStack = this.computeStack();
+  constructor(options: {
+    initialEvents: DomainEvent[];
+    changePublisher: Publisher<{ effects: Effect[]; stack: Stack }>;
+    updateScheduler: Scheduler;
+  }) {
+    this.events = options.initialEvents;
+    this.latestStackSnapshot = aggregate(this.events, Date.now());
+    this.changePublisher = options.changePublisher;
+    this.updateScheduler = options.updateScheduler;
   }
 
   getStack(): Stack {
-    return this.previousStack;
+    return this.latestStackSnapshot;
   }
 
   dispatchEvent(event: DomainEvent): void {
     this.events.push(event);
-    this.updateStack();
+    this.updateSnapshot();
   }
 
   subscribeChanges(
@@ -41,15 +42,41 @@ export class SyncAggregator implements Aggregator {
     });
   }
 
-  private computeStack(): Stack {
-    return aggregate(this.events, Date.now());
+  private updateSnapshot(): void {
+    const previousSnapshot = this.latestStackSnapshot;
+    const currentSnapshot = aggregate(this.events, Date.now());
+    const effects = produceEffects(previousSnapshot, currentSnapshot);
+
+    if (effects.length > 0) {
+      this.latestStackSnapshot = currentSnapshot;
+      this.changePublisher.publish({
+        effects,
+        stack: this.latestStackSnapshot,
+      });
+    }
+
+    const earliestUpcomingTransitionStateUpdate =
+      this.calculateEarliestUpcomingTransitionStateUpdate();
+
+    if (earliestUpcomingTransitionStateUpdate) {
+      this.updateScheduler.schedule(async (options) => {
+        await delay(
+          earliestUpcomingTransitionStateUpdate.timestamp - Date.now(),
+          { signal: options?.signal },
+        );
+
+        if (options?.signal?.aborted) return;
+
+        this.updateSnapshot();
+      });
+    }
   }
 
-  private predictUpcomingTransitionStateUpdate(): {
+  private calculateEarliestUpcomingTransitionStateUpdate(): {
     event: DomainEvent;
     timestamp: number;
   } | null {
-    const activeActivities = this.previousStack.activities.filter(
+    const activeActivities = this.latestStackSnapshot.activities.filter(
       (activity) =>
         activity.transitionState === "enter-active" ||
         activity.transitionState === "exit-active",
@@ -67,25 +94,4 @@ export class SyncAggregator implements Aggregator {
         }
       : null;
   }
-
-  private updateStack(): void {
-    const previousStack = this.previousStack;
-    const currentStack = this.computeStack();
-    const effects = produceEffects(previousStack, currentStack);
-
-    if (effects.length > 0) {
-      this.changePublisher.publish({ effects, stack: currentStack });
-
-      this.previousStack = currentStack;
-
-      const upcomingTransitionStateUpdate =
-        this.predictUpcomingTransitionStateUpdate();
-
-      if (upcomingTransitionStateUpdate) {
-        this.updateScheduler.schedule(async () => {
-          this.updateStack();
-        });
-      }
-    }
-  }
 }

core/src/makeCoreStore.ts

@@ -73,11 +73,13 @@ export function makeCoreStore(options: MakeCoreStoreOptions): CoreStore {
     options.handlers?.onInitialActivityNotFound?.();
   }
 
-  const aggregator: Aggregator = new SyncAggregator(
-    [...initialRemainingEvents, ...initialPushedEvents],
-    new ScheduledPublisher(new SequentialScheduler(new Mutex())),
-    new SwitchScheduler(new Mutex()),
-  );
+  const aggregator: Aggregator = new SyncAggregator({
+    initialEvents: [...initialRemainingEvents, ...initialPushedEvents],
+    changePublisher: new ScheduledPublisher(
+      new SequentialScheduler(new Mutex()),
+    ),
+    updateScheduler: new SwitchScheduler(new Mutex()),
+  });
 
   aggregator.subscribeChanges((effects) => {
     triggerPostEffectHooks(effects, pluginInstances, actions);

core/src/utils/Mutex.ts

@@ -15,7 +15,12 @@ export class Mutex {
         reject(getAbortReason(signal));
       };
       const lockWaiter = (lockHandle: LockHandle) => {
-        resolve(lockHandle);
+        if (signal?.aborted) {
+          reject(getAbortReason(signal));
+          lockHandle.release();
+        } else {
+          resolve(lockHandle);
+        }
 
         signal?.removeEventListener("abort", abortHandler);
       };

core/src/utils/delay.ts

@@ -0,0 +1,27 @@
+import { getAbortReason } from "./getAbortReason";
+
+export function delay(
+  ms: number,
+  options?: { signal?: AbortSignal },
+): Promise<void> {
+  return new Promise((resolve, reject) => {
+    const signal = options?.signal;
+
+    if (signal?.aborted) throw getAbortReason(signal);
+
+    const abortHandler = () => {
+      if (!signal) return;
+
+      clearTimeout(timeoutId);
+      reject(getAbortReason(signal));
+    };
+    const timeoutId = setTimeout(() => {
+      if (signal?.aborted) abortHandler();
+      else resolve();
+
+      signal?.removeEventListener("abort", abortHandler);
+    }, ms);
+
+    signal?.addEventListener("abort", abortHandler, { once: true });
+  });
+}