fix(refCount): Support multiple, sync subscribes.

Author: cartant

source/operators/refCountAuditTime-spec.ts

@@ -5,7 +5,7 @@
 /*tslint:disable:no-unused-expression*/
 
 import { expect } from "chai";
-import { of } from "rxjs";
+import { defer, of } from "rxjs";
 import { mergeMapTo, publish, toArray } from "rxjs/operators";
 import { marbles } from "rxjs-marbles";
 import { refCountAuditTime } from "./refCountAuditTime";
@@ -129,4 +129,19 @@ describe("refCountAuditTime", () => {
         const published = source.pipe(publish(), refCountAuditTime(0), toArray());
         return published.toPromise().then(value => expect(value).to.deep.equal([1, 2, 3]));
     });
+
+    it("should support mutliple, synchronous subscriptions", () => {
+
+        let subscribes = 0;
+        const source = defer(() => {
+            ++subscribes;
+            return of(1, 2, 3);
+        });
+
+        const published = source.pipe(publish(), refCountAuditTime(0), toArray());
+        return Promise.all([
+            published.toPromise(),
+            published.toPromise()
+        ]).then(() => expect(subscribes).to.equal(1));
+    });
 });

source/operators/refCountAuditTime.ts

@@ -4,6 +4,7 @@
  */
 
 import {
+    asapScheduler,
     ConnectableObservable,
     MonoTypeOperatorFunction,
     NEVER,
@@ -19,7 +20,7 @@ import { scan, switchMap, tap } from "rxjs/operators";
 
 export function refCountAuditTime<T>(
     duration: number,
-    scheduler?: SchedulerLike
+    scheduler: SchedulerLike = asapScheduler
 ): MonoTypeOperatorFunction<T> {
 
     return (source: Observable<T>) => {
@@ -32,31 +33,39 @@ export function refCountAuditTime<T>(
         // https://github.com/ReactiveX/rxjs/issues/171#issuecomment-267881847
 
         const connectable: ConnectableObservable<T> = source as any;
-        let subscription: Subscription | null = null;
+        let connectableSubscription: Subscription | null = null;
+        let connectorSubscription: Subscription | null = null;
 
         const notifier = new Subject<number>();
         const connector = notifier.pipe(
             scan((count, step) => count + step, 0),
             switchMap(count => {
-                switch (count) {
-                case 0:
+                if (count === 0) {
                     return timer(duration, scheduler).pipe(tap(() => {
-                        subscription!.unsubscribe();
-                        subscription = null;
+                        if (connectableSubscription) {
+                            connectableSubscription.unsubscribe();
+                            connectableSubscription = null;
+                        }
+                        if (connectorSubscription) {
+                            connectorSubscription.unsubscribe();
+                            connectorSubscription = null;
+                        }
                     }));
-                case 1:
+                }
+                if (!connectableSubscription && (count > 0)) {
                     return timer(0, scheduler).pipe(tap(() => {
-                        subscription!.add(connectable.connect());
+                        if (!connectableSubscription) {
+                            connectableSubscription = connectable.connect();
+                        }
                     }));
-                default:
-                    return NEVER;
                 }
+                return NEVER;
             })
         );
 
         return using(() => {
-            if (subscription === null) {
-                subscription = connector.subscribe();
+            if (!connectorSubscription) {
+                connectorSubscription = connector.subscribe();
             }
             notifier.next(1);
             return { unsubscribe: () => notifier.next(-1) };