fix(Subscriber): use only local Subscriber instances

This removes any checks for `Subscriber` symbol or . That was a micro-optimization that proved to be problematic for node projects with more than one instance of RxJS.

This means that all local `Subject` instances will be wrapped by
`SafeSubscriber`, but prior to that, they were wrapped in
`SubjectSubscriber` anyhow, so the difference is minimal.

Leaves behind `SubjectSubscriber` and implementations of `rxSubscriber`,
which are still functioning, but now cruft.

spec/Subscriber-spec.ts

@@ -1,7 +1,6 @@
 import { expect } from 'chai';
 import { SafeSubscriber } from 'rxjs/internal/Subscriber';
 import { Subscriber } from 'rxjs';
-import { rxSubscriber } from 'rxjs/internal/symbol/rxSubscriber';
 
 /** @test {Subscriber} */
 describe('Subscriber', () => {
@@ -20,17 +19,6 @@ describe('Subscriber', () => {
  expect(times).to.equal(2);
  });
 
- it('should accept subscribers as a destination if they meet the proper criteria', () => {
- const fakeSubscriber = {
- [rxSubscriber](this: any) { return this; },
- add() { /* noop */ },
- syncErrorThrowable: false
- };
-
- const subscriber = new Subscriber(fakeSubscriber as any);
- expect((subscriber as any).destination).to.equal(fakeSubscriber);
- });
-
  it('should wrap unsafe observers in a safe subscriber', () => {
  const observer = {
  next(x: any) { /* noop */ },

src/internal/Subscriber.ts

@@ -72,13 +72,10 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
  break;
  }
  if (typeof destinationOrNext === 'object') {
- // 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);
+ if (destinationOrNext instanceof Subscriber) {
+ this.syncErrorThrowable = destinationOrNext.syncErrorThrowable;
+ this.destination = destinationOrNext;
+ destinationOrNext.add(this);
  } else {
  this.syncErrorThrowable = true;
  this.destination = new SafeSubscriber<T>(this, <PartialObserver<any>> destinationOrNext);
@@ -308,7 +305,3 @@ export class SafeSubscriber<T> extends Subscriber<T> {
  _parentSubscriber.unsubscribe();
  }
 }
-
-export function isTrustedSubscriber(obj: any) {
- return obj instanceof Subscriber || ('syncErrorThrowable' in obj && obj[rxSubscriberSymbol]);
-}

src/internal/symbol/rxSubscriber.ts

@@ -1,3 +1,4 @@
+/** @deprecated do not use, this is no longer checked by RxJS internals */
 export const rxSubscriber =
  (typeof Symbol === 'function' && typeof Symbol.for === 'function')
  ? Symbol.for('rxSubscriber')

src/internal/util/canReportError.ts

@@ -1,4 +1,4 @@
-import { isTrustedSubscriber, Subscriber } from '../Subscriber';
+import { Subscriber } from '../Subscriber';
 import { Subject } from '../Subject';
 
 /**
@@ -12,7 +12,7 @@ export function canReportError(observer: Subscriber<any> | Subject<any>): boolea
  const { closed, destination, isStopped } = observer as any;
  if (closed || isStopped) {
  return false;
- } else if (destination && isTrustedSubscriber(destination)) {
+ } else if (destination && destination instanceof Subscriber) {
  observer = destination;
  } else {
  observer = null;