Merge pull request #84 from brendan-ch/codex/implement-eventemitter-for-updates

codex/implement-eventemitter-for-updates

src/notifications/schedulers/ETANotificationScheduler.ts

@@ -1,4 +1,4 @@
-import { ShuttleGetterRepository } from "../../repositories/shuttle/ShuttleGetterRepository";
+import { ShuttleGetterRepository, ShuttleRepositoryEvent } from "../../repositories/shuttle/ShuttleGetterRepository";
 import { IEta } from "../../entities/ShuttleRepositoryEntities";
 import { AppleNotificationSender, NotificationAlertArguments } from "../senders/AppleNotificationSender";
 import {
@@ -90,10 +90,10 @@ export class ETANotificationScheduler {
 
   // The following is a workaround for the constructor being called twice
   public startListeningForUpdates() {
-    this.shuttleRepository.subscribeToEtaUpdates(this.etaSubscriberCallback);
+    this.shuttleRepository.on(ShuttleRepositoryEvent.ETA_UPDATED, this.etaSubscriberCallback);
   }
 
   public stopListeningForUpdates() {
-    this.shuttleRepository.subscribeToEtaUpdates(this.etaSubscriberCallback);
+    this.shuttleRepository.off(ShuttleRepositoryEvent.ETA_UPDATED, this.etaSubscriberCallback);
   }
 }

src/repositories/shuttle/ShuttleGetterRepository.ts

@@ -1,9 +1,31 @@
 import { IEta, IOrderedStop, IRoute, IShuttle, IStop } from "../../entities/ShuttleRepositoryEntities";
+import type EventEmitter from "node:events";
+
+export const ShuttleRepositoryEvent = {
+  ETA_UPDATED: "etaUpdated",
+  ETA_REMOVED: "etaRemoved",
+  ETA_DATA_CLEARED: "etaDataCleared",
+} as const;
+
+export type ShuttleRepositoryEventName = typeof ShuttleRepositoryEvent[keyof typeof ShuttleRepositoryEvent];
+
+export type EtaRemovedEventPayload = IEta;
+export type EtaDataClearedEventPayload = IEta[];
+
+export interface ShuttleRepositoryEventPayloads {
+  [ShuttleRepositoryEvent.ETA_UPDATED]: IEta;
+  [ShuttleRepositoryEvent.ETA_REMOVED]: EtaRemovedEventPayload;
+  [ShuttleRepositoryEvent.ETA_DATA_CLEARED]: EtaDataClearedEventPayload;
+}
+
+export type ShuttleRepositoryEventListener<T extends ShuttleRepositoryEventName> = (
+  payload: ShuttleRepositoryEventPayloads[T],
+) => void;
 
 /**
  * Shuttle getter repository to be linked to a system.
  */
-export interface ShuttleGetterRepository {
+export interface ShuttleGetterRepository extends EventEmitter {
   getStops(): Promise<IStop[]>;
   getStopById(stopId: string): Promise<IStop | null>;
 
@@ -18,23 +40,11 @@ export interface ShuttleGetterRepository {
   getEtasForStopId(stopId: string): Promise<IEta[]>;
   getEtaForShuttleAndStopId(shuttleId: string, stopId: string): Promise<IEta | null>;
 
-  /**
-   * Subscribe to all updates in ETA data.
-   * The subscriber persists even if the ETA data does not
-   * exist within the repository, and may fire again
-   * if ETA data is restored.
-   * @param listener
-   */
-  subscribeToEtaUpdates(
-    listener: (eta: IEta) => void,
-  ): void;
-
-  /**
-   * Unsubscribe from all ETA updates for the given callback.
-   * Callback must be passed by reference.
-   * @param listener
-   */
-  unsubscribeFromEtaUpdates(listener: (eta: IEta) => void): void;
+  on<T extends ShuttleRepositoryEventName>(event: T, listener: ShuttleRepositoryEventListener<T>): this;
+  once<T extends ShuttleRepositoryEventName>(event: T, listener: ShuttleRepositoryEventListener<T>): this;
+  off<T extends ShuttleRepositoryEventName>(event: T, listener: ShuttleRepositoryEventListener<T>): this;
+  addListener<T extends ShuttleRepositoryEventName>(event: T, listener: ShuttleRepositoryEventListener<T>): this;
+  removeListener<T extends ShuttleRepositoryEventName>(event: T, listener: ShuttleRepositoryEventListener<T>): this;
 
   getOrderedStopByRouteAndStopId(routeId: string, stopId: string): Promise<IOrderedStop | null>;
 

src/repositories/shuttle/UnoptimizedInMemoryShuttleRepository.ts

@@ -1,21 +1,76 @@
+import EventEmitter from "node:events";
 import { ShuttleGetterSetterRepository } from "./ShuttleGetterSetterRepository";
 import { IEta, IOrderedStop, IRoute, IShuttle, IStop } from "../../entities/ShuttleRepositoryEntities";
 import { IEntityWithId } from "../../entities/SharedEntities";
+import {
+  ShuttleRepositoryEvent,
+  ShuttleRepositoryEventListener,
+  ShuttleRepositoryEventName,
+  ShuttleRepositoryEventPayloads,
+} from "./ShuttleGetterRepository";
 
 /**
  * An unoptimized in memory repository.
  * (I would optimize it with actual data structures, but I'm
  * switching to another data store later anyways)
  */
-export class UnoptimizedInMemoryShuttleRepository implements ShuttleGetterSetterRepository {
+export class UnoptimizedInMemoryShuttleRepository
+  extends EventEmitter
+  implements ShuttleGetterSetterRepository {
+  public override on<T extends ShuttleRepositoryEventName>(
+    event: T,
+    listener: ShuttleRepositoryEventListener<T>,
+  ): this;
+  public override on(event: string | symbol, listener: (...args: any[]) => void): this {
+    return super.on(event, listener);
+  }
+
+  public override once<T extends ShuttleRepositoryEventName>(
+    event: T,
+    listener: ShuttleRepositoryEventListener<T>,
+  ): this;
+  public override once(event: string | symbol, listener: (...args: any[]) => void): this {
+    return super.once(event, listener);
+  }
+
+  public override off<T extends ShuttleRepositoryEventName>(
+    event: T,
+    listener: ShuttleRepositoryEventListener<T>,
+  ): this;
+  public override off(event: string | symbol, listener: (...args: any[]) => void): this {
+    return super.off(event, listener);
+  }
+
+  public override addListener<T extends ShuttleRepositoryEventName>(
+    event: T,
+    listener: ShuttleRepositoryEventListener<T>,
+  ): this;
+  public override addListener(event: string | symbol, listener: (...args: any[]) => void): this {
+    return super.addListener(event, listener);
+  }
+
+  public override removeListener<T extends ShuttleRepositoryEventName>(
+    event: T,
+    listener: ShuttleRepositoryEventListener<T>,
+  ): this;
+  public override removeListener(event: string | symbol, listener: (...args: any[]) => void): this {
+    return super.removeListener(event, listener);
+  }
+
+  public override emit<T extends ShuttleRepositoryEventName>(
+    event: T,
+    payload: ShuttleRepositoryEventPayloads[T],
+  ): boolean;
+  public override emit(event: string | symbol, ...args: any[]): boolean {
+    return super.emit(event, ...args);
+  }
+
   private stops: IStop[] = [];
   private routes: IRoute[] = [];
   private shuttles: IShuttle[] = [];
   private etas: IEta[] = [];
   private orderedStops: IOrderedStop[] = [];
 
-  private subscribers: ((eta: IEta) => void)[] = [];
-
   public async getStops(): Promise<IStop[]> {
     return this.stops;
   }
@@ -52,17 +107,6 @@ export class UnoptimizedInMemoryShuttleRepository implements ShuttleGetterSetter
     return this.etas.filter(eta => eta.stopId === stopId);
   }
 
-  public subscribeToEtaUpdates(listener: (eta: IEta) => void) {
-    this.subscribers.push(listener);
-  }
-
-  public unsubscribeFromEtaUpdates(listener: (eta: IEta) => void) {
-    const index = this.subscribers.findIndex((existingListener) => existingListener == listener);
-    if (index >= 0) {
-      this.subscribers.splice(index, 1);
-    }
-  }
-
   public async getEtaForShuttleAndStopId(shuttleId: string, stopId: string) {
     return this.findEntityByMatcher<IEta>((value) => value.stopId === stopId && value.shuttleId === shuttleId, this.etas);
   }
@@ -134,13 +178,7 @@ export class UnoptimizedInMemoryShuttleRepository implements ShuttleGetterSetter
     } else {
       this.etas.push(eta);
     }
-    this.publishEtaUpdateToSubscribers(eta);
-  }
-
-  private publishEtaUpdateToSubscribers(eta: IEta) {
-    this.subscribers.forEach(subscriber => {
-      subscriber(eta);
-    });
+    this.emit(ShuttleRepositoryEvent.ETA_UPDATED, eta);
   }
 
   private async removeEntityByMatcherIfExists<T>(callback: (value: T) => boolean, arrayToSearchIn: T[]) {
@@ -178,10 +216,14 @@ export class UnoptimizedInMemoryShuttleRepository implements ShuttleGetterSetter
   }
 
   public async removeEtaIfExists(shuttleId: string, stopId: string): Promise<IEta | null> {
-    return await this.removeEntityByMatcherIfExists((eta) => {
+    const removedEta = await this.removeEntityByMatcherIfExists((eta) => {
       return eta.stopId === stopId
         && eta.shuttleId === shuttleId
     }, this.etas);
+    if (removedEta) {
+      this.emit(ShuttleRepositoryEvent.ETA_REMOVED, removedEta);
+    }
+    return removedEta;
   }
 
   public async clearShuttleData(): Promise<void> {
@@ -189,7 +231,9 @@ export class UnoptimizedInMemoryShuttleRepository implements ShuttleGetterSetter
   }
 
   public async clearEtaData(): Promise<void> {
+    const removedEtas = [...this.etas];
     this.etas = [];
+    this.emit(ShuttleRepositoryEvent.ETA_DATA_CLEARED, removedEtas);
   }
 
   public async clearOrderedStopData(): Promise<void> {

src/repositories/shuttle/__tests__/UnoptimizedInMemoryShuttleRepositoryTests.test.ts

@@ -1,5 +1,6 @@
 import { beforeEach, describe, expect, jest, test } from "@jest/globals";
 import { UnoptimizedInMemoryShuttleRepository } from "../UnoptimizedInMemoryShuttleRepository";
+import { ShuttleRepositoryEvent } from "../ShuttleGetterRepository";
 import {
   generateMockEtas,
   generateMockOrderedStops,
@@ -186,49 +187,66 @@ describe("UnoptimizedInMemoryRepository", () => {
     });
   });
 
-  describe("subscribeToEtaChanges", () => {
+  describe("on/addListener", () => {
     test("notifies listeners if etas have been added or changed", async () => {
-      const mockCallback = jest.fn(); // Jest mock function to simulate a listener
-      repository.subscribeToEtaUpdates(mockCallback);
+      const mockListener = jest.fn();
+      repository.on(ShuttleRepositoryEvent.ETA_UPDATED, mockListener);
 
       const mockEtas = generateMockEtas();
       for (const eta of mockEtas) {
-        await repository.addOrUpdateEta(eta); // Trigger changes in ETAs
+        await repository.addOrUpdateEta(eta);
       }
 
-      expect(mockCallback).toHaveBeenCalledTimes(mockEtas.length);
-      expect(mockCallback).toHaveBeenCalledWith(mockEtas[0]); // First notification
-      expect(mockCallback).toHaveBeenCalledWith(mockEtas[mockEtas.length - 1]); // Last notification
+      expect(mockListener).toHaveBeenCalledTimes(mockEtas.length);
+      expect(mockListener).toHaveBeenCalledWith(mockEtas[0]); // First notification
+      expect(mockListener).toHaveBeenCalledWith(mockEtas[mockEtas.length - 1]); // Last notification
+    });
+
+    test("does not notify listener if removed", async () => {
+      const mockListener = jest.fn();
+      repository.on(ShuttleRepositoryEvent.ETA_UPDATED, mockListener);
+
+      const mockEtas = generateMockEtas();
+
+      repository.off(ShuttleRepositoryEvent.ETA_UPDATED, mockListener);
+      await repository.addOrUpdateEta(mockEtas[0]);
+      expect(mockListener).toHaveBeenCalledTimes(0);
     });
   });
 
-  describe("unsubscribeFromEtaChanges", () => {
+  describe("off/removeListener", () => {
     test("stops notifying listeners after etas have stopped changing", async () => {
-      const mockCallback = jest.fn(); // Jest mock function to simulate a listener
-      repository.subscribeToEtaUpdates(mockCallback);
+      const mockListener = jest.fn(); // Jest mock function to simulate a listener
+      repository.on(ShuttleRepositoryEvent.ETA_UPDATED, mockListener);
 
       const mockEtas = generateMockEtas();
       await repository.addOrUpdateEta(mockEtas[0]);
 
-      repository.unsubscribeFromEtaUpdates(mockCallback);
+      repository.off(ShuttleRepositoryEvent.ETA_UPDATED, mockListener);
 
       await repository.addOrUpdateEta(mockEtas[mockEtas.length - 1]);
 
-      expect(mockCallback).toHaveBeenCalledTimes(1);
-      expect(mockCallback).toHaveBeenCalledWith(mockEtas[0]); // First notification
-      expect(mockCallback).not.toHaveBeenCalledWith(mockEtas[mockEtas.length - 1]); // Last notification
+      expect(mockListener).toHaveBeenCalledTimes(1);
+      expect(mockListener).toHaveBeenCalledWith(mockEtas[0]); // First notification
+      expect(mockListener).not.toHaveBeenCalledWith(mockEtas[mockEtas.length - 1]); // Last notification
     });
 
-    test("does nothing if the listener doesn't exist", async () => {
-      const mockCallback = jest.fn();
-      repository.subscribeToEtaUpdates(mockCallback);
+    test("does not remove listener if wrong reference", async () => {
+      const mockListener = jest.fn();
+      repository.on(ShuttleRepositoryEvent.ETA_UPDATED, mockListener);
 
       const mockEtas = generateMockEtas();
 
-      repository.unsubscribeFromEtaUpdates(() => {});
+      repository.off(ShuttleRepositoryEvent.ETA_UPDATED, () => {});
+
       await repository.addOrUpdateEta(mockEtas[0]);
-      expect(mockCallback).toHaveBeenCalledTimes(1);
+      
+      expect(mockListener).toHaveBeenCalledTimes(1);
+      expect(mockListener).toHaveBeenCalledWith(mockEtas[0]);
     });
+  })
+
+  describe("ETA update events", () => {
   });
 
   describe("getOrderedStopByRouteAndStopId", () => {
@@ -569,6 +587,19 @@ describe("UnoptimizedInMemoryRepository", () => {
       const remainingEtas = await repository.getEtasForStopId(stopId);
       expect(remainingEtas).toHaveLength(mockEtas.length);
     });
+
+    test("emits an eta removed event when an eta is removed", async () => {
+      const mockEtas = generateMockEtas();
+      const etaToRemove = mockEtas[0];
+      const listener = jest.fn();
+      repository.on(ShuttleRepositoryEvent.ETA_REMOVED, listener);
+
+      await repository.addOrUpdateEta(etaToRemove);
+      await repository.removeEtaIfExists(etaToRemove.shuttleId, etaToRemove.stopId);
+
+      expect(listener).toHaveBeenCalledTimes(1);
+      expect(listener).toHaveBeenCalledWith(etaToRemove);
+    });
   });
 
   describe("clearShuttleData", () => {
@@ -597,6 +628,21 @@ describe("UnoptimizedInMemoryRepository", () => {
       const result = await repository.getEtasForShuttleId("shuttle1");
       expect(result).toEqual([]);
     });
+
+    test("emits an event with the cleared etas", async () => {
+      const mockEtas = generateMockEtas();
+      const listener = jest.fn();
+      repository.on(ShuttleRepositoryEvent.ETA_DATA_CLEARED, listener);
+
+      for (const eta of mockEtas) {
+        await repository.addOrUpdateEta(eta);
+      }
+
+      await repository.clearEtaData();
+
+      expect(listener).toHaveBeenCalledTimes(1);
+      expect(listener).toHaveBeenCalledWith(mockEtas);
+    });
   });
 
   describe("clearOrderedStopData", () => {