Merge sync status into subscription hub.

5 files changed, 203 insertions, 229 deletions

diff --git a/Timeline/ClientApp/src/app/data/SubscriptionHub.ts b/Timeline/ClientApp/src/app/data/SubscriptionHub.ts
index 7c24983b..e19c547c 100644
--- a/Timeline/ClientApp/src/app/data/SubscriptionHub.ts
+++ b/Timeline/ClientApp/src/app/data/SubscriptionHub.ts
@@ -6,11 +6,17 @@ export type Subscriber<TData> = (data: TData) => void;
 export interface ISubscriptionLine<TData> {

   readonly value: undefined | TData;

   next(value: TData): void;

+  readonly isSyncing: boolean;

+  beginSync(): void;

+  endSync(): void;

+  endSyncAndNext(value: TData): void;

 }

 

 export class SubscriptionLine<TData> implements ISubscriptionLine<TData> {

   private _current: TData | undefined = undefined;

 

+  private _syncing = false;

+

   private _observers: Subscriber<TData>[] = [];

 

   constructor(

@@ -38,6 +44,22 @@ export class SubscriptionLine<TData> implements ISubscriptionLine<TData> {
     this._observers.forEach((observer) => observer(value));

   }

 

+  get isSyncing(): boolean {

+    return this._syncing;

+  }

+

+  beginSync(): void {

+    if (!this._syncing) {

+      this._syncing = true;

+    }

+  }

+

+  endSync(): void {

+    if (this._syncing) {

+      this._syncing = false;

+    }

+  }

+

   get destroyable(): boolean {

     const customDestroyable = this.config?.destroyable;

 

@@ -46,6 +68,11 @@ export class SubscriptionLine<TData> implements ISubscriptionLine<TData> {
       (customDestroyable != null ? customDestroyable(this._current) : true)

     );

   }

+

+  endSyncAndNext(value: TData): void {

+    this.endSync();

+    this.next(value);

+  }

 }

 

 export class SubscriptionHub<TKey, TData> {

diff --git a/Timeline/ClientApp/src/app/data/SyncStatusHub.ts b/Timeline/ClientApp/src/app/data/SyncStatusHub.ts
deleted file mode 100644
index ed84f056..00000000
--- a/Timeline/ClientApp/src/app/data/SyncStatusHub.ts
+++ /dev/null
@@ -1,19 +0,0 @@
-export class SyncStatusHub {

-  private map = new Map<string, boolean>();

-

-  get(key: string): boolean {

-    return this.map.get(key) ?? false;

-  }

-

-  begin(key: string): void {

-    this.map.set(key, true);

-  }

-

-  end(key: string): void {

-    this.map.set(key, false);

-  }

-}

-

-export const syncStatusHub = new SyncStatusHub();

-

-export default syncStatusHub;

diff --git a/Timeline/ClientApp/src/app/data/common.ts b/Timeline/ClientApp/src/app/data/common.ts
index 786279f2..35934a29 100644
--- a/Timeline/ClientApp/src/app/data/common.ts
+++ b/Timeline/ClientApp/src/app/data/common.ts
@@ -1,5 +1,7 @@
 import localforage from 'localforage';

 

+import { HttpNetworkError } from '../http/common';

+

 export const dataStorage = localforage.createInstance({

   name: 'data',

   description: 'Database for offline data.',

@@ -11,3 +13,9 @@ export class ForbiddenError extends Error {
     super(message);

   }

 }

+

+export function throwIfNotNetworkError(e: unknown): void {

+  if (!(e instanceof HttpNetworkError)) {

+    throw e;

+  }

+}

diff --git a/Timeline/ClientApp/src/app/data/timeline.ts b/Timeline/ClientApp/src/app/data/timeline.ts
index 842d4bee..aacb5f29 100644
--- a/Timeline/ClientApp/src/app/data/timeline.ts
+++ b/Timeline/ClientApp/src/app/data/timeline.ts
@@ -6,9 +6,8 @@ import { uniqBy } from 'lodash';
 

 import { convertError } from '../utilities/rxjs';

 

-import { dataStorage } from './common';

+import { dataStorage, throwIfNotNetworkError } from './common';

 import { SubscriptionHub } from './SubscriptionHub';

-import syncStatusHub from './SyncStatusHub';

 

 import { UserAuthInfo, checkLogin, userService, userInfoService } from './user';

 

@@ -30,12 +29,7 @@ import {
   HttpTimelineNotExistError,

   HttpTimelineNameConflictError,

 } from '../http/timeline';

-import {

-  BlobWithEtag,

-  NotModified,

-  HttpNetworkError,

-  HttpForbiddenError,

-} from '../http/common';

+import { BlobWithEtag, NotModified, HttpForbiddenError } from '../http/common';

 import { HttpUser } from '../http/user';

 

 export type TimelineInfo = HttpTimelineInfo;

@@ -116,6 +110,15 @@ export class TimelineService {
     return dataStorage.getItem<TimelineData | null>(`timeline.${timelineName}`);

   }

 

+  private saveTimeline(

+    timelineName: string,

+    data: TimelineData

+  ): Promise<void> {

+    return dataStorage

+      .setItem<TimelineData>(`timeline.${timelineName}`, data)

+      .then();

+  }

+

   private convertHttpTimelineToData(timeline: HttpTimelineInfo): TimelineData {

     return {

       ...timeline,

@@ -125,9 +128,15 @@ export class TimelineService {
   }

 

   private async syncTimeline(timelineName: string): Promise<void> {

-    const syncStatusKey = `timeline.${timelineName}`;

-    if (syncStatusHub.get(syncStatusKey)) return;

-    syncStatusHub.begin(syncStatusKey);

+    const line = this._timelineHub.getLineOrCreateWithoutSetup(timelineName);

+    if (line.isSyncing) return;

+

+    if (line.value == undefined) {

+      const cache = await this.getCachedTimeline(timelineName);

+      if (cache != null) {

+        line.next({ type: 'cache', timeline: cache });

+      }

+    }

 

     try {

       const httpTimeline = await getHttpTimelineClient().getTimeline(

@@ -139,33 +148,19 @@ export class TimelineService {
       );

 

       const timeline = this.convertHttpTimelineToData(httpTimeline);

-      await dataStorage.setItem<TimelineData>(

-        `timeline.${timelineName}`,

-        timeline

-      );

-

-      syncStatusHub.end(syncStatusKey);

-      this._timelineHub

-        .getLine(timelineName)

-        ?.next({ type: 'synced', timeline });

+      await this.saveTimeline(timelineName, timeline);

+      line.endSyncAndNext({ type: 'synced', timeline });

     } catch (e) {

-      syncStatusHub.end(syncStatusKey);

       if (e instanceof HttpTimelineNotExistError) {

-        this._timelineHub

-          .getLine(timelineName)

-          ?.next({ type: 'synced', timeline: null });

+        line.endSyncAndNext({ type: 'synced', timeline: null });

       } else {

         const cache = await this.getCachedTimeline(timelineName);

-        if (cache == null)

-          this._timelineHub

-            .getLine(timelineName)

-            ?.next({ type: 'offline', timeline: null });

-        else

-          this._timelineHub

-            .getLine(timelineName)

-            ?.next({ type: 'offline', timeline: cache });

-

-        if (!(e instanceof HttpNetworkError)) throw e;

+        if (cache == null) {

+          line.endSyncAndNext({ type: 'offline', timeline: null });

+        } else {

+          line.endSyncAndNext({ type: 'offline', timeline: cache });

+        }

+        throwIfNotNetworkError(e);

       }

     }

   }

@@ -181,13 +176,8 @@ export class TimelineService {
         timeline: TimelineData | null;

       }

   >({

-    setup: (key, line) => {

-      void this.getCachedTimeline(key).then((timeline) => {

-        if (timeline != null) {

-          line.next({ type: 'cache', timeline });

-        }

-        return this.syncTimeline(key);

-      });

+    setup: (key) => {

+      void this.syncTimeline(key);

     },

   });

 

@@ -293,22 +283,34 @@ export class TimelineService {
     return posts.map((post) => this.convertHttpPostToData(post));

   }

 

-  private async getCachedPosts(

+  private getCachedPosts(

     timelineName: string

-  ): Promise<TimelinePostData[]> {

-    const posts = await dataStorage.getItem<TimelinePostData[] | null>(

+  ): Promise<TimelinePostData[] | null> {

+    return dataStorage.getItem<TimelinePostData[] | null>(

       `timeline.${timelineName}.posts`

     );

-    if (posts == null) return [];

-    return posts;

   }

 

-  private async syncPosts(timelineName: string): Promise<void> {

-    const syncStatusKey = `timeline.posts.${timelineName}`;

+  private savePosts(

+    timelineName: string,

+    data: TimelinePostData[]

+  ): Promise<void> {

+    return dataStorage

+      .setItem<TimelinePostData[]>(`timeline.${timelineName}.posts`, data)

+      .then();

+  }

 

-    const dataKey = `timeline.${timelineName}.posts`;

-    if (syncStatusHub.get(syncStatusKey)) return;

-    syncStatusHub.begin(syncStatusKey);

+  private async syncPosts(timelineName: string): Promise<void> {

+    const line = this._postsHub.getLineOrCreateWithoutSetup(timelineName);

+    if (line.isSyncing) return;

+    line.beginSync();

+

+    if (line.value == null) {

+      const cache = await this.getCachedPosts(timelineName);

+      if (cache != null) {

+        line.next({ type: 'cache', posts: cache });

+      }

+    }

 

     try {

       const httpPosts = await getHttpTimelineClient().listPost(

@@ -322,31 +324,21 @@ export class TimelineService {
       ).forEach((user) => void userInfoService.saveUser(user));

 

       const posts = this.convertHttpPostToDataList(httpPosts);

-      await dataStorage.setItem<TimelinePostData[]>(dataKey, posts);

-

-      syncStatusHub.end(syncStatusKey);

-      this._postsHub.getLine(timelineName)?.next({ type: 'synced', posts });

+      await this.savePosts(timelineName, posts);

+      line.endSyncAndNext({ type: 'synced', posts });

     } catch (e) {

-      syncStatusHub.end(syncStatusKey);

       if (e instanceof HttpTimelineNotExistError) {

-        this._postsHub

-          .getLine(timelineName)

-          ?.next({ type: 'notexist', posts: [] });

+        line.endSyncAndNext({ type: 'notexist', posts: [] });

       } else if (e instanceof HttpForbiddenError) {

-        this._postsHub

-          .getLine(timelineName)

-          ?.next({ type: 'forbid', posts: [] });

+        line.endSyncAndNext({ type: 'forbid', posts: [] });

       } else {

         const cache = await this.getCachedPosts(timelineName);

-        if (cache == null)

-          this._postsHub

-            .getLine(timelineName)

-            ?.next({ type: 'offline', posts: [] });

-        else

-          this._postsHub

-            .getLine(timelineName)

-            ?.next({ type: 'offline', posts: cache });

-        if (!(e instanceof HttpNetworkError)) throw e;

+        if (cache == null) {

+          line.endSyncAndNext({ type: 'offline', posts: [] });

+        } else {

+          line.endSyncAndNext({ type: 'offline', posts: cache });

+        }

+        throwIfNotNetworkError(e);

       }

     }

   }

@@ -358,13 +350,8 @@ export class TimelineService {
       posts: TimelinePostData[];

     }

   >({

-    setup: (key, line) => {

-      void this.getCachedPosts(key).then((posts) => {

-        if (posts != null) {

-          line.next({ type: 'cache', posts });

-        }

-        return this.syncPosts(key);

-      });

+    setup: (key) => {

+      void this.syncPosts(key);

     },

   });

 

@@ -418,76 +405,73 @@ export class TimelineService {
     );

   }

 

-  private getCachedPostData(

-    timelineName: string,

-    postId: number

-  ): Promise<Blob | null> {

-    return dataStorage

-      .getItem<BlobWithEtag | null>(

-        `timeline.${timelineName}.post.${postId}.data`

-      )

-      .then((data) => data?.data ?? null);

+  private getCachedPostData(key: {

+    timelineName: string;

+    postId: number;

+  }): Promise<BlobWithEtag | null> {

+    return dataStorage.getItem<BlobWithEtag | null>(

+      `timeline.${key.timelineName}.post.${key.postId}.data`

+    );

   }

 

-  private async syncPostData(

-    timelineName: string,

-    postId: number

+  private savePostData(

+    key: {

+      timelineName: string;

+      postId: number;

+    },

+    data: BlobWithEtag

   ): Promise<void> {

-    const syncStatusKey = `user.timeline.${timelineName}.post.data.${postId}`;

-    if (syncStatusHub.get(syncStatusKey)) return;

-    syncStatusHub.begin(syncStatusKey);

+    return dataStorage

+      .setItem<BlobWithEtag>(

+        `timeline.${key.timelineName}.post.${key.postId}.data`,

+        data

+      )

+      .then();

+  }

 

-    const dataKey = `timeline.${timelineName}.post.${postId}.data`;

+  private async syncPostData(key: {

+    timelineName: string;

+    postId: number;

+  }): Promise<void> {

+    const line = this._postDataHub.getLineOrCreateWithoutSetup(key);

+    if (line.isSyncing) return;

+    line.beginSync();

+

+    const cache = await this.getCachedPostData(key);

+    if (line.value == null) {

+      if (cache != null) {

+        line.next({ type: 'cache', data: cache.data });

+      }

+    }

 

-    const cache = await dataStorage.getItem<BlobWithEtag | null>(dataKey);

     if (cache == null) {

       try {

-        const data = await getHttpTimelineClient().getPostData(

-          timelineName,

-          postId

+        const res = await getHttpTimelineClient().getPostData(

+          key.timelineName,

+          key.postId

         );

-        await dataStorage.setItem<BlobWithEtag>(dataKey, data);

-        syncStatusHub.end(syncStatusKey);

-        this._postDataHub

-          .getLine({ timelineName, postId })

-          ?.next({ data: data.data, type: 'synced' });

+        await this.savePostData(key, res);

+        line.endSyncAndNext({ data: res.data, type: 'synced' });

       } catch (e) {

-        syncStatusHub.end(syncStatusKey);

-        this._postDataHub

-          .getLine({ timelineName, postId })

-          ?.next({ type: 'offline' });

-        if (!(e instanceof HttpNetworkError)) {

-          throw e;

-        }

+        line.endSyncAndNext({ type: 'offline' });

+        throwIfNotNetworkError(e);

       }

     } else {

       try {

         const res = await getHttpTimelineClient().getPostData(

-          timelineName,

-          postId,

+          key.timelineName,

+          key.postId,

           cache.etag

         );

         if (res instanceof NotModified) {

-          syncStatusHub.end(syncStatusKey);

-          this._postDataHub

-            .getLine({ timelineName, postId })

-            ?.next({ data: cache.data, type: 'synced' });

+          line.endSyncAndNext({ data: cache.data, type: 'synced' });

         } else {

-          const avatar = res;

-          await dataStorage.setItem<BlobWithEtag>(dataKey, avatar);

-          syncStatusHub.end(syncStatusKey);

-          this._postDataHub

-            .getLine({ timelineName, postId })

-            ?.next({ data: avatar.data, type: 'synced' });

+          await this.savePostData(key, res);

+          line.endSyncAndNext({ data: res.data, type: 'synced' });

         }

       } catch (e) {

-        syncStatusHub.end(syncStatusKey);

-        this._postDataHub

-          .getLine({ timelineName, postId })

-          ?.next({ data: cache.data, type: 'offline' });

-        if (!(e instanceof HttpNetworkError)) {

-          throw e;

-        }

+        line.endSyncAndNext({ data: cache.data, type: 'offline' });

+        throwIfNotNetworkError(e);

       }

     }

   }

@@ -498,13 +482,8 @@ export class TimelineService {
     | { data?: undefined; type: 'notexist' | 'offline' }

   >({

     keyToString: (key) => `${key.timelineName}.${key.postId}`,

-    setup: (key, line) => {

-      void this.getCachedPostData(key.timelineName, key.postId).then((data) => {

-        if (data != null) {

-          line.next({ data: data, type: 'cache' });

-        }

-        return this.syncPostData(key.timelineName, key.postId);

-      });

+    setup: (key) => {

+      void this.syncPostData(key);

     },

   });

 

diff --git a/Timeline/ClientApp/src/app/data/user.ts b/Timeline/ClientApp/src/app/data/user.ts
index 5b96e2b6..d19a6323 100644
--- a/Timeline/ClientApp/src/app/data/user.ts
+++ b/Timeline/ClientApp/src/app/data/user.ts
@@ -6,8 +6,7 @@ import { UiLogicError } from '../common';
 import { convertError } from '../utilities/rxjs';

 import { pushAlert } from '../common/alert-service';

 

-import { dataStorage } from './common';

-import { syncStatusHub } from './SyncStatusHub';

+import { dataStorage, throwIfNotNetworkError } from './common';

 import { SubscriptionHub } from './SubscriptionHub';

 

 import { HttpNetworkError, BlobWithEtag, NotModified } from '../http/common';

@@ -229,47 +228,45 @@ export class UserNotExistError extends Error {}
 

 export class UserInfoService {

   async saveUser(user: HttpUser): Promise<void> {

-    const syncStatusKey = `user.${user.username}`;

-    if (syncStatusHub.get(syncStatusKey)) return;

-    syncStatusHub.begin(syncStatusKey);

+    const key = user.username;

+    const line = this._userHub.getLineOrCreateWithoutSetup(key);

+    if (line.isSyncing) return;

+    line.beginSync();

     await this.doSaveUser(user);

-    syncStatusHub.end(syncStatusKey);

-    this._userHub.getLine(user.username)?.next({ user, type: 'synced' });

+    line.endSyncAndNext({ user, type: 'synced' });

   }

 

   private getCachedUser(username: string): Promise<User | null> {

     return dataStorage.getItem<HttpUser | null>(`user.${username}`);

   }

 

-  private async doSaveUser(user: HttpUser): Promise<void> {

-    await dataStorage.setItem<HttpUser>(`user.${user.username}`, user);

+  private doSaveUser(user: HttpUser): Promise<void> {

+    return dataStorage.setItem<HttpUser>(`user.${user.username}`, user).then();

   }

 

   private async syncUser(username: string): Promise<void> {

-    const syncStatusKey = `user.${username}`;

-    if (syncStatusHub.get(syncStatusKey)) return;

-    syncStatusHub.begin(syncStatusKey);

+    const line = this._userHub.getLineOrCreateWithoutSetup(username);

+    if (line.isSyncing) return;

+    line.beginSync();

+

+    if (line.value == undefined) {

+      const cache = await this.getCachedUser(username);

+      if (cache != null) {

+        line.next({ user: cache, type: 'cache' });

+      }

+    }

 

     try {

       const res = await getHttpUserClient().get(username);

       await this.doSaveUser(res);

-      syncStatusHub.end(syncStatusKey);

-      this._userHub.getLine(username)?.next({ user: res, type: 'synced' });

+      line.endSyncAndNext({ user: res, type: 'synced' });

     } catch (e) {

       if (e instanceof HttpUserNotExistError) {

-        syncStatusHub.end(syncStatusKey);

-        this._userHub.getLine(username)?.next({ type: 'notexist' });

+        line.endSyncAndNext({ type: 'notexist' });

       } else {

-        syncStatusHub.end(syncStatusKey);

-        const line = this._userHub.getLine(username);

-        if (line != null) {

-          const cache = await this.getCachedUser(username);

-          if (cache == null) line.next({ type: 'offline' });

-          else line.next({ user: cache, type: 'offline' });

-        }

-        if (!(e instanceof HttpNetworkError)) {

-          throw e;

-        }

+        const cache = await this.getCachedUser(username);

+        line.endSyncAndNext({ user: cache ?? undefined, type: 'offline' });

+        throwIfNotNetworkError(e);

       }

     }

   }

@@ -279,13 +276,8 @@ export class UserInfoService {
     | { user: User; type: 'cache' | 'synced' | 'offline' }

     | { user?: undefined; type: 'notexist' | 'offline' }

   >({

-    setup: (key, line) => {

-      void this.getCachedUser(key).then((cache) => {

-        if (cache != null) {

-          line.next({ user: cache, type: 'cache' });

-        }

-        return this.syncUser(key);

-      });

+    setup: (key) => {

+      void this.syncUser(key);

     },

   });

 

@@ -296,58 +288,50 @@ export class UserInfoService {
     );

   }

 

-  private getCachedAvatar(username: string): Promise<Blob | null> {

+  private getCachedAvatar(username: string): Promise<BlobWithEtag | null> {

+    return dataStorage.getItem<BlobWithEtag | null>(`user.${username}.avatar`);

+  }

+

+  private saveAvatar(username: string, data: BlobWithEtag): Promise<void> {

     return dataStorage

-      .getItem<BlobWithEtag | null>(`user.${username}.avatar`)

-      .then((data) => data?.data ?? null);

+      .setItem<BlobWithEtag>(`user.${username}.avatar`, data)

+      .then();

   }

 

   private async syncAvatar(username: string): Promise<void> {

-    const syncStatusKey = `user.avatar.${username}`;

-    if (syncStatusHub.get(syncStatusKey)) return;

-    syncStatusHub.begin(syncStatusKey);

+    const line = this._avatarHub.getLineOrCreateWithoutSetup(username);

+    if (line.isSyncing) return;

+    line.beginSync();

+

+    const cache = await this.getCachedAvatar(username);

+    if (line.value == null) {

+      if (cache != null) {

+        line.next({ data: cache.data, type: 'cache' });

+      }

+    }

 

-    const dataKey = `user.${username}.avatar`;

-    const cache = await dataStorage.getItem<BlobWithEtag | null>(dataKey);

     if (cache == null) {

       try {

         const avatar = await getHttpUserClient().getAvatar(username);

-        await dataStorage.setItem<BlobWithEtag>(dataKey, avatar);

-        syncStatusHub.end(syncStatusKey);

-        this._avatarHub

-          .getLine(username)

-          ?.next({ data: avatar.data, type: 'synced' });

+        await this.saveAvatar(username, avatar);

+        line.endSyncAndNext({ data: avatar.data, type: 'synced' });

       } catch (e) {

-        syncStatusHub.end(syncStatusKey);

-        this._avatarHub.getLine(username)?.next({ type: 'offline' });

-        if (!(e instanceof HttpNetworkError)) {

-          throw e;

-        }

+        line.endSyncAndNext({ type: 'offline' });

+        throwIfNotNetworkError(e);

       }

     } else {

       try {

         const res = await getHttpUserClient().getAvatar(username, cache.etag);

         if (res instanceof NotModified) {

-          syncStatusHub.end(syncStatusKey);

-          this._avatarHub

-            .getLine(username)

-            ?.next({ data: cache.data, type: 'synced' });

+          line.endSyncAndNext({ data: cache.data, type: 'synced' });

         } else {

           const avatar = res;

-          await dataStorage.setItem<BlobWithEtag>(dataKey, avatar);

-          syncStatusHub.end(syncStatusKey);

-          this._avatarHub

-            .getLine(username)

-            ?.next({ data: avatar.data, type: 'synced' });

+          await this.saveAvatar(username, avatar);

+          line.endSyncAndNext({ data: avatar.data, type: 'synced' });

         }

       } catch (e) {

-        syncStatusHub.end(syncStatusKey);

-        this._avatarHub

-          .getLine(username)

-          ?.next({ data: cache.data, type: 'offline' });

-        if (!(e instanceof HttpNetworkError)) {

-          throw e;

-        }

+        line.endSyncAndNext({ data: cache.data, type: 'offline' });

+        throwIfNotNetworkError(e);

       }

     }

   }

@@ -357,13 +341,8 @@ export class UserInfoService {
     | { data: Blob; type: 'cache' | 'synced' | 'offline' }

     | { data?: undefined; type: 'notexist' | 'offline' }

   >({

-    setup: (key, line) => {

-      void this.getCachedAvatar(key).then((avatar) => {

-        if (avatar != null) {

-          line.next({ data: avatar, type: 'cache' });

-        }

-        return this.syncAvatar(key);

-      });

+    setup: (key) => {

+      void this.syncAvatar(key);

     },

   });