Share replay of computed observables

GetStream · Sep 20, 2023 · 1092f56 · 1092f56
1 parent da140cb
commit 1092f56
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 11 deletions.
diff --git a/packages/client/src/helpers/DynascaleManager.ts b/packages/client/src/helpers/DynascaleManager.ts
@@ -168,7 +168,7 @@ export class DynascaleManager {
       ),
       takeWhile((participant) => !!participant),
       distinctUntilChanged(),
-      shareReplay(1),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     // keep copy for resize observer handler
@@ -317,6 +317,7 @@ export class DynascaleManager {
       ),
       takeWhile((p) => !!p),
       distinctUntilChanged(),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     const updateMediaStreamSubscription = participant$

diff --git a/packages/client/src/sorting/comparator.ts b/packages/client/src/sorting/comparator.ts
@@ -39,7 +39,7 @@ export const descending = <T>(comparator: Comparator<T>): Comparator<T> => {
  * Creates a new comparator which conditionally applies the given comparator.
  *
  * @example
- * const shouldSortByValue = () => return false; // to turn it off
+ * const shouldSortByValue = (a, b) => a % 2 === 0; // return false to turn it off
  * const byValue = (a, b) => a < b ? - 1 : a > b ? 1 : 0;
  * const comparator = conditional(shouldSortByValue)(byValue);
  *

diff --git a/packages/client/src/store/CallState.ts b/packages/client/src/store/CallState.ts
@@ -1,5 +1,5 @@
 import { BehaviorSubject, Observable } from 'rxjs';
-import { distinctUntilChanged, map } from 'rxjs/operators';
+import { distinctUntilChanged, map, shareReplay } from 'rxjs/operators';
 import type { Patch } from './rxUtils';
 import * as RxUtils from './rxUtils';
 import {
@@ -320,24 +320,30 @@ export class CallState {
    */
   constructor() {
     this.logger = getLogger(['CallState']);
-    this.participants$ = this.participantsSubject.pipe(
-      map((ps) => ps.sort(this.sortParticipantsBy)),
+    this.participants$ = this.participantsSubject.asObservable().pipe(
+      // TODO: replace with Array.toSorted once available
+      map((ps) => [...ps].sort(this.sortParticipantsBy)),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     this.localParticipant$ = this.participants$.pipe(
       map((participants) => participants.find(isStreamVideoLocalParticipant)),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     this.remoteParticipants$ = this.participants$.pipe(
       map((participants) => participants.filter((p) => !p.isLocalParticipant)),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     this.pinnedParticipants$ = this.participants$.pipe(
       map((participants) => participants.filter((p) => !!p.pin)),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     this.dominantSpeaker$ = this.participants$.pipe(
       map((participants) => participants.find((p) => p.isDominantSpeaker)),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     this.hasOngoingScreenShare$ = this.participants$.pipe(
@@ -347,6 +353,7 @@ export class CallState {
         ),
       ),
       distinctUntilChanged(),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
 
     this.startedAt$ = this.startedAtSubject.asObservable();

diff --git a/packages/client/src/store/__tests__/CallState.test.ts b/packages/client/src/store/__tests__/CallState.test.ts
@@ -66,7 +66,8 @@ describe('CallState', () => {
       });
 
       const ps2 = state.participants;
-      expect(ps2.map((p) => p.name)).toEqual(['F', 'B', 'E', 'A', 'C', 'D']);
+      // should resolve in initial - non-mutated state as set at the beginning
+      expect(ps2.map((p) => p.name)).toEqual(['A', 'B', 'C', 'D', 'E', 'F']);
     });
 
     it('should support custom sorting', () => {

diff --git a/packages/client/src/store/rxUtils.ts b/packages/client/src/store/rxUtils.ts
@@ -1,5 +1,4 @@
-import { Observable, Subject } from 'rxjs';
-import { take } from 'rxjs/operators';
+import { Observable, Subject, combineLatest } from 'rxjs';
 
 /**
  * A value or a function which takes the current value and returns a new value.
@@ -15,10 +14,9 @@ export type Patch<T> = T | ((currentValue: T) => T);
 export const getCurrentValue = <T>(observable$: Observable<T>) => {
   let value!: T;
   let err: Error | undefined = undefined;
-  observable$
-    .pipe(take(1))
+  combineLatest([observable$])
     .subscribe({
-      next: (v) => {
+      next: ([v]) => {
         value = v;
       },
       error: (e) => {