2 files changed, 193 insertions, 47 deletions
diff --git a/FrontEnd/src/app/services/DataHub2.ts b/FrontEnd/src/app/services/DataHub2.ts
new file mode 100644
index 00000000..88849da3
--- /dev/null
+++ b/FrontEnd/src/app/services/DataHub2.ts
@@ -0,0 +1,171 @@
+import { Observable } from "rxjs";
+
+export type DataStatus = "syncing" | "synced" | "offline";
+
+export type Subscriber<TData> = (data: TData) => void;
+
+export interface DataAndStatus<TData> {
+  data: TData | null;
+  status: DataStatus;
+}
+
+export class DataLine2<TData> {
+  constructor(
+    private config: {
+      saveData: (data: TData) => Promise<void>;
+      getSavedData: () => Promise<TData | null>;
+      // return null for offline
+      fetchData: (savedData: TData | null) => Promise<TData | null>;
+    }
+  ) {}
+
+  private _current: DataAndStatus<TData> | null = null;
+  private _observers: Subscriber<DataAndStatus<TData>>[] = [];
+
+  get currentData(): DataAndStatus<TData> | null {
+    return this._current;
+  }
+
+  get isDestroyable(): boolean {
+    const { _observers, currentData } = this;
+    return (
+      _observers.length === 0 &&
+      (currentData == null || currentData.status !== "syncing")
+    );
+  }
+
+  private next(data: DataAndStatus<TData>): void {
+    this._current = data;
+    this._observers.forEach((o) => o(data));
+  }
+
+  subscribe(subsriber: Subscriber<DataAndStatus<TData>>): void {
+    this.sync(); // TODO: Should I sync at this point or let the user sync explicitly.
+    this._observers.push(subsriber);
+    const { currentData } = this;
+    if (currentData != null) {
+      subsriber(currentData);
+    }
+  }
+
+  unsubscribe(subsriber: Subscriber<DataAndStatus<TData>>): void {
+    const index = this._observers.indexOf(subsriber);
+    if (index > -1) this._observers.splice(index, 1);
+  }
+
+  getObservalble(): Observable<DataAndStatus<TData>> {
+    return new Observable<DataAndStatus<TData>>((observer) => {
+      const f = (data: DataAndStatus<TData>): void => {
+        observer.next(data);
+      };
+      this.subscribe(f);
+
+      return () => {
+        this.unsubscribe(f);
+      };
+    });
+  }
+
+  sync(): void {
+    const { currentData } = this;
+    if (currentData != null && currentData.status === "syncing") return;
+    this.next({ data: currentData?.data ?? null, status: "syncing" });
+    void this.config.getSavedData().then((savedData) => {
+      if (currentData == null && savedData != null) {
+        this.next({ data: savedData, status: "syncing" });
+      }
+      return this.config.fetchData(savedData).then((data) => {
+        if (data == null) {
+          this.next({
+            data: savedData,
+            status: "offline",
+          });
+        } else {
+          return this.config.saveData(data).then(() => {
+            this.next({ data: data, status: "synced" });
+          });
+        }
+      });
+    });
+  }
+
+  save(data: TData): void {
+    const { currentData } = this;
+    if (currentData != null && currentData.status === "syncing") return;
+    this.next({ data: currentData?.data ?? null, status: "syncing" });
+    void this.config.saveData(data).then(() => {
+      this.next({ data: data, status: "synced" });
+    });
+  }
+
+  getSavedData(): Promise<TData | null> {
+    return this.config.getSavedData();
+  }
+}
+
+export class DataHub2<TKey, TData> {
+  private readonly subscriptionLineMap = new Map<string, DataLine2<TData>>();
+
+  private keyToString: (key: TKey) => string;
+
+  private cleanTimerId = 0;
+
+  // setup is called after creating line and if it returns a function as destroyer, then when the line is destroyed the destroyer will be called.
+  constructor(
+    private config: {
+      saveData: (key: TKey, data: TData) => Promise<void>;
+      getSavedData: (key: TKey) => Promise<TData | null>;
+      fetchData: (key: TKey, savedData: TData | null) => Promise<TData | null>;
+      keyToString?: (key: TKey) => string;
+    }
+  ) {
+    this.keyToString =
+      config.keyToString ??
+      ((value): string => {
+        if (typeof value === "string") return value;
+        else
+          throw new Error(
+            "Default keyToString function only pass string value."
+          );
+      });
+  }
+
+  private cleanLines(): void {
+    const toDelete: string[] = [];
+    for (const [key, line] of this.subscriptionLineMap.entries()) {
+      if (line.isDestroyable) {
+        toDelete.push(key);
+      }
+    }
+
+    if (toDelete.length === 0) return;
+
+    for (const key of toDelete) {
+      this.subscriptionLineMap.delete(key);
+    }
+
+    if (this.subscriptionLineMap.size === 0) {
+      window.clearInterval(this.cleanTimerId);
+      this.cleanTimerId = 0;
+    }
+  }
+
+  private createLine(key: TKey): DataLine2<TData> {
+    const keyString = this.keyToString(key);
+    const newLine: DataLine2<TData> = new DataLine2<TData>({
+      saveData: (data) => this.config.saveData(key, data),
+      getSavedData: () => this.config.getSavedData(key),
+      fetchData: (savedData) => this.config.fetchData(key, savedData),
+    });
+    this.subscriptionLineMap.set(keyString, newLine);
+    if (this.subscriptionLineMap.size === 1) {
+      this.cleanTimerId = window.setInterval(this.cleanLines.bind(this), 20000);
+    }
+    return newLine;
+  }
+
+  getLine(key: TKey): DataLine2<TData> {
+    const keyString = this.keyToString(key);
+    return this.subscriptionLineMap.get(keyString) ?? this.createLine(key);
+  }
+}
diff --git a/FrontEnd/src/app/services/user.ts b/FrontEnd/src/app/services/user.ts
index f08802a5..3407ad02 100644
--- a/FrontEnd/src/app/services/user.ts
+++ b/FrontEnd/src/app/services/user.ts
@@ -25,6 +25,7 @@ import {
 import { dataStorage, throwIfNotNetworkError } from "./common";
 import { DataHub } from "./DataHub";
 import { pushAlert } from "./alert";
+import { DataHub2 } from "./DataHub2";
 
 export type User = HttpUser;
 
@@ -251,69 +252,43 @@ export class UserNotExistError extends Error {}
 
 export class UserInfoService {
   saveUser(user: HttpUser): void {
-    const key = user.username;
-    void this._userHub.optionalInitLineWithSyncAction(key, async (line) => {
-      await this.doSaveUser(user);
-      line.next({ user, type: "synced" });
-    });
+    this.userHub.getLine(user.username).save(user);
   }
 
   saveUsers(users: HttpUser[]): void {
     return users.forEach((user) => this.saveUser(user));
   }
 
-  private _getCachedUser(username: string): Promise<User | null> {
-    return dataStorage.getItem<HttpUser | null>(`user.${username}`);
-  }
-
-  private doSaveUser(user: HttpUser): Promise<void> {
-    return dataStorage.setItem<HttpUser>(`user.${user.username}`, user).then();
+  private generateUserDataStorageKey(username: string): string {
+    return `user.${username}`;
   }
 
-  getCachedUser(username: string): Promise<User | null> {
-    return this._getCachedUser(username);
-  }
-
-  syncUser(username: string): Promise<void> {
-    return this._userHub.getLineOrCreate(username).sync();
-  }
-
-  private _userHub = new DataHub<
-    string,
-    | { user: User; type: "cache" | "synced" | "offline" }
-    | { user?: undefined; type: "notexist" | "offline" }
-  >({
-    sync: async (key, line) => {
-      if (line.value == undefined) {
-        const cache = await this._getCachedUser(key);
-        if (cache != null) {
-          line.next({ user: cache, type: "cache" });
-        }
-      }
-
+  readonly userHub = new DataHub2<string, HttpUser | "notexist">({
+    saveData: (username, data) => {
+      if (typeof data === "string") return Promise.resolve();
+      return dataStorage
+        .setItem<HttpUser>(this.generateUserDataStorageKey(username), data)
+        .then();
+    },
+    getSavedData: (username) => {
+      return dataStorage.getItem<HttpUser | null>(
+        this.generateUserDataStorageKey(username)
+      );
+    },
+    fetchData: async (username) => {
       try {
-        const res = await getHttpUserClient().get(key);
-        await this.doSaveUser(res);
-        line.next({ user: res, type: "synced" });
+        return await getHttpUserClient().get(username);
       } catch (e) {
         if (e instanceof HttpUserNotExistError) {
-          line.next({ type: "notexist" });
-        } else {
-          const cache = await this._getCachedUser(key);
-          line.next({ user: cache ?? undefined, type: "offline" });
-          throwIfNotNetworkError(e);
+          return "notexist";
+        } else if (e instanceof HttpNetworkError) {
+          return null;
         }
+        throw e;
       }
     },
   });
 
-  getUser$(username: string): Observable<User> {
-    return this._userHub.getObservable(username).pipe(
-      map((state) => state?.user),
-      filter((user): user is User => user != null)
-    );
-  }
-
   private _getCachedAvatar(username: string): Promise<BlobWithEtag | null> {
     return dataStorage.getItem<BlobWithEtag | null>(`user.${username}.avatar`);
   }