fix(zone.js): zone patch rxjs should return null _unsubscribe correctly. (#37091)

Close #31684.

In some rxjs operator, such as `retryWhen`, rxjs internally will set
`Subscription._unsubscribe` method to null, and the current zone.js monkey patch
didn't handle this case correctly, even rxjs set _unsubscribe to null, zone.js
still return a function by finding the prototype chain.

This PR fix this issue and the following test will pass.

```
const errorGenerator = () => {
  return throwError(new Error('error emit'));
};

const genericRetryStrategy = (finalizer: () => void) => (attempts: Observable<any>) =>
    attempts.pipe(
      mergeMap((error, i) => {
        const retryAttempt = i + 1;
        if (retryAttempt > 3) {
          return throwError(error);
        }
        return timer(retryAttempt * 1);
      }),
      finalize(() => finalizer()));

errorGenerator()
  .pipe(
    retryWhen(genericRetryStrategy(() => {
      expect(log.length).toBe(3);
      done();
    })),
    catchError(error => of(error)))
  .subscribe()
```

PR Close #37091

packages/zone.js/lib/rxjs/rxjs.ts

@@ -115,7 +115,7 @@ type ZoneSubscriberContext = {
       _zoneUnsubscribe: {value: null, writable: true, configurable: true},
       _unsubscribe: {
         get: function(this: Subscription) {
-          if ((this as any)._zoneUnsubscribe) {
+          if ((this as any)._zoneUnsubscribe || (this as any)._zoneUnsubscribeCleared) {
             return (this as any)._zoneUnsubscribe;
           }
           const proto = Object.getPrototypeOf(this);
@@ -125,7 +125,13 @@ type ZoneSubscriberContext = {
           (this as any)._zone = Zone.current;
           if (!unsubscribe) {
             (this as any)._zoneUnsubscribe = unsubscribe;
+            // In some operator such as `retryWhen`, the _unsubscribe
+            // method will be set to null, so we need to set another flag
+            // to tell that we should return null instead of finding
+            // in the prototype chain.
+            (this as any)._zoneUnsubscribeCleared = true;
           } else {
+            (this as any)._zoneUnsubscribeCleared = false;
             (this as any)._zoneUnsubscribe = function() {
               if (this._zone && this._zone !== Zone.current) {
                 return this._zone.run(unsubscribe, this, arguments);

packages/zone.js/test/rxjs/rxjs.retry.spec.ts

@@ -0,0 +1,41 @@
+/**
+ * @license
+ * Copyright Google Inc. All Rights Reserved.
+ *
+ * Use of this source code is governed by an MIT-style license that can be
+ * found in the LICENSE file at https://angular.io/license
+ */
+import {Observable, of, throwError, timer} from 'rxjs';
+import {catchError, finalize, mergeMap, retryWhen} from 'rxjs/operators';
+
+describe('retryWhen', () => {
+  let log: any[];
+  const genericRetryStrategy = (finalizer: () => void) => (attempts: Observable<any>) =>
+      attempts.pipe(
+          mergeMap((error, i) => {
+            const retryAttempt = i + 1;
+            if (retryAttempt > 3) {
+              return throwError(error);
+            }
+            log.push(error);
+            return timer(retryAttempt * 1);
+          }),
+          finalize(() => finalizer()));
+
+  const errorGenerator = () => {
+    return throwError(new Error('error emit'));
+  };
+  beforeEach(() => {
+    log = [];
+  });
+
+  it('should retry max 3 times',
+     (done: DoneFn) => {errorGenerator()
+                            .pipe(
+                                retryWhen(genericRetryStrategy(() => {
+                                  expect(log.length).toBe(3);
+                                  done();
+                                })),
+                                catchError(error => of(error)))
+                            .subscribe()});
+});

packages/zone.js/test/rxjs/rxjs.spec.ts

@@ -27,6 +27,7 @@ import './rxjs.merge.spec';
 import './rxjs.never.spec';
 import './rxjs.of.spec';
 import './rxjs.range.spec';
+import './rxjs.retry.spec';
 import './rxjs.throw.spec';
 import './rxjs.timer.spec';
 import './rxjs.zip.spec';