fix(shareReplay): properly uses lift (#2924)

resolves #2921

spec/operators/shareReplay-spec.ts

@@ -179,4 +179,36 @@ describe('Observable.prototype.shareReplay', () => {
     rxTestScheduler.flush();
     expect(results).to.deep.equal([0, 1, 2, 4, 5, 6, 7, 8, 9]);
   });
+
+  it('should not break lift() composability', (done: MochaDone) => {
+    class MyCustomObservable<T> extends Rx.Observable<T> {
+      lift<R>(operator: Rx.Operator<T, R>): Rx.Observable<R> {
+        const observable = new MyCustomObservable<R>();
+        (<any>observable).source = this;
+        (<any>observable).operator = operator;
+        return observable;
+      }
+    }
+
+    const result = new MyCustomObservable((observer: Rx.Observer<number>) => {
+      observer.next(1);
+      observer.next(2);
+      observer.next(3);
+      observer.complete();
+    }).shareReplay();
+
+    expect(result instanceof MyCustomObservable).to.be.true;
+
+    const expected = [1, 2, 3];
+
+    result
+      .subscribe((n: any) => {
+        expect(expected.length).to.be.greaterThan(0);
+        expect(n).to.equal(expected.shift());
+      }, (x) => {
+        done(new Error('should not be called'));
+      }, () => {
+        done();
+      });
+  });
 });

src/operators/shareReplay.ts

@@ -3,18 +3,24 @@ import { ReplaySubject } from '../ReplaySubject';
 import { IScheduler } from '../Scheduler';
 import { Subscription } from '../Subscription';
 import { MonoTypeOperatorFunction } from '../interfaces';
+import { Subscriber } from '../Subscriber';
+
 /**
  * @method shareReplay
  * @owner Observable
  */
 export function shareReplay<T>(bufferSize?: number, windowTime?: number, scheduler?: IScheduler ): MonoTypeOperatorFunction<T> {
+  return (source: Observable<T>) => source.lift(shareReplayOperator(bufferSize, windowTime, scheduler));
+}
+
+export function shareReplayOperator<T>(bufferSize?: number, windowTime?: number, scheduler?: IScheduler) {
   let subject: ReplaySubject<T>;
   let refCount = 0;
   let subscription: Subscription;
   let hasError = false;
   let isComplete = false;
 
-  return (source: Observable<T>) => new Observable<T>(observer => {
+  return function shareReplayOperation(this: Subscriber<T>, source: Observable<T>) {
     refCount++;
     if (!subject || hasError) {
       hasError = false;
@@ -32,7 +38,7 @@ export function shareReplay<T>(bufferSize?: number, windowTime?: number, schedul
       });
     }
 
-    const innerSub = subject.subscribe(observer);
+    const innerSub = subject.subscribe(this);
 
     return () => {
       refCount--;
@@ -41,5 +47,5 @@ export function shareReplay<T>(bufferSize?: number, windowTime?: number, schedul
         subscription.unsubscribe();
       }
     };
-  });
-};
+  };
+};