fix: tests (#2516)

frontend/appflowy_tauri/src/appflowy_app/components/tests/DatabaseTestHelper.ts

@@ -26,13 +26,13 @@ import { FieldInfo } from '../../stores/effects/database/field/field_controller'
 import { TypeOptionController } from '../../stores/effects/database/field/type_option/type_option_controller';
 import { makeSingleSelectTypeOptionContext } from '../../stores/effects/database/field/type_option/type_option_context';
 import { SelectOptionBackendService } from '../../stores/effects/database/cell/select_option_bd_svc';
+import { Log } from '$app/utils/log';
 
 // Create a database view for specific layout type
 // Do not use it production code. Just for testing
 export async function createTestDatabaseView(layout: ViewLayoutPB): Promise<ViewPB> {
   const workspaceSetting: WorkspaceSettingPB = await FolderEventReadCurrentWorkspace().then((result) => result.unwrap());
-  const app = workspaceSetting.workspace.views[0];
-  const appService = new AppBackendService(app.id);
+  const appService = new AppBackendService(workspaceSetting.workspace.id);
   return await appService.createView({ name: 'New Grid', layoutType: layout });
 }
 

frontend/appflowy_tauri/src/appflowy_app/stores/effects/database/cell/cell_controller.ts

@@ -30,6 +30,7 @@ export class CellController<T, D> {
     this.cellDataNotifier = new CellDataNotifier(cellCache.get<T>(this.cacheKey));
     this.cellObserver = new CellObserver(cellIdentifier.rowId, cellIdentifier.fieldId);
     this.fieldNotifier = new DatabaseFieldObserver(cellIdentifier.fieldId);
+
     void this.cellObserver.subscribe({
       /// 1.Listen on user edit event and load the new cell data if needed.
       /// For example:
@@ -37,7 +38,11 @@ export class CellController<T, D> {
       ///  cell display: $12
       onCellChanged: async () => {
         this.cellCache.remove(this.cacheKey);
-        await this._loadCellData();
+        try {
+          await this._loadCellData();
+        } catch (e) {
+          Log.error(e);
+        }
       },
     });
 
@@ -57,7 +62,7 @@ export class CellController<T, D> {
 
   subscribeChanged = (callbacks: Callbacks<T>) => {
     this.subscribeCallbacks = callbacks;
-    this.cellDataNotifier.observer.subscribe((cellData) => {
+    this.cellDataNotifier.observer?.subscribe((cellData) => {
       if (cellData !== null) {
         callbacks.onCellChanged(Some(cellData));
       }
@@ -108,7 +113,6 @@ export class CellController<T, D> {
   };
 
   dispose = async () => {
-    this.cellDataNotifier.unsubscribe();
     await this.cellObserver.unsubscribe();
     await this.fieldNotifier.unsubscribe();
   };

frontend/appflowy_tauri/src/appflowy_app/stores/effects/database/cell/cell_observer.ts

@@ -2,21 +2,20 @@ import { Ok, Result } from 'ts-results';
 import { ChangeNotifier } from '$app/utils/change_notifier';
 import { DatabaseNotificationObserver } from '../notifications/observer';
 import { DatabaseNotification, FlowyError } from '@/services/backend';
+import { Subscription } from 'rxjs';
 
-type UpdateCellNotifiedValue = Result<void, FlowyError>;
-
-export type CellChangedCallback = (value: UpdateCellNotifiedValue) => void;
+export type CellChangedCallback = (value: Result<void, FlowyError>) => void;
 
 export class CellObserver {
-  private notifier?: ChangeNotifier<UpdateCellNotifiedValue>;
+  private notifier?: ChangeNotifier<Result<void, FlowyError>>;
   private listener?: DatabaseNotificationObserver;
+  private subscription?: Subscription;
 
   constructor(public readonly rowId: string, public readonly fieldId: string) {}
 
   subscribe = async (callbacks: { onCellChanged: CellChangedCallback }) => {
     this.notifier = new ChangeNotifier();
-    this.notifier?.observer.subscribe(callbacks.onCellChanged);
-
+    this.subscription = this.notifier?.observer?.subscribe(callbacks.onCellChanged);
     this.listener = new DatabaseNotificationObserver({
       // The rowId combine with fieldId can identifier the cell.
       // This format rowId:fieldId is also defined in the backend,
@@ -40,7 +39,7 @@ export class CellObserver {
   };
 
   unsubscribe = async () => {
-    this.notifier?.unsubscribe();
+    this.subscription?.unsubscribe();
     await this.listener?.stop();
   };
 }

frontend/appflowy_tauri/src/appflowy_app/stores/effects/database/field/field_controller.ts

@@ -1,8 +1,8 @@
-import { Log } from '$app/utils/log';
-import { DatabaseBackendService } from '../database_bd_svc';
-import { DatabaseFieldChangesetObserver } from './field_observer';
-import { FieldIdPB, FieldPB, IndexFieldPB } from '@/services/backend';
-import { ChangeNotifier } from '$app/utils/change_notifier';
+import { Log } from "$app/utils/log";
+import { DatabaseBackendService } from "../database_bd_svc";
+import { DatabaseFieldChangesetObserver } from "./field_observer";
+import { FieldIdPB, FieldPB, IndexFieldPB } from "@/services/backend";
+import { ChangeNotifier } from "$app/utils/change_notifier";
 
 export class FieldController {
   private backendService: DatabaseBackendService;
@@ -37,7 +37,7 @@ export class FieldController {
   };
 
   subscribe = (callbacks: { onNumOfFieldsChanged?: (fieldInfos: readonly FieldInfo[]) => void }) => {
-    this.numOfFieldsNotifier.observer.subscribe((fieldInfos) => {
+    this.numOfFieldsNotifier.observer?.subscribe((fieldInfos) => {
       callbacks.onNumOfFieldsChanged?.(fieldInfos);
     });
   };
@@ -53,7 +53,7 @@ export class FieldController {
         } else {
           Log.error(result.val);
         }
-      },
+      }
     });
   };
 
@@ -122,5 +122,6 @@ class NumOfFieldsNotifier extends ChangeNotifier<FieldInfo[]> {
 }
 
 export class FieldInfo {
-  constructor(public readonly field: FieldPB) {}
+  constructor(public readonly field: FieldPB) {
+  }
 }

frontend/appflowy_tauri/src/appflowy_app/stores/effects/database/field/field_observer.ts

@@ -1,7 +1,7 @@
-import { Ok, Result } from 'ts-results';
-import { DatabaseNotification, DatabaseFieldChangesetPB, FlowyError, FieldPB } from '@/services/backend';
-import { ChangeNotifier } from '$app/utils/change_notifier';
-import { DatabaseNotificationObserver } from '../notifications/observer';
+import { Ok, Result } from "ts-results";
+import { DatabaseNotification, DatabaseFieldChangesetPB, FlowyError, FieldPB } from "@/services/backend";
+import { ChangeNotifier } from "$app/utils/change_notifier";
+import { DatabaseNotificationObserver } from "../notifications/observer";
 
 export type FieldChangesetSubscribeCallback = (value: Result<DatabaseFieldChangesetPB, FlowyError>) => void;
 
@@ -9,11 +9,12 @@ export class DatabaseFieldChangesetObserver {
   private notifier?: ChangeNotifier<Result<DatabaseFieldChangesetPB, FlowyError>>;
   private listener?: DatabaseNotificationObserver;
 
-  constructor(public readonly viewId: string) {}
+  constructor(public readonly viewId: string) {
+  }
 
   subscribe = async (callbacks: { onFieldsChanged: FieldChangesetSubscribeCallback }) => {
     this.notifier = new ChangeNotifier();
-    this.notifier?.observer.subscribe(callbacks.onFieldsChanged);
+    this.notifier?.observer?.subscribe(callbacks.onFieldsChanged);
 
     this.listener = new DatabaseNotificationObserver({
       id: this.viewId,
@@ -29,7 +30,7 @@ export class DatabaseFieldChangesetObserver {
           default:
             break;
         }
-      },
+      }
     });
     await this.listener.start();
   };
@@ -46,11 +47,12 @@ export class DatabaseFieldObserver {
   private _notifier?: ChangeNotifier<Result<FieldPB, FlowyError>>;
   private _listener?: DatabaseNotificationObserver;
 
-  constructor(public readonly fieldId: string) {}
+  constructor(public readonly fieldId: string) {
+  }
 
   subscribe = async (callbacks: { onFieldChanged: FieldSubscribeCallback }) => {
     this._notifier = new ChangeNotifier();
-    this._notifier?.observer.subscribe(callbacks.onFieldChanged);
+    this._notifier?.observer?.subscribe(callbacks.onFieldChanged);
 
     this._listener = new DatabaseNotificationObserver({
       id: this.fieldId,
@@ -66,7 +68,7 @@ export class DatabaseFieldObserver {
           default:
             break;
         }
-      },
+      }
     });
     await this._listener.start();
   };

frontend/appflowy_tauri/src/appflowy_app/stores/effects/database/group/group_controller.ts

@@ -1,9 +1,9 @@
-import { DatabaseNotification, FlowyError, GroupPB, GroupRowsNotificationPB, RowPB } from '@/services/backend';
-import { ChangeNotifier } from '$app/utils/change_notifier';
-import { None, Ok, Option, Result, Some } from 'ts-results';
-import { DatabaseNotificationObserver } from '../notifications/observer';
-import { Log } from '$app/utils/log';
-import { DatabaseBackendService } from '../database_bd_svc';
+import { DatabaseNotification, FlowyError, GroupPB, GroupRowsNotificationPB, RowPB } from "@/services/backend";
+import { ChangeNotifier } from "$app/utils/change_notifier";
+import { None, Ok, Option, Result, Some } from "ts-results";
+import { DatabaseNotificationObserver } from "../notifications/observer";
+import { Log } from "$app/utils/log";
+import { DatabaseBackendService } from "../database_bd_svc";
 
 export type GroupDataCallbacks = {
   onRemoveRow: (groupId: string, rowId: string) => void;
@@ -83,7 +83,7 @@ export class DatabaseGroupController {
         } else {
           Log.error(result.val);
         }
-      },
+      }
     });
   };
 
@@ -111,11 +111,12 @@ class GroupDataObserver {
   private notifier?: ChangeNotifier<Result<GroupRowsNotificationPB, FlowyError>>;
   private listener?: DatabaseNotificationObserver;
 
-  constructor(public readonly groupId: string) {}
+  constructor(public readonly groupId: string) {
+  }
 
   subscribe = async (callbacks: { onRowsChanged: GroupRowsSubscribeCallback }) => {
     this.notifier = new ChangeNotifier();
-    this.notifier?.observer.subscribe(callbacks.onRowsChanged);
+    this.notifier?.observer?.subscribe(callbacks.onRowsChanged);
 
     this.listener = new DatabaseNotificationObserver({
       id: this.groupId,
@@ -131,7 +132,7 @@ class GroupDataObserver {
           default:
             break;
         }
-      },
+      }
     });
     await this.listener.start();
   };

frontend/appflowy_tauri/src/appflowy_app/stores/effects/database/group/group_observer.ts

@@ -1,7 +1,7 @@
-import { ChangeNotifier } from '$app/utils/change_notifier';
-import { Ok, Result } from 'ts-results';
-import { DatabaseNotification, FlowyError, GroupChangesetPB, GroupPB } from '@/services/backend';
-import { DatabaseNotificationObserver } from '../notifications/observer';
+import { ChangeNotifier } from "$app/utils/change_notifier";
+import { Ok, Result } from "ts-results";
+import { DatabaseNotification, FlowyError, GroupChangesetPB, GroupPB } from "@/services/backend";
+import { DatabaseNotificationObserver } from "../notifications/observer";
 
 export type GroupByFieldCallback = (value: Result<GroupPB[], FlowyError>) => void;
 export type GroupChangesetSubscribeCallback = (value: Result<GroupChangesetPB, FlowyError>) => void;
@@ -11,17 +11,18 @@ export class DatabaseGroupObserver {
   private groupChangesetNotifier?: ChangeNotifier<Result<GroupChangesetPB, FlowyError>>;
   private listener?: DatabaseNotificationObserver;
 
-  constructor(public readonly viewId: string) {}
+  constructor(public readonly viewId: string) {
+  }
 
   subscribe = async (callbacks: {
     onGroupBy: GroupByFieldCallback;
     onGroupChangeset: GroupChangesetSubscribeCallback;
   }) => {
     this.groupByNotifier = new ChangeNotifier();
-    this.groupByNotifier?.observer.subscribe(callbacks.onGroupBy);
+    this.groupByNotifier?.observer?.subscribe(callbacks.onGroupBy);
 
     this.groupChangesetNotifier = new ChangeNotifier();
-    this.groupChangesetNotifier?.observer.subscribe(callbacks.onGroupChangeset);
+    this.groupChangesetNotifier?.observer?.subscribe(callbacks.onGroupChangeset);
 
     this.listener = new DatabaseNotificationObserver({
       id: this.viewId,
@@ -44,7 +45,7 @@ export class DatabaseGroupObserver {
           default:
             break;
         }
-      },
+      }
     });
 
     await this.listener.start();

frontend/appflowy_tauri/src/appflowy_app/stores/effects/database/row/row_cache.ts

@@ -6,15 +6,15 @@ import {
   OptionalRowPB,
   RowsChangesetPB,
   RowsVisibilityChangesetPB,
-  ReorderSingleRowPB,
-} from '@/services/backend';
-import { ChangeNotifier } from '$app/utils/change_notifier';
-import { FieldInfo } from '../field/field_controller';
-import { CellCache, CellCacheKey } from '../cell/cell_cache';
-import { CellIdentifier } from '../cell/cell_bd_svc';
-import { DatabaseEventGetRow } from '@/services/backend/events/flowy-database2';
-import { None, Option, Some } from 'ts-results';
-import { Log } from '$app/utils/log';
+  ReorderSingleRowPB
+} from "@/services/backend";
+import { ChangeNotifier } from "$app/utils/change_notifier";
+import { FieldInfo } from "../field/field_controller";
+import { CellCache, CellCacheKey } from "../cell/cell_cache";
+import { CellIdentifier } from "../cell/cell_bd_svc";
+import { DatabaseEventGetRow } from "@/services/backend/events/flowy-database2";
+import { None, Option, Some } from "ts-results";
+import { Log } from "$app/utils/log";
 
 export type CellByFieldId = Map<string, CellIdentifier>;
 
@@ -56,7 +56,7 @@ export class RowCache {
   subscribe = (callbacks: {
     onRowsChanged: (reason: RowChangedReason, cellMap?: Map<string, CellIdentifier>) => void;
   }) => {
-    return this.notifier.observer.subscribe((change) => {
+    return this.notifier.observer?.subscribe((change) => {
       if (change.rowId !== undefined) {
         callbacks.onRowsChanged(change.reason, this._toCellMap(change.rowId, this.getFieldInfos()));
       } else {
@@ -339,7 +339,8 @@ export class RowInfo {
     public readonly viewId: string,
     public readonly fieldInfos: readonly FieldInfo[],
     public readonly row: RowPB
-  ) {}
+  ) {
+  }
 
   copyWith = (params: { row?: RowPB; fieldInfos?: readonly FieldInfo[] }) => {
     return new RowInfo(this.viewId, params.fieldInfos || this.fieldInfos, params.row || this.row);
@@ -347,15 +348,18 @@ export class RowInfo {
 }
 
 export class DeletedRow {
-  constructor(public readonly index: number, public readonly rowInfo: RowInfo) {}
+  constructor(public readonly index: number, public readonly rowInfo: RowInfo) {
+  }
 }
 
 export class InsertedRow {
-  constructor(public readonly index: number, public readonly rowId: string) {}
+  constructor(public readonly index: number, public readonly rowId: string) {
+  }
 }
 
 export class RowChanged {
-  constructor(public readonly reason: RowChangedReason, public readonly rowId?: string) {}
+  constructor(public readonly reason: RowChangedReason, public readonly rowId?: string) {
+  }
 }
 
 // eslint-disable-next-line no-shadow

frontend/appflowy_tauri/src/appflowy_app/stores/effects/database/view/view_row_observer.ts

@@ -1,14 +1,14 @@
-import { Ok, Result } from 'ts-results';
+import { Ok, Result } from "ts-results";
 import {
   DatabaseNotification,
   FlowyError,
   ReorderAllRowsPB,
   ReorderSingleRowPB,
   RowsChangesetPB,
-  RowsVisibilityChangesetPB,
-} from '@/services/backend';
-import { ChangeNotifier } from '$app/utils/change_notifier';
-import { DatabaseNotificationObserver } from '../notifications/observer';
+  RowsVisibilityChangesetPB
+} from "@/services/backend";
+import { ChangeNotifier } from "$app/utils/change_notifier";
+import { DatabaseNotificationObserver } from "../notifications/observer";
 
 export type RowsVisibilityNotifyValue = Result<RowsVisibilityChangesetPB, FlowyError>;
 export type RowsNotifyValue = Result<RowsChangesetPB, FlowyError>;
@@ -23,7 +23,8 @@ export class DatabaseViewRowsObserver {
 
   private _listener?: DatabaseNotificationObserver;
 
-  constructor(public readonly viewId: string) {}
+  constructor(public readonly viewId: string) {
+  }
 
   subscribe = async (callbacks: {
     onRowsVisibilityChanged?: (value: RowsVisibilityNotifyValue) => void;
@@ -32,10 +33,10 @@ export class DatabaseViewRowsObserver {
     onReorderSingleRow?: (value: ReorderSingleRowNotifyValue) => void;
   }) => {
     //
-    this.rowsVisibilityNotifier.observer.subscribe(callbacks.onRowsVisibilityChanged);
-    this.rowsNotifier.observer.subscribe(callbacks.onNumberOfRowsChanged);
-    this.reorderRowsNotifier.observer.subscribe(callbacks.onReorderRows);
-    this.reorderSingleRowNotifier.observer.subscribe(callbacks.onReorderSingleRow);
+    this.rowsVisibilityNotifier.observer?.subscribe(callbacks.onRowsVisibilityChanged);
+    this.rowsNotifier.observer?.subscribe(callbacks.onNumberOfRowsChanged);
+    this.reorderRowsNotifier.observer?.subscribe(callbacks.onReorderRows);
+    this.reorderSingleRowNotifier.observer?.subscribe(callbacks.onReorderSingleRow);
 
     this._listener = new DatabaseNotificationObserver({
       id: this.viewId,
@@ -72,7 +73,7 @@ export class DatabaseViewRowsObserver {
           default:
             break;
         }
-      },
+      }
     });
     await this._listener.start();
   };

frontend/appflowy_tauri/src/appflowy_app/stores/effects/folder/app/app_observer.ts

@@ -1,7 +1,7 @@
-import { Ok, Result } from 'ts-results';
-import { FlowyError, FolderNotification, RepeatedViewPB } from '@/services/backend';
-import { ChangeNotifier } from '$app/utils/change_notifier';
-import { FolderNotificationObserver } from '../notifications/observer';
+import { Ok, Result } from "ts-results";
+import { FlowyError, FolderNotification, RepeatedViewPB } from "@/services/backend";
+import { ChangeNotifier } from "$app/utils/change_notifier";
+import { FolderNotificationObserver } from "../notifications/observer";
 
 export type AppUpdateNotifyCallback = (value: Result<RepeatedViewPB, FlowyError>) => void;
 
@@ -9,10 +9,11 @@ export class AppObserver {
   _appNotifier = new ChangeNotifier<Result<RepeatedViewPB, FlowyError>>();
   _listener?: FolderNotificationObserver;
 
-  constructor(public readonly appId: string) {}
+  constructor(public readonly appId: string) {
+  }
 
   subscribe = async (callbacks: { onAppChanged: AppUpdateNotifyCallback }) => {
-    this._appNotifier?.observer.subscribe(callbacks.onAppChanged);
+    this._appNotifier?.observer?.subscribe(callbacks.onAppChanged);
     this._listener = new FolderNotificationObserver({
       viewId: this.appId,
       parserHandler: (notification, result) => {
@@ -27,7 +28,7 @@ export class AppObserver {
           default:
             break;
         }
-      },
+      }
     });
     await this._listener.start();
   };

frontend/appflowy_tauri/src/appflowy_app/stores/effects/folder/view/view_observer.ts

@@ -1,7 +1,7 @@
-import { Ok, Result } from 'ts-results';
-import { DeletedViewPB, FolderNotification, ViewPB, FlowyError } from '@/services/backend';
-import { ChangeNotifier } from '$app/utils/change_notifier';
-import { FolderNotificationObserver } from '../notifications/observer';
+import { Ok, Result } from "ts-results";
+import { DeletedViewPB, FolderNotification, ViewPB, FlowyError } from "@/services/backend";
+import { ChangeNotifier } from "$app/utils/change_notifier";
+import { FolderNotificationObserver } from "../notifications/observer";
 
 type DeleteViewNotifyValue = Result<ViewPB, FlowyError>;
 type UpdateViewNotifyValue = Result<ViewPB, FlowyError>;
@@ -15,7 +15,8 @@ export class ViewObserver {
   private _moveToTashNotifier = new ChangeNotifier<MoveToTrashViewNotifyValue>();
   private _listener?: FolderNotificationObserver;
 
-  constructor(public readonly viewId: string) {}
+  constructor(public readonly viewId: string) {
+  }
 
   subscribe = async (callbacks: {
     onViewUpdate?: (value: UpdateViewNotifyValue) => void;
@@ -24,19 +25,19 @@ export class ViewObserver {
     onViewMoveToTrash?: (value: MoveToTrashViewNotifyValue) => void;
   }) => {
     if (callbacks.onViewDelete !== undefined) {
-      this._deleteViewNotifier.observer.subscribe(callbacks.onViewDelete);
+      this._deleteViewNotifier.observer?.subscribe(callbacks.onViewDelete);
     }
 
     if (callbacks.onViewUpdate !== undefined) {
-      this._updateViewNotifier.observer.subscribe(callbacks.onViewUpdate);
+      this._updateViewNotifier.observer?.subscribe(callbacks.onViewUpdate);
     }
 
     if (callbacks.onViewRestored !== undefined) {
-      this._restoreViewNotifier.observer.subscribe(callbacks.onViewRestored);
+      this._restoreViewNotifier.observer?.subscribe(callbacks.onViewRestored);
     }
 
     if (callbacks.onViewMoveToTrash !== undefined) {
-      this._moveToTashNotifier.observer.subscribe(callbacks.onViewMoveToTrash);
+      this._moveToTashNotifier.observer?.subscribe(callbacks.onViewMoveToTrash);
     }
 
     this._listener = new FolderNotificationObserver({
@@ -74,7 +75,7 @@ export class ViewObserver {
           default:
             break;
         }
-      },
+      }
     });
     await this._listener.start();
   };

frontend/appflowy_tauri/src/appflowy_app/stores/effects/folder/workspace/workspace_observer.ts

@@ -1,7 +1,7 @@
-import { Ok, Result } from 'ts-results';
-import { FolderNotification, WorkspacePB, FlowyError, RepeatedViewPB, ViewPB } from '@/services/backend';
-import { ChangeNotifier } from '$app/utils/change_notifier';
-import { FolderNotificationObserver } from '../notifications/observer';
+import { Ok, Result } from "ts-results";
+import { FolderNotification, WorkspacePB, FlowyError, RepeatedViewPB, ViewPB } from "@/services/backend";
+import { ChangeNotifier } from "$app/utils/change_notifier";
+import { FolderNotificationObserver } from "../notifications/observer";
 
 export type AppListNotifyValue = Result<ViewPB[], FlowyError>;
 export type AppListNotifyCallback = (value: AppListNotifyValue) => void;
@@ -13,14 +13,15 @@ export class WorkspaceObserver {
   private workspaceNotifier = new ChangeNotifier<WorkspaceNotifyValue>();
   private listener?: FolderNotificationObserver;
 
-  constructor(public readonly workspaceId: string) {}
+  constructor(public readonly workspaceId: string) {
+  }
 
   subscribe = async (callbacks: {
     onAppListChanged: AppListNotifyCallback;
     onWorkspaceChanged: WorkspaceNotifyCallback;
   }) => {
-    this.appListNotifier?.observer.subscribe(callbacks.onAppListChanged);
-    this.workspaceNotifier?.observer.subscribe(callbacks.onWorkspaceChanged);
+    this.appListNotifier?.observer?.subscribe(callbacks.onAppListChanged);
+    this.workspaceNotifier?.observer?.subscribe(callbacks.onWorkspaceChanged);
 
     this.listener = new FolderNotificationObserver({
       viewId: this.workspaceId,
@@ -43,7 +44,7 @@ export class WorkspaceObserver {
           default:
             break;
         }
-      },
+      }
     });
     await this.listener.start();
   };

frontend/appflowy_tauri/src/appflowy_app/utils/change_notifier.ts

@@ -1,17 +1,29 @@
-import { Subject } from 'rxjs';
+import { Observable, Subject } from 'rxjs';
 
 export class ChangeNotifier<T> {
+  private isUnsubscribe = false;
   private subject = new Subject<T>();
 
   notify(value: T) {
     this.subject.next(value);
   }
 
-  get observer() {
+  get observer(): Observable<T> | null {
+    if (this.isUnsubscribe) {
+      return null;
+    }
     return this.subject.asObservable();
   }
 
+  // Unsubscribe the subject might cause [UnsubscribedError] error if there is
+  // ongoing Observable execution.
+  //
+  // Maybe you should use the [Subscription] that returned when call subscribe on
+  // [Observable] to unsubscribe.
   unsubscribe = () => {
-    this.subject.unsubscribe();
+    if (!this.isUnsubscribe) {
+      this.isUnsubscribe = true;
+      this.subject.unsubscribe();
+    }
   };
 }