Merge pull request #682 from SuperViz/fix/multiple-subscriptions-stores

fix: properly unsubscribe from subject that was subscribed to twice

src/common/types/stores.types.ts

@@ -13,7 +13,7 @@ export enum StoreType {
 
 type Subject<T extends (...args: any[]) => any, K extends keyof ReturnType<T>> = ReturnType<T>[K];
 
-type StoreApiWithoutDestroy<T extends (...args: any[]) => any> = {
+type IncompleteStoreApi<T extends (...args: any[]) => any> = {
   [K in keyof ReturnType<T>]: {
     subscribe(callback?: (value: Subject<T, K>['value']) => void): void;
     subject: Subject<T, K>;
@@ -22,8 +22,9 @@ type StoreApiWithoutDestroy<T extends (...args: any[]) => any> = {
   };
 };
 
-type StoreApi<T extends (...args: any[]) => any> = Omit<StoreApiWithoutDestroy<T>, 'destroy'> & {
+type StoreApi<T extends (...args: any[]) => any> = IncompleteStoreApi<T> & {
   destroy(): void;
+  restart(): void;
 };
 
 type GlobalStore = StoreType.GLOBAL | `${StoreType.GLOBAL}`;

src/components/video/index.test.ts

@@ -7,7 +7,6 @@ import { EVENT_BUS_MOCK } from '../../../__mocks__/event-bus.mock';
 import { MOCK_OBSERVER_HELPER } from '../../../__mocks__/observer-helper.mock';
 import { MOCK_AVATAR, MOCK_LOCAL_PARTICIPANT } from '../../../__mocks__/participants.mock';
 import { ABLY_REALTIME_MOCK } from '../../../__mocks__/realtime.mock';
-import { ROOM_STATE_MOCK } from '../../../__mocks__/roomState.mock';
 import {
   DeviceEvent,
   FrameEvent,
@@ -25,7 +24,6 @@ import { useStore } from '../../common/utils/use-store';
 import { IOC } from '../../services/io';
 import { Presence3DManager } from '../../services/presence-3d-manager';
 import { ParticipantInfo } from '../../services/realtime/base/types';
-import { RoomStateService } from '../../services/roomState';
 import { VideoFrameState } from '../../services/video-conference-manager/types';
 import { ComponentNames } from '../types';
 
@@ -97,12 +95,13 @@ describe('VideoConference', () => {
   let VideoConferenceInstance: VideoConference;
 
   const { localParticipant, hasJoinedRoom } = useStore(StoreType.GLOBAL);
-  localParticipant.publish(MOCK_LOCAL_PARTICIPANT);
-  hasJoinedRoom.publish(true);
   beforeEach(() => {
     jest.clearAllMocks();
     jest.restoreAllMocks();
 
+    localParticipant.publish(MOCK_LOCAL_PARTICIPANT);
+    hasJoinedRoom.publish(true);
+
     VideoConferenceInstance = new VideoConference({
       userType: 'host',
       allowGuests: false,
@@ -118,6 +117,7 @@ describe('VideoConference', () => {
       useStore,
     });
 
+    VideoConferenceInstance['startVideo']();
     VideoConferenceInstance['onFrameStateChange'](VideoFrameState.INITIALIZED);
   });
 

src/components/who-is-online/index.test.ts

@@ -7,7 +7,6 @@ import {
   MOCK_ABLY_PARTICIPANT_DATA_1,
 } from '../../../__mocks__/participants.mock';
 import { ABLY_REALTIME_MOCK } from '../../../__mocks__/realtime.mock';
-import { ROOM_STATE_MOCK } from '../../../__mocks__/roomState.mock';
 import { RealtimeEvent, WhoIsOnlineEvent } from '../../common/types/events.types';
 import { MeetingColorsHex } from '../../common/types/meeting-colors.types';
 import { StoreType } from '../../common/types/stores.types';

src/core/launcher/index.test.ts

@@ -13,7 +13,6 @@ import { DefaultAttachComponentOptions } from '../../components/base/types';
 import { ComponentNames } from '../../components/types';
 import { IOC } from '../../services/io';
 import LimitsService from '../../services/limits';
-import { AblyParticipant } from '../../services/realtime/ably/types';
 import { useGlobalStore } from '../../services/stores';
 
 import { LauncherFacade, LauncherOptions } from './types';

src/services/stores/presence3D/index.ts

@@ -20,6 +20,7 @@ class Presence3DStore {
 
   public destroy() {
     this.hasJoined3D.destroy();
+    this.participants.destroy();
   }
 }
 

src/services/stores/subject/index.test.ts

@@ -82,7 +82,7 @@ describe('base subject for all stores', () => {
       const unsubscribe = jest.fn();
 
       instance['subscribe'](testValues.id, callback);
-      instance['subscriptions'].get(testValues.id)!.unsubscribe = unsubscribe;
+      instance['subscriptions'].get(testValues.id)![0].unsubscribe = unsubscribe;
 
       instance['unsubscribe'](testValues.id);
 
@@ -97,16 +97,14 @@ describe('base subject for all stores', () => {
       instance = subject<string>(testValues.str1);
 
       const callback = jest.fn();
-      const unsubscribe = jest.fn();
 
       instance['subscribe'](testValues.id, callback);
-      instance['subscriptions'].get(testValues.id)!.unsubscribe = unsubscribe;
 
       instance['destroy']();
 
       expect(instance['subscriptions'].size).toBe(0);
       expect(instance['subscriptions'].get(testValues.id)).toBeUndefined();
-      expect(unsubscribe).toHaveBeenCalled();
+      expect(instance.state).toBe(instance['firstState']);
     });
   });
 

src/services/stores/subject/index.ts

@@ -5,11 +5,15 @@ import { PublicSubject } from '../common/types';
 
 export class Subject<T> {
   public state: T;
+  private firstState: T;
+
   private subject: BehaviorSubject<T>;
-  private subscriptions: Map<string | this, Subscription> = new Map();
+  private subscriptions: Map<string | this, Subscription[]> = new Map();
 
   constructor(state: T, subject: BehaviorSubject<T>) {
     this.state = state;
+    this.firstState = state;
+
     this.subject = subject.pipe(
       distinctUntilChanged(),
       shareReplay({ bufferSize: 1, refCount: true }),
@@ -27,17 +31,34 @@ export class Subject<T> {
 
   public subscribe = (subscriptionId: string | this, callback: (value: T) => void) => {
     const subscription = this.subject.subscribe(callback);
-    this.subscriptions.set(subscriptionId, subscription);
+
+    if (this.subscriptions.has(subscriptionId)) {
+      this.subscriptions.get(subscriptionId).push(subscription);
+      return;
+    }
+
+    this.subscriptions.set(subscriptionId, [subscription]);
   };
 
   public unsubscribe(subscriptionId: string) {
-    this.subscriptions.get(subscriptionId)?.unsubscribe();
+    this.subscriptions.get(subscriptionId)?.forEach((subscription) => subscription.unsubscribe());
     this.subscriptions.delete(subscriptionId);
   }
 
   public destroy() {
-    this.subscriptions.forEach((subscription) => subscription.unsubscribe());
     this.subscriptions.clear();
+    this.subject.complete();
+
+    this.restart();
+  }
+
+  private restart() {
+    this.state = this.firstState;
+
+    this.subject = new BehaviorSubject<T>(this.firstState).pipe(
+      distinctUntilChanged(),
+      shareReplay({ bufferSize: 1, refCount: true }),
+    ) as BehaviorSubject<T>;
   }
 
   public expose(): PublicSubject<T> {

src/services/stores/who-is-online/index.ts

@@ -42,8 +42,6 @@ export class WhoIsOnlineStore {
     this.everyoneFollowsMe.destroy();
     this.privateMode.destroy();
     this.following.destroy();
-
-    instance.value = null;
   }
 }