feat(useDeprecatedNextContext): Puts deprecated next context behavior…

… behind a flag

- Adds a flag to `import { config } from 'rxjs';` that allows users to use the undocumented feature that gives access to `unsubscribe` via the `this` context of a `next` handler passed as part of an observer object to `subscribe`. This behavior is deprecated because it has very bad performance implications on all subscriptions and is relatively unused, definitely undocumented, and certainly mostly unknown.
- Adds a flag to silence console warn messages that are emitted when "bad" configuration settings are used.
- Adds some documentation.
- Adds tests.

BREAKING CHANGE: `unsubscribe` no longer available via the `this` context of observer functions. To reenable, set `config.useDeprecatedNextContext = true` on the rxjs `config` found at `import { config } from 'rxjs';`. Note that enabling this will result in a performance penalty for all consumer subscriptions.

api_guard/dist/types/index.d.ts

@@ -167,8 +167,10 @@ export declare function concat<O1 extends ObservableInput<any>, O2 extends Obser
 export declare function concat<A extends ObservableInput<any>[]>(...observables: A): Observable<ObservedValueUnionFromArray<A>>;
 
 export declare const config: {
+    quietBadConfig: boolean;
     Promise: PromiseConstructorLike;
     useDeprecatedSynchronousErrorHandling: boolean;
+    useDeprecatedNextContext: boolean;
 };
 
 export declare class ConnectableObservable<T> extends Observable<T> {

spec/Subscriber-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { SafeSubscriber } from 'rxjs/internal/Subscriber';
-import { Subscriber, Observable } from 'rxjs';
+import { Subscriber, Observable, config, of } from 'rxjs';
 import { asInteropSubscriber } from './helpers/interop-helper';
 
 /** @test {Subscriber} */
@@ -129,4 +129,93 @@ describe('Subscriber', () => {
     subscriber.unsubscribe();
     expect(count).to.equal(1);
   });
+
+  it('should NOT break this context on next methods from unfortunate consumers', () => {
+    // This is a contrived class to illustrate that we can pass another
+    // object that is "observer shaped" and not have it lose its context
+    // as it would have in v5 - v6.
+    class CustomConsumer {
+      valuesProcessed: string[] = [];
+
+      // In here, we access instance state and alter it.
+      next(value: string) {
+        if (value === 'reset') {
+          this.valuesProcessed = [];
+        } else {
+          this.valuesProcessed.push(value);
+        }
+      }
+    };
+
+    const consumer = new CustomConsumer();
+
+    of('old', 'old', 'reset', 'new', 'new').subscribe(consumer);
+
+    expect(consumer.valuesProcessed).not.to.equal(['new', 'new']);
+  });
+
+  describe('deprecated next context mode', () => {
+    beforeEach(() => {
+      config.quietBadConfig = true;
+      config.useDeprecatedNextContext = true;
+    });
+
+    afterEach(() => {
+      config.useDeprecatedNextContext = false;
+      config.quietBadConfig = false;
+    });
+
+    it('should allow changing the context of `this` in a POJO subscriber', () => {
+      const results: any[] = [];
+
+      const source = new Observable<number>(subscriber => {
+        for (let i = 0; i < 10 && !subscriber.closed; i++) {
+          subscriber.next(i);
+        }
+        subscriber.complete();
+
+        return () => {
+          results.push('teardown');
+        }
+      });
+
+      source.subscribe({
+        next: function (this: any, value) {
+          expect(this.unsubscribe).to.be.a('function');
+          results.push(value);
+          if (value === 3) {
+            this.unsubscribe();
+          }
+        },
+        complete() {
+          throw new Error('should not be called');
+        }
+      });
+
+      expect(results).to.deep.equal([0, 1, 2, 3, 'teardown'])
+    });
+
+    it('should NOT break this context on next methods from unfortunate consumers', () => {
+      // This is a contrived class to illustrate that we can pass another
+      // object that is "observer shaped"
+      class CustomConsumer {
+        valuesProcessed: string[] = [];
+
+        // In here, we access instance state and alter it.
+        next(value: string) {
+          if (value === 'reset') {
+            this.valuesProcessed = [];
+          } else {
+            this.valuesProcessed.push(value);
+          }
+        }
+      };
+
+      const consumer = new CustomConsumer();
+
+      of('old', 'old', 'reset', 'new', 'new').subscribe(consumer);
+
+      expect(consumer.valuesProcessed).not.to.equal(['new', 'new']);
+    });
+  });
 });

src/internal/Subscriber.ts

@@ -162,15 +162,22 @@ export class SafeSubscriber<T> extends Subscriber<T> {
     let next: ((value: T) => void) | undefined;
 
     if (isFunction(observerOrNext)) {
-      next = (<((value: T) => void)> observerOrNext);
+      next = observerOrNext;
     } else if (observerOrNext) {
-      next = (<PartialObserver<T>> observerOrNext).next;
-      error = (<PartialObserver<T>> observerOrNext).error;
-      complete = (<PartialObserver<T>> observerOrNext).complete;
+      next = observerOrNext.next;
+      error = observerOrNext.error;
+      complete = observerOrNext.complete;
       if (observerOrNext !== emptyObserver) {
-        next = next && next.bind(observerOrNext);
-        error = error && error.bind(observerOrNext);
-        complete = complete && complete.bind(observerOrNext);
+        let context: any;
+        if (config.useDeprecatedNextContext) {
+          context = Object.create(observerOrNext);
+          context.unsubscribe = this.unsubscribe.bind(this);
+        } else {
+          context = observerOrNext;
+        }
+        next = next && next.bind(context);
+        error = error && error.bind(context);
+        complete = complete && complete.bind(context);
         if (isSubscription(observerOrNext)) {
           observerOrNext.add(this.unsubscribe.bind(this));
         }
@@ -244,7 +251,7 @@ export class SafeSubscriber<T> extends Subscriber<T> {
 
   private __tryOrUnsub(fn: Function, value?: any): void {
     try {
-      fn.call(this, value);
+      fn(value);
     } catch (err) {
       this.unsubscribe();
       if (config.useDeprecatedSynchronousErrorHandling) {
@@ -260,7 +267,7 @@ export class SafeSubscriber<T> extends Subscriber<T> {
       throw new Error('bad call');
     }
     try {
-      fn.call(this, value);
+      fn(value);
     } catch (err) {
       if (config.useDeprecatedSynchronousErrorHandling) {
         parent.syncErrorValue = err;

src/internal/config.ts

@@ -1,10 +1,18 @@
+/** @prettier */
 let _enable_super_gross_mode_that_will_cause_bad_things = false;
+let _enable_deoptimized_subscriber_creation = false;
 
 /**
  * The global configuration object for RxJS, used to configure things
  * like what Promise contructor should used to create Promises
  */
 export const config = {
+  /**
+   * If true, console logs for deprecation warnings will not be emitted.
+   * @deprecated this will be removed in v8 when all deprecated settings are removed.
+   */
+  quietBadConfig: false,
+
   /**
    * The promise constructor used by default for methods such as
    * {@link toPromise} and {@link forEach}
@@ -28,11 +36,13 @@ export const config = {
    * behaviors described above.
    */
   set useDeprecatedSynchronousErrorHandling(value: boolean) {
-    if (value) {
-      const error = new Error();
-      console.warn('DEPRECATED! RxJS was set to use deprecated synchronous error handling behavior by code at: \n' + error.stack);
-    } else if (_enable_super_gross_mode_that_will_cause_bad_things) {
-      console.log('RxJS: Back to a better error behavior. Thank you. <3');
+    if (!this.quietBadConfig) {
+      if (value) {
+        const error = new Error();
+        console.warn('DEPRECATED! RxJS was set to use deprecated synchronous error handling behavior by code at: \n' + error.stack);
+      } else if (_enable_super_gross_mode_that_will_cause_bad_things) {
+        console.log('RxJS: Back to a better error behavior. Thank you. <3');
+      }
     }
     _enable_super_gross_mode_that_will_cause_bad_things = value;
   },
@@ -45,4 +55,43 @@ export const config = {
   get useDeprecatedSynchronousErrorHandling() {
     return _enable_super_gross_mode_that_will_cause_bad_things;
   },
+
+  /**
+   * If true, enables an as-of-yet undocumented feature from v5: The ability to access
+   * `unsubscribe()` via `this` context in `next` functions created in observers passed
+   * to `subscribe`.
+   *
+   * This is being removed because the performance was severely problematic, and it could also cause
+   * issues when types other than POJOs are passed to subscribe as subscribers, as they will likely have
+   * their `this` context overwritten.
+   *
+   * @deprecated remove in v8. As of version 8, RxJS will no longer support altering the
+   * context of next functions provided as part of an observer to Subscribe. Instead,
+   * you will have access to a subscription or a signal or token that will allow you to do things like
+   * unsubscribe and test closed status.
+   */
+  set useDeprecatedNextContext(value: boolean) {
+    if (!this.quietBadConfig) {
+      if (value) {
+        const error = new Error();
+        console.warn(
+          'DEPRECATED! RxJS was set to use deprecated next context. This will result in deoptimizations when creating any new subscription. \n' +
+            error.stack
+        );
+      } else if (_enable_deoptimized_subscriber_creation) {
+        console.log('RxJS: back to more optimized subscription creation. Thank you. <3');
+      }
+    }
+    _enable_deoptimized_subscriber_creation = value;
+  },
+
+  /**
+   * @deprecated remove in v8. As of version 8, RxJS will no longer support altering the
+   * context of next functions provided as part of an observer to Subscribe. Instead,
+   * you will have access to a subscription or a signal or token that will allow you to do things like
+   * unsubscribe and test closed status.
+   */
+  get useDeprecatedNextContext(): boolean {
+    return _enable_deoptimized_subscriber_creation;
+  },
 };