fix(shareReplay): properly retains history on subscribe

spec/operators/shareReplay-spec.ts

@@ -7,6 +7,8 @@ declare const hot: typeof marbleTestingSignature.hot;
 declare const cold: typeof marbleTestingSignature.cold;
 declare const expectObservable: typeof marbleTestingSignature.expectObservable;
 declare const expectSubscriptions: typeof marbleTestingSignature.expectSubscriptions;
+declare const time: typeof marbleTestingSignature.time;
+declare const rxTestScheduler: typeof marbleTestingSignature.rxTestScheduler;
 
 const Observable = Rx.Observable;
 
@@ -164,4 +166,17 @@ describe('Observable.prototype.shareReplay', () => {
     expectObservable(subscriber3).toBe(expected3);
     expectSubscriptions(source.subscriptions).toBe(subs);
   });
+
+  it('should not restart if refCount hits 0 due to unsubscriptions', () => {
+    const results = [];
+    const source = Rx.Observable.interval(10, rxTestScheduler)
+      .take(10)
+      .shareReplay(1);
+    const subs = source.subscribe(x => results.push(x));
+    rxTestScheduler.schedule(() => subs.unsubscribe(), 35);
+    rxTestScheduler.schedule(() => source.subscribe(x => results.push(x)), 54);
+
+    rxTestScheduler.flush();
+    expect(results).to.deep.equal([0, 1, 2, 4, 5, 6, 7, 8, 9]);
+  });
 });

src/operators/shareReplay.ts

@@ -1,25 +1,45 @@
 import { Observable } from '../Observable';
-import { multicast } from './multicast';
-import { refCount } from './refCount';
 import { ReplaySubject } from '../ReplaySubject';
-import { ConnectableObservable } from '../observable/ConnectableObservable';
 import { IScheduler } from '../Scheduler';
-
+import { Subscription } from '../Subscription';
 import { MonoTypeOperatorFunction } from '../interfaces';
-
 /**
  * @method shareReplay
  * @owner Observable
  */
 export function shareReplay<T>(bufferSize?: number, windowTime?: number, scheduler?: IScheduler ): MonoTypeOperatorFunction<T> {
   let subject: ReplaySubject<T>;
+  let refCount = 0;
+  let subscription: Subscription;
+  let hasError = false;
+  let isComplete = true;
 
-  const connectable = multicast(function shareReplaySubjectFactory(this: ConnectableObservable<T>) {
-    if (this._isComplete) {
-      return subject;
-    } else {
-      return (subject = new ReplaySubject<T>(bufferSize, windowTime, scheduler));
+  return (source: Observable<T>) => new Observable<T>(observer => {
+    refCount++;
+    if (!subject || hasError) {
+      hasError = false;
+      subject = new ReplaySubject<T>(bufferSize, windowTime, scheduler);
+      subscription = source.subscribe({
+        next(value) { subject.next(value); },
+        error(err) {
+          hasError = true;
+          subject.error(err);
+        },
+        complete() {
+          isComplete = true;
+          subject.complete();
+        },
+      });
     }
+
+    const innerSub = subject.subscribe(observer);
+
+    return () => {
+      refCount--;
+      innerSub.unsubscribe();
+      if (subscription && refCount === 0 && !isComplete) {
+        subscription.unsubscribe();
+      }
+    };
   });
-  return ((source: Observable<T>) => refCount()(connectable(source))) as MonoTypeOperatorFunction<T>;
-};
+};