fix: chain safe subscribers to interop subscribers

Closes ReactiveX#5469 ReactiveX#5311 ReactiveX#2675
cartant · Jun 9, 2020 · a9eaae8 · a9eaae8
1 parent 4439308
commit a9eaae8
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 3 deletions.
diff --git a/src/internal/Subscriber.ts b/src/internal/Subscriber.ts
@@ -1,7 +1,7 @@
 import { isFunction } from './util/isFunction';
 import { empty as emptyObserver } from './Observer';
 import { Observer, PartialObserver } from './types';
-import { Subscription } from './Subscription';
+import { Subscription, isSubscription } from './Subscription';
 import { rxSubscriber as rxSubscriberSymbol } from '../internal/symbol/rxSubscriber';
 import { config } from './config';
 import { hostReportError } from './util/hostReportError';
@@ -187,8 +187,8 @@ export class SafeSubscriber<T> extends Subscriber<T> {
       complete = (<PartialObserver<T>> observerOrNext).complete;
       if (observerOrNext !== emptyObserver) {
         context = Object.create(observerOrNext);
-        if (isFunction(context.unsubscribe)) {
-          this.add(<() => void> context.unsubscribe.bind(context));
+        if (isSubscription(observerOrNext)) {
+          observerOrNext.add(this.unsubscribe.bind(this));
         }
         context.unsubscribe = this.unsubscribe.bind(this);
       }

diff --git a/src/internal/Subscription.ts b/src/internal/Subscription.ts
@@ -206,6 +206,12 @@ export class Subscription implements SubscriptionLike {
   }
 }
 
+export function isSubscription(value: object): value is Subscription {
+  return value &&
+    typeof (value as any).add === 'function' &&
+    typeof (value as any).unsubscribe === 'function';
+}
+
 function flattenUnsubscriptionErrors(errors: any[]) {
  return errors.reduce((errs, err) => errs.concat((err instanceof UnsubscriptionError) ? err.errors : err), []);
 }