fix(shareReplay): subscribe to subject before source subscription starts

spec/operators/shareReplay-spec.ts

@@ -3,7 +3,7 @@ import * as sinon from 'sinon';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
 import { shareReplay, mergeMapTo, retry } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
-import { Observable, Operator, Observer, of } from 'rxjs';
+import { Observable, Operator, Observer, of, from } from 'rxjs';
 
 declare const rxTestScheduler: TestScheduler;
 
@@ -265,4 +265,23 @@ describe('shareReplay operator', () => {
         done();
       });
   });
+
+  it('should not skip values on a sync source', () => {
+    const a = from(['a', 'b', 'c', 'd']);
+    // We would like for the previous line to read like this:
+    //
+    // const a = cold('(abcd|)');
+    //
+    // However, that would synchronously emit multiple values at frame 0,
+    // but it's not synchronous upon-subscription.
+    // TODO: revisit once https://github.com/ReactiveX/rxjs/issues/5523 is fixed
+
+    const x = cold(  'x-------x');
+    const expected = '(abcd)--d';
+
+    const shared = a.pipe(shareReplay(1));
+    const result = x.pipe(mergeMapTo(shared));
+    expectObservable(result).toBe(expected);
+  });
+
 });

src/internal/operators/shareReplay.ts

@@ -155,9 +155,11 @@ function shareReplayOperator<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); },
         error(err) {
@@ -169,9 +171,10 @@ function shareReplayOperator<T>({
           subject!.complete();
         },
       });
+    } else {
+      innerSub = subject.subscribe(this);
     }
 
-    const innerSub = subject.subscribe(this);
     this.add(() => {
       refCount--;
       innerSub.unsubscribe();