How to receive values from observable after an error

[jorroll]

In my application, I have CURRENT_USER$ and ASSERT_CURRENT_USER$ observables. The CURRENT_USER$ observable will return a user document if the user is signed in or null if they are not. ASSERT_CURRENT_USER$ will only ever return a user document.

I was defining ASSERT_CURRENT_USER$ like so

export const ASSERT_CURRENT_USER$ = CURRENT_USER$.pipe(
  switchMap((user) =>
    !user ? NEVER : of(user),
  ),
);

The goal was to simplify queries that depend on the current user. By using the ASSERT_CURRENT_USER$ observable, the query wouldn't need to worry about handling the case with the user isn't signed in. The problem with this is that, since NEVER doesn't emit, when a user signs out their old queries stick around until a new user signs in (potentially causing http errors if the queries depend on an authenticated user).

Thinking I was being clever, I decided to update the code to the following

export const ASSERT_CURRENT_USER$ = CURRENT_USER$.pipe(
  switchMap((user) =>
    !user ? throwError(() => new NoCurrentUserError()) : of(user),
  ),
);

I then added a catchNoCurrentUserError operator which I could use downstream to handle the scenario when a user isn't signed in.

export function catchNoCurrentUserError<I, T>(defaultValue: () => T) {
  return pipe(
    catchError<I, ObservableInput<T>>((err) => {
      if (err instanceof NoCurrentUserError) {
        return of(defaultValue());
      }

      throw err;
    }),
  );
}

This greatly simplifies the code for some complex observables, since most of the code doesn't need to worry about handling a null current user.

The problem is that after the error condition of the ASSERT_CURRENT_USER$ observable is triggered, it never emits again. This is surprising, since the CURRENT_USER$ observable hasn't errored and since I'm handling the error downstream.

Is there any way to accomplish this goal?

Separately, I'm guessing the current behavior in RXJS was intentional but it's not obvious to me why it's desirable?

Thanks!

[voliva]

how is CURRENT_USER$ defined?

In RxJS by default observables are lazy and unicast, meaning each subscriber will have their own side effect / state. When an observable throws an error, this error gets propagated all the way down to all subscribers, and all of those subscriptions are closed. As I understand, errors in RxJS were designed in a similar way of a regular throw new Error()... that process is killed. You can catch the error and recover from it, but whatever happened there it's now dead.

If CURRENT_USER$ is not multicast, then even though it didn't throw an error, its only subscription did, which caused an unsubscription throughout the whole chain.

A posible "simple" solution is to make CURRENT_USER$ multicast and tell it to not reset when everyone unsubscribes (e.g. share({ resetOnRefCountZero: false })), then on each of the leaves do some sort of retry or retryWhen. Or if you want to keep your operator with a defaultValue, I think something like this could work:

export function catchNoCurrentUserError<I, T>(defaultValue: () => T) {
  return (source$: Observable<I>) => {
    function handleError(err: any) {
      if (err instanceof NoCurrentUserError) {
        return source$.pipe(
          startWith(defaultValue()),
          catchError(handleError)
        );
      }

      throw err;
    }

    return source$.pipe(
      catchError(handleError)
    );
  };
}

Essentially, when there's an error of type NoCurrentUserError this will act as a retry but also spitting out the defaultValue. Everything will unsubscribe (because the error happened), but we're resubscribing immediately. Because we have configured CURRENT_USER$ to not reset when refcount reaches zero, it will be at the same point it was left before.

On React-RxJS we're currently working on a feature that will essentially solve this. It's still experimental and we're just trying it out, but essentially it will solve your problem by looking like this (again, experimental, API will change when we release... we have a few ideas that will help with versatility and type defs):

// Assuming type of CURRENT_USER$ is `User | null`
export const ASSERT_CURRENT_USER$ = CURRENT_USER$.pipe(
  sinkEffects(null) // This throws every null as an error, the result is Observable<User>
);

ASSERT_CURRENT_USER$.pipe(
  ... // Some stateful stuff,
  liftEffects() // Will resurface the `null` as actual value
)

the cool thing about this is that CURRENT_USER$ doesn't need to be multicast, the subscription will never die unless the unsubscription happens for another reason than the effect. Everything is still getting propagated as an error, so anything between sinkEffects and liftEffects will reset when a null is found.

By now I would suggest to stick with the simple solution of share + retryWhen.

[jorroll]

THIS IS GENIUS!!!! 🎉

Using your very helpful advice, I was able to get things to work. I had to slightly tweak your suggestion because your example code immediately resubscribes to ASSERT_CURRENT_USER$ yet I only want it to resubscribe when there is a current user available again. I updated the catchNoCurrentUserError operator to only resubscribe when a user is available.

Working code:

export function catchNoCurrentUserError<I, T>(defaultValue: () => T) {
  return (source$: Observable<I>) => {
    function handleError(err: unknown): Observable<I | T> {
      if (err instanceof NoCurrentUserError) {
        return CURRENT_USER$.pipe(
          switchMap((currentUser) => (!currentUser ? NEVER : source$)),
          startWith(defaultValue),
          catchError(handleError),
        );
      }

      throw err;
    }

    return source$.pipe(catchError(handleError));
  };
}

On React-RxJS we're currently working on a feature that will essentially solve this. It's rx-state/core#7 and we're just trying it out, but essentially it will solve your problem by looking like this

Sounds like I'm lucky you saw my question!! Seems like you were unusually well positioned to help me. Thank you so much for taking the time to answer! I had started the process of updating my codebase with a much, much uglier and confusing solution. Just implemented this change and it's working beautifully!

[jorroll]

Actually, I improved the catchNoCurrentUserError operator some more with the option to not provide any default value. Here, if you don't provide a defaultValue factory function then catchNoCurrentUserError will unsubscribe from the previous sources and then emit nothing until the user is logged in again.

export function catchNoCurrentUserError<I>(): (source: Observable<I>) => Observable<I>;
export function catchNoCurrentUserError<I, T>(defaultValue: () => T): (source: Observable<I>) => Observable<I | T>;
export function catchNoCurrentUserError<I, T>(defaultValue?: () => T) {
  return (source$: Observable<I>) => {
    function handleError(err: unknown): Observable<I | T> {
      if (err instanceof NoCurrentUserError) {
        return defaultValue
          ? CURRENT_USER$.pipe(
              switchMap((currentUser) => (!currentUser ? NEVER : source$)),
              startWith(defaultValue),
              catchError(handleError),
            )
          : CURRENT_USER$.pipe(
              switchMap((currentUser) => (!currentUser ? NEVER : source$)),
              catchError(handleError),
            );
      }

      throw err;
    }

    return source$.pipe(catchError(handleError));
  };
}