[WIP] fix: ensure all interop operators are chained correctly

spec/helpers/interop-helper.ts

@@ -1,4 +1,4 @@
-import { Observable, Subscriber, Subscription } from 'rxjs';
+import { Observable, Operator, Subscriber, Subscription } from 'rxjs';
 import { rxSubscriber as symbolSubscriber } from 'rxjs/internal/symbol/rxSubscriber';
 
 /**
@@ -10,16 +10,21 @@ import { rxSubscriber as symbolSubscriber } from 'rxjs/internal/symbol/rxSubscri
 export function asInteropObservable<T>(observable: Observable<T>): Observable<T> {
   return new Proxy(observable, {
     get(target: Observable<T>, key: string | number | symbol) {
+      if (key === 'lift') {
+        const { lift } = target;
+        return interopLift(lift);
+      }
       if (key === 'subscribe') {
         const { subscribe } = target;
         return interopSubscribe(subscribe);
       }
       return Reflect.get(target, key);
     },
     getPrototypeOf(target: Observable<T>) {
-      const { subscribe, ...rest } = Object.getPrototypeOf(target);
+      const { lift, subscribe, ...rest } = Object.getPrototypeOf(target);
       return {
         ...rest,
+        lift: interopLift(lift),
         subscribe: interopSubscribe(subscribe)
       };
     }
@@ -46,6 +51,18 @@ export function asInteropSubscriber<T>(subscriber: Subscriber<T>): Subscriber<T>
   });
 }
 
+function interopLift<T, R>(lift: (operator: Operator<T, R>) => Observable<R>) {
+  return function (this: Observable<T>, operator: Operator<T, R>): Observable<R> {
+    const observable = lift.call(this, operator);
+    const { call } = observable.operator;
+    observable.operator.call = function (this: Operator<T, R>, subscriber: Subscriber<T>, source: any) {
+      return call.call(this, asInteropSubscriber(subscriber), source);
+    };
+    observable.source = asInteropObservable(observable.source);
+    return asInteropObservable(observable);
+  };
+}
+
 function interopSubscribe<T>(subscribe: (...args: any[]) => Subscription) {
   return function (this: Observable<T>, ...args: any[]): Subscription {
     const [arg] = args;
@@ -54,4 +71,4 @@ function interopSubscribe<T>(subscribe: (...args: any[]) => Subscription) {
     }
     return subscribe.apply(this, args);
   };
-}
+}

spec/operators/finalize-spec.ts

@@ -2,7 +2,8 @@ import { expect } from 'chai';
 import { finalize, map, share } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { of, timer, interval } from 'rxjs';
+import { of, timer, interval, NEVER } from 'rxjs';
+import { asInteropObservable } from '../helpers/interop-helper';
 
 declare const type: Function;
 
@@ -161,4 +162,14 @@ describe('finalize operator', () => {
     rxTestScheduler.flush();
     expect(executed).to.be.true;
   });
+
+  it('should handle interop source observables', () => {
+    // https://github.com/ReactiveX/rxjs/issues/5237
+    let finalized = false;
+    const subscription = asInteropObservable(NEVER).pipe(
+      finalize(() => finalized = true)
+    ).subscribe();
+    subscription.unsubscribe();
+    expect(finalized).to.be.true;
+  });
 });

src/internal/operators/finalize.ts

@@ -3,6 +3,7 @@ import { Subscriber } from '../Subscriber';
 import { Subscription } from '../Subscription';
 import { Observable } from '../Observable';
 import { MonoTypeOperatorFunction, TeardownLogic } from '../types';
+import { subscribeAndChainOperator } from '../util/subscribeAndChainOperator';
 
 /**
  * Returns an Observable that mirrors the source Observable, but will call a specified function when
@@ -69,7 +70,10 @@ class FinallyOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source.subscribe(new FinallySubscriber(subscriber, this.callback));
+    return subscribeAndChainOperator(
+      new FinallySubscriber(subscriber, this.callback),
+      source
+    );
   }
 }
 

src/internal/util/subscribeAndChainOperator.ts

@@ -0,0 +1,21 @@
+import { Observable } from '../Observable';
+import { Subscriber } from '../Subscriber';
+import { Subscription } from '../Subscription';
+
+/**
+ * Subscribes an operator's subscriber to its source and ensures that
+ * unsubscription is chained for interop operators.
+ */
+export function subscribeAndChainOperator<T>(
+  subscriber: Subscriber<T>,
+  source: Observable<T>
+): Subscription {
+  const subscription = source.subscribe(subscriber);
+  // The returned subscription will usually be the operator's subscriber.
+  // However, interop subscribers will be wrapped and for
+  // unsubscriptions to chain correctly, the wrapper needs to be added, too.
+  if (subscription !== subscriber) {
+    subscription.add(subscriber);
+  }
+  return subscription;
+}