observer.error emit errors multiple times

[bgfist]

Bug Report

Current Behavior
observer.error can emit errors multiple times

Reproduction

• REPL or Repo link:

https://stackblitz.com/edit/8blr49

import { of, interval, throwError, EMPTY, Observable } from "rxjs";
import {
  takeUntil,
  switchMapTo,
  timeout,
  catchError,
  tap,
  take,
  timeInterval
} from "rxjs/operators";

const errors = new Observable(observer => {
  setTimeout(() => {
    console.log("first");
    observer.error(1);
  }, 1000);
  setTimeout(() => {
    console.log("second");
    observer.error(2);
  }, 4000);

  return () => {
    console.log("unsubscribe");
  };
});

errors
  .pipe(
    catchError(error => {
      console.log("error", error);
      return interval(error * 1000).pipe(timeInterval());
    }),
    tap(x => {
      console.log(222, x);
    })
  )
  .subscribe(x => {}, console.error);

this prints:

first
error 1
unsubscribe
222 TimeInterval {value: 0, interval: 1002}
222 TimeInterval {value: 1, interval: 1002}
second
error 2
222 TimeInterval {value: 0, interval: 2001}
222 TimeInterval {value: 1, interval: 2000}

Expected behavior

I suppose the catchError callback should only be called once. But it's executed twice, and even unsubscribe has been called.

Environment

• Runtime: [e.g. Node v${x}, Chrome v${x}]
• RxJS version: 6.6.2
• (If bug is related) Loader, build configuration: [e.g webpack, angular-cli version, config]

Possible Solution
the next、error、complete should be wrapped call , which can obey to Observable Grammar or Contract

Additional context/Screenshots
Add any other context about the problem here. If applicable, add screenshots to help explain.

[josepot]

Fascinating. I've looked a little bit into this and I think that this is likely to be an issue with the current catchError operator.

These are the findings that I made in the few mins that I've been looking into this:

• In the example provided by @bgfist if we log the closed property of the observer in the following lines:

const errors = new Observable(observer => {
  setTimeout(() => {
    console.log("first");
    observer.error(1);
    console.log('after 1st error', observer.closed);
  }, 1000);
  setTimeout(() => {
    console.log('before 2nd error', observer.closed)
    observer.error(2);
  }, 4000);

  return () => {
    console.log("unsubscribe", observer.closed);
  };
});

We can appreciate that right on the moment of unsubscribing, the observer is closed... However, in the next two logs the observer is not closed 😮 .

• I wrote a customCatchOperator, like this:

const customCatchError = <In, Out>(fn: (error: any) => Observable<Out>) => (
  source$: Observable<In>,
): Observable<In | Out> =>
  new Observable<In | Out>((observer) => {
    let subscription = source$.subscribe(
      (val) => observer.next(val),
      (e) => {
        subscription = fn(e).subscribe(observer)
      },
      () => {
        observer.complete()
      },
    )

    return () => {
      subscription.unsubscribe()
    }
  })

And with that custom version the problem goes away.

• This issue also affects v7.0.0-beta.4

I haven't looked into the implementation of the catchError operator yet.

[josepot]

I went to write a failing test for this issue on master and the test wasn't failing... Using git bisect I found that this issue got fixed on #5627 , it's just that those changes haven't been released yet.

[bgfist]

I went to write a failing test for this issue on master and the test wasn't failing... Using git bisect I found that this issue got fixed on #5627 , it's just that those changes haven't been released yet.

thanks anyway.

[josepot]

thanks anyway.

Np! FWIW this is the test that I wrote (the one that was failing before #5627 )

it('should leave the subscriber of the source closed after catching the error', (done: MochaDone) => {
  let isClosed = false

  const error$ = new Observable(subscriber => {
    subscriber.error();
    isClosed = subscriber.closed
  });

  error$.pipe(catchError(() => EMPTY)).subscribe(
    () => {
      // noop
    },
    () => done(new Error('should not be called')),
    () => {
      expect(isClosed).to.equal(true);
      done();
    });
})