Merge branch 'main' into feat/notification-service

src/repositories/GetterRepository.ts

@@ -16,9 +16,26 @@ export interface GetterRepository {
 
   getEtasForShuttleId(shuttleId: string): Promise<IEta[]>;
   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;
+
   getOrderedStopByRouteAndStopId(routeId: string, stopId: string): Promise<IOrderedStop | null>;
 
   /**

src/repositories/UnoptimizedInMemoryRepository.ts

@@ -14,6 +14,8 @@ export class UnoptimizedInMemoryRepository implements GetterSetterRepository {
   private etas: IEta[] = [];
   private orderedStops: IOrderedStop[] = [];
 
+  private subscribers: ((eta: IEta) => void)[] = [];
+
   public async getSystems() {
     return this.systems;
   }
@@ -58,6 +60,17 @@ export class UnoptimizedInMemoryRepository implements GetterSetterRepository {
     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);
   }
@@ -138,6 +151,13 @@ export class UnoptimizedInMemoryRepository implements GetterSetterRepository {
     } else {
       this.etas.push(eta);
     }
+    this.publishEtaUpdateToSubscribers(eta);
+  }
+
+  private publishEtaUpdateToSubscribers(eta: IEta) {
+    this.subscribers.forEach(subscriber => {
+      subscriber(eta);
+    });
   }
 
   private async removeEntityByMatcherIfExists<T>(callback: (value: T) => boolean, arrayToSearchIn: T[]) {

test/repositories/UnoptimizedInMemoryRepositoryTests.test.ts

@@ -1,4 +1,4 @@
-import { beforeEach, describe, expect, test } from "@jest/globals";
+import { beforeEach, describe, expect, jest, test } from "@jest/globals";
 import { UnoptimizedInMemoryRepository } from "../../src/repositories/UnoptimizedInMemoryRepository";
 import {
   generateMockEtas,
@@ -225,6 +225,51 @@ describe("UnoptimizedInMemoryRepository", () => {
     });
   });
 
+  describe("subscribeToEtaChanges", () => {
+    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 mockEtas = generateMockEtas();
+      for (const eta of mockEtas) {
+        await repository.addOrUpdateEta(eta); // Trigger changes in ETAs
+      }
+
+      expect(mockCallback).toHaveBeenCalledTimes(mockEtas.length);
+      expect(mockCallback).toHaveBeenCalledWith(mockEtas[0]); // First notification
+      expect(mockCallback).toHaveBeenCalledWith(mockEtas[mockEtas.length - 1]); // Last notification
+    });
+  });
+
+  describe("unsubscribeFromEtaChanges", () => {
+    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 mockEtas = generateMockEtas();
+      await repository.addOrUpdateEta(mockEtas[0]);
+
+      repository.unsubscribeFromEtaUpdates(mockCallback);
+
+      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
+    });
+
+    test("does nothing if the listener doesn't exist", async () => {
+      const mockCallback = jest.fn();
+      repository.subscribeToEtaUpdates(mockCallback);
+
+      const mockEtas = generateMockEtas();
+
+      repository.unsubscribeFromEtaUpdates(() => {});
+      await repository.addOrUpdateEta(mockEtas[0]);
+      expect(mockCallback).toHaveBeenCalledTimes(1);
+    });
+  });
+
   describe("getOrderedStopByRouteAndStopId", () => {
     test("gets an ordered stop by route ID and stop ID", async () => {
       const mockOrderedStops = generateMockOrderedStops();