fix(node): no more silent failure for many copies of rxjs (#3477)

closes #3475

node-tests/run.sh

@@ -0,0 +1,15 @@
+rm -rf node-tests/node_modules/
+
+mkdir -p node-tests/node_modules/rxjs
+mkdir -p node-tests/node_modules/rxjs1
+
+cp -R dist/package/. node-tests/node_modules/rxjs
+cp -R dist/package/. node-tests/node_modules/rxjs1
+
+{
+   node node-tests/test.js
+  # node --inspect-brk node-tests/test.js
+} || {
+  echo 'TEST FAILED'
+}
+

node-tests/test.js

@@ -0,0 +1,32 @@
+
+var of = require('rxjs/observable/of').of;
+var of1 = require('rxjs1/observable/of').of;
+var from1 = require('rxjs1/observable/from').from;
+var mergeMap = require('rxjs/operators/mergeMap').mergeMap;
+var mergeMap1 = require('rxjs1/operators/mergeMap').mergeMap;
+
+var actual = [];
+var expected = [1];
+
+var id = setTimeout(function () {
+  throw new Error('TIMEOUT: Observable did not complete');
+}, 200);
+
+of1(0).pipe(
+  mergeMap1(function (x) { return of(x); }),
+  mergeMap(function () { return from1(Promise.resolve(1)); })
+).subscribe({
+  next: function (value) { actual.push(value); },
+  error: function (err) {
+    console.error(err);
+    throw new Error('should not error');
+  },
+  complete: function () {
+    if (actual.length !== expected.length || actual[0] !== expected[0] || actual[1] !== expected[1]) {
+      throw new Error(actual + ' does not equal ' + expected);
+    } else {
+      clearTimeout(id);
+      console.log('TEST PASSED');
+    }
+  },
+});

src/Subscriber.ts

@@ -67,10 +67,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): To resolve an issue where Node users may have multiple
+          // copies of rxjs in their node_modules directory.
+          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);
@@ -277,3 +280,7 @@ class SafeSubscriber<T> extends Subscriber<T> {
     _parentSubscriber.unsubscribe();
   }
 }
+
+function isTrustedSubscriber(obj: any) {
+  return obj instanceof Subscriber || ('syncErrorThrowable' in obj && obj[rxSubscriberSymbol]);
+}