fix(shareReplay): handle possible memory leaks

ReactiveX · Feb 16, 2021 · aff9283 · aff9283
1 parent 1e1e022
commit aff9283
Showing 1 changed file with 32 additions and 12 deletions.
diff --git a/src/internal/operators/shareReplay.ts b/src/internal/operators/shareReplay.ts
@@ -1,8 +1,8 @@
-import { Observable } from '../Observable';
-import { ReplaySubject } from '../ReplaySubject';
-import { Subscription } from '../Subscription';
-import { MonoTypeOperatorFunction, SchedulerLike } from '../types';
-import { Subscriber } from '../Subscriber';
+import { Observable } from "../Observable";
+import { ReplaySubject } from "../ReplaySubject";
+import { Subscription } from "../Subscription";
+import { MonoTypeOperatorFunction, SchedulerLike } from "../types";
+import { Subscriber } from "../Subscriber";
 
 export interface ShareReplayConfig {
   bufferSize?: number;
@@ -56,22 +56,28 @@ export interface ShareReplayConfig {
  * @method shareReplay
  * @owner Observable
  */
-export function shareReplay<T>(config: ShareReplayConfig): MonoTypeOperatorFunction<T>;
-export function shareReplay<T>(bufferSize?: number, windowTime?: number, scheduler?: SchedulerLike): MonoTypeOperatorFunction<T>;
+export function shareReplay<T>(
+  config: ShareReplayConfig
+): MonoTypeOperatorFunction<T>;
+export function shareReplay<T>(
+  bufferSize?: number,
+  windowTime?: number,
+  scheduler?: SchedulerLike
+): MonoTypeOperatorFunction<T>;
 export function shareReplay<T>(
   configOrBufferSize?: ShareReplayConfig | number,
   windowTime?: number,
   scheduler?: SchedulerLike
 ): MonoTypeOperatorFunction<T> {
   let config: ShareReplayConfig;
-  if (configOrBufferSize && typeof configOrBufferSize === 'object') {
+  if (configOrBufferSize && typeof configOrBufferSize === "object") {
     config = configOrBufferSize as ShareReplayConfig;
   } else {
     config = {
       bufferSize: configOrBufferSize as number | undefined,
       windowTime,
       refCount: false,
-      scheduler
+      scheduler,
     };
   }
   return (source: Observable<T>) => source.lift(shareReplayOperator(config));
@@ -81,23 +87,28 @@ function shareReplayOperator<T>({
   bufferSize = Number.POSITIVE_INFINITY,
   windowTime = Number.POSITIVE_INFINITY,
   refCount: useRefCount,
-  scheduler
+  scheduler,
 }: ShareReplayConfig) {
   let subject: ReplaySubject<T> | undefined;
   let refCount = 0;
   let subscription: Subscription | undefined;
   let hasError = false;
   let isComplete = false;
 
-  return function shareReplayOperation(this: Subscriber<T>, source: Observable<T>) {
+  return function shareReplayOperation(
+    this: Subscriber<T>,
+    source: Observable<T>
+  ) {
     refCount++;
     let innerSub: Subscription;
     if (!subject || hasError) {
       hasError = false;
       subject = new ReplaySubject<T>(bufferSize, windowTime, scheduler);
       innerSub = subject.subscribe(this);
       subscription = source.subscribe({
-        next(value) { subject.next(value); },
+        next(value) {
+          subject.next(value);
+        },
         error(err) {
           hasError = true;
           subject.error(err);
@@ -108,13 +119,22 @@ function shareReplayOperator<T>({
           subject.complete();
         },
       });
+
+      // Here we need to check to see if the source synchronously completed. Although
+      // we're setting `subscription = undefined` in the completion handler, if the source
+      // is synchronous, that will happen *before* subscription is set by the return of
+      // the `subscribe` call.
+      if (isComplete) {
+        subscription = undefined;
+      }
     } else {
       innerSub = subject.subscribe(this);
     }
 
     this.add(() => {
       refCount--;
       innerSub.unsubscribe();
+      innerSub = undefined;
       if (subscription && !isComplete && useRefCount && refCount === 0) {
         subscription.unsubscribe();
         subscription = undefined;