Test

Author: anakin_karrot

core/src/Aggregator/SyncAggregator.ts

@@ -0,0 +1,128 @@
+import { produceEffects } from "produceEffects";
+import { aggregate } from "../aggregate";
+import type { Effect } from "../Effect";
+import type {
+  DomainEvent,
+  PoppedEvent,
+  PushedEvent,
+  ReplacedEvent,
+} from "../event-types";
+import type { Stack } from "../Stack";
+import type { Publisher } from "../utils/Publisher/Publisher";
+import type { Aggregator } from "./Aggregator";
+
+export class SyncAggregator implements Aggregator {
+  private events: DomainEvent[];
+  private changePublisher: Publisher<{ effects: Effect[]; stack: Stack }>;
+  private autoUpdateTask: DynamicallyScheduledTask;
+  private previousStack: Stack;
+
+  constructor(
+    events: DomainEvent[],
+    changePublisher: Publisher<{ effects: Effect[]; stack: Stack }>,
+  ) {
+    this.events = events;
+    this.changePublisher = changePublisher;
+    this.autoUpdateTask = new DynamicallyScheduledTask(async () => {
+      this.updateStack();
+    });
+    this.previousStack = this.computeStack();
+  }
+
+  getStack(): Stack {
+    return this.computeStack();
+  }
+
+  dispatchEvent(event: DomainEvent): void {
+    this.events.push(event);
+    this.updateStack();
+  }
+
+  subscribeChanges(
+    listener: (effects: Effect[], stack: Stack) => void,
+  ): () => void {
+    return this.changePublisher.subscribe(({ effects, stack }) => {
+      listener(effects, stack);
+    });
+  }
+
+  private computeStack(): Stack {
+    return aggregate(this.events, Date.now());
+  }
+
+  private predictUpcomingTransitionStateUpdate(): {
+    event: DomainEvent;
+    timestamp: number;
+  } | null {
+    const stack = this.computeStack();
+    const activeActivities = stack.activities.filter(
+      (activity) =>
+        activity.transitionState === "enter-active" ||
+        activity.transitionState === "exit-active",
+    );
+    const mostRecentlyActivatedActivity = activeActivities.sort(
+      (a, b) => b.estimatedTransitionEnd! - a.estimatedTransitionEnd!,
+    )[0];
+
+    return mostRecentlyActivatedActivity
+      ? {
+          event:
+            mostRecentlyActivatedActivity.exitedBy ??
+            mostRecentlyActivatedActivity.enteredBy,
+          timestamp: mostRecentlyActivatedActivity.estimatedTransitionEnd!,
+        }
+      : 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.autoUpdateTask.schedule(upcomingTransitionStateUpdate.timestamp);
+      }
+    }
+  }
+}
+
+class DynamicallyScheduledTask {
+  private task: () => Promise<void>;
+  private scheduleId: number | null;
+
+  constructor(task: () => Promise<void>) {
+    this.task = task;
+    this.scheduleId = null;
+  }
+
+  schedule(timestamp: number): void {
+    if (this.scheduleId !== null) {
+      clearTimeout(this.scheduleId);
+      this.scheduleId = null;
+    }
+
+    const timeoutId = setTimeout(
+      () => {
+        if (this.scheduleId !== timeoutId) return;
+        if (Date.now() < timestamp) {
+          this.schedule(timestamp);
+          return;
+        }
+
+        this.task();
+        this.scheduleId = null;
+      },
+      Math.max(0, timestamp - Date.now()),
+    );
+
+    this.scheduleId = timeoutId;
+  }
+}

core/src/Stack.ts

@@ -29,6 +29,7 @@ export type Activity = {
   id: string;
   name: string;
   transitionState: ActivityTransitionState;
+  estimatedTransitionEnd?: number;
   params: {
     [key: string]: string | undefined;
   };

core/src/activity-utils/makeActivitiesReducer.ts

@@ -27,6 +27,8 @@ export function makeActivitiesReducer({
     Pushed(activities: Activity[], event: PushedEvent): Activity[] {
       const isTransitionDone =
         now - (resumedAt ?? event.eventDate) >= transitionDuration;
+      const estimatedTransitionEnd =
+        (resumedAt ?? event.eventDate) + transitionDuration;
 
       const transitionState: ActivityTransitionState =
         event.skipEnterActiveState || isTransitionDone
@@ -37,7 +39,7 @@ export function makeActivitiesReducer({
 
       return [
         ...activities.slice(0, reservedIndex),
-        makeActivityFromEvent(event, transitionState),
+        makeActivityFromEvent(event, transitionState, estimatedTransitionEnd),
         ...activities.slice(reservedIndex + 1),
       ];
     },
@@ -48,6 +50,8 @@ export function makeActivitiesReducer({
     Replaced(activities: Activity[], event: ReplacedEvent): Activity[] {
       const isTransitionDone =
         now - (resumedAt ?? event.eventDate) >= transitionDuration;
+      const estimatedTransitionEnd =
+        (resumedAt ?? event.eventDate) + transitionDuration;
 
       const reservedIndex = findNewActivityIndex(activities, event);
 
@@ -60,7 +64,7 @@ export function makeActivitiesReducer({
 
       return [
         ...activities.slice(0, reservedIndex),
-        makeActivityFromEvent(event, transitionState),
+        makeActivityFromEvent(event, transitionState, estimatedTransitionEnd),
         ...activities.slice(reservedIndex + 1),
       ];
     },

core/src/activity-utils/makeActivityFromEvent.ts

@@ -4,11 +4,13 @@ import type { Activity, ActivityTransitionState } from "../Stack";
 export function makeActivityFromEvent(
   event: PushedEvent | ReplacedEvent,
   transitionState: ActivityTransitionState,
+  estimatedTransitionEnd: number,
 ): Activity {
   return {
     id: event.activityId,
     name: event.activityName,
     transitionState,
+    estimatedTransitionEnd,
     params: event.activityParams,
     context: event.activityContext,
     steps: [

core/src/activity-utils/makeActivityReducer.ts

@@ -49,6 +49,8 @@ export function makeActivityReducer(context: {
         ...activity,
         exitedBy: event,
         transitionState,
+        estimatedTransitionEnd:
+          (context.resumedAt ?? event.eventDate) + context.transitionDuration,
         params:
           transitionState === "exit-done"
             ? activity.steps[0].params

core/src/makeCoreStore.ts

@@ -1,13 +1,12 @@
-import isEqual from "react-fast-compare";
+import { ExclusiveTaskQueue } from "utils/TaskQueue/ExclusiveTaskQueue";
 import type { Aggregator } from "./Aggregator/Aggregator";
-import { aggregate } from "./aggregate";
+import { SyncAggregator } from "./Aggregator/SyncAggregator";
 import type { DomainEvent, PushedEvent, StepPushedEvent } from "./event-types";
 import { makeEvent } from "./event-utils";
 import type { StackflowActions, StackflowPlugin } from "./interfaces";
-import { produceEffects } from "./produceEffects";
-import type { Stack } from "./Stack";
 import { divideBy, once } from "./utils";
 import { makeActions } from "./utils/makeActions";
+import { QueuingPublisher } from "./utils/Publisher/QueuingPublisher";
 import { triggerPostEffectHooks } from "./utils/triggerPostEffectHooks";
 
 const SECOND = 1000;
@@ -77,7 +76,10 @@ export function makeCoreStore(options: MakeCoreStoreOptions): CoreStore {
     options.handlers?.onInitialActivityNotFound?.();
   }
 
-  const aggregator: Aggregator = undefined as any;
+  const aggregator: Aggregator = new SyncAggregator(
+    [...initialRemainingEvents, ...initialPushedEvents],
+    new QueuingPublisher(new ExclusiveTaskQueue()),
+  );
 
   aggregator.subscribeChanges((effects) => {
     triggerPostEffectHooks(effects, pluginInstances, actions);

core/src/produceEffects.ts

@@ -9,12 +9,14 @@ export function produceEffects(prevOutput: Stack, nextOutput: Stack): Effect[] {
 
   const somethingChanged = !isEqual(prevOutput, nextOutput);
 
-  if (somethingChanged) {
-    output.push({
-      _TAG: "%SOMETHING_CHANGED%",
-    });
+  if (!somethingChanged) {
+    return output;
   }
 
+  output.push({
+    _TAG: "%SOMETHING_CHANGED%",
+  });
+
   const isPaused =
     prevOutput.globalTransitionState !== "paused" &&
     nextOutput.globalTransitionState === "paused";