fix(node): Subscriber no longer trampled if from another copy of rxjs

Only fixes v6 and above, fix coming for v5 (stable) shortly

related ReactiveX#3475

node-tests/test.js

@@ -16,7 +16,7 @@ var id = setTimeout(function () {
 }, 200);
 
 of1(0).pipe(
- mergeMap1(function () { return of(x); }),
+ mergeMap1(function (x) { return of(x); }),
  mergeMap(function () { return from1(Promise.resolve(1)); })
 ).subscribe({
  next: function (value) { actual.push(value); },

src/internal/Subscriber.ts

@@ -70,10 +70,13 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
  break;
  }
  if (typeof destinationOrNext === 'object') {
- if (destinationOrNext instanceof Subscriber) {
- this.syncErrorThrowable = destinationOrNext.syncErrorThrowable;
- this.destination = (<Subscriber<any>> destinationOrNext);
- (<any> this.destination).add(this);
+ // HACK(benlesh): For situations where Node has multiple copies of rxjs in
+ // node_modules, we cannot rely on `instanceof` checks
+ if (isTrustedSubscriber(destinationOrNext)) {
+ const trustedSubscriber = destinationOrNext[rxSubscriberSymbol]() as Subscriber<any>;
+ this.syncErrorThrowable = trustedSubscriber.syncErrorThrowable;
+ this.destination = trustedSubscriber;
+ trustedSubscriber.add(this);
  } else {
  this.syncErrorThrowable = true;
  this.destination = new SafeSubscriber<T>(this, <PartialObserver<any>> destinationOrNext);
@@ -299,4 +302,8 @@ class SafeSubscriber<T> extends Subscriber<T> {
  this._parentSubscriber = null;
  _parentSubscriber.unsubscribe();
  }
-}
+}
+
+function isTrustedSubscriber(obj: any) {
+ return obj instanceof Subscriber || ('syncErrorThrowable' in obj && obj[rxSubscriberSymbol]);
+}