diff --git a/src/SuspenseSubject.ts b/src/SuspenseSubject.ts index 4514a586..4bb74133 100644 --- a/src/SuspenseSubject.ts +++ b/src/SuspenseSubject.ts @@ -147,31 +147,31 @@ export class SuspenseSubject extends Subject { } _subscribe(subscriber: Subscriber): Subscription { + // `catchError(() => EMPTY)` converts upstream errors into completions, so a direct subscriber completes on error. + return this._attach((value) => subscriber.next(value), () => subscriber.complete(), false); + } + + // For `useObservable`. Completing a `Subscriber` auto-unsubscribes it while the hook is still mounted, + // so this reports every change through `onStoreChange` and never completes. + _subscribeToStore(onStoreChange: () => void): Subscription { + return this._attach(onStoreChange, onStoreChange, true); + } + + private _attach(next: (value: T) => void, complete: () => void, evictErroredEntryOnUnsubscribe: boolean): Subscription { // A mounted subscriber owns the source now; the warm-up release would evict a mounted one-shot entry. this._warmupRelease.unsubscribe(); if (!this._error && !this._isComplete) { this._onReregister(); } this._subscriberCount++; - const innerSubscription = this._innerObservable.subscribe({ - next: (value) => subscriber.next(value), - complete: () => { - if (this._error) { - // `catchError(() => EMPTY)` converts upstream errors into completions. - // Notify `useObservable` (`onStoreChange`) without calling `subscriber.complete()`, - // because `subscriber.complete()` auto-unsubscribes `subscriber` before React unmounts. - subscriber.next(this._value); - } else { - subscriber.complete(); - } - }, - }); + const innerSubscription = this._innerObservable.subscribe({ next, complete }); // Hand off ownership of the shared source from the warm-up subscription to the real subscriber. this._warmupSubscription.unsubscribe(); return new Subscription(() => { innerSubscription.unsubscribe(); this._subscriberCount--; - if (this._subscriberCount === 0 && this._error) { + // A direct subscriber is unsubscribed when it completes on error, while a suspended render may still need the entry. + if (evictErroredEntryOnUnsubscribe && this._subscriberCount === 0 && this._error) { this._onEvict(); } }); diff --git a/src/useObservable.ts b/src/useObservable.ts index 42575419..296ddb76 100644 --- a/src/useObservable.ts +++ b/src/useObservable.ts @@ -99,18 +99,7 @@ export function useObservable(observableId: string, source: Observa } const subscribe = React.useCallback((onStoreChange: () => void) => { - const subscription = observable.subscribe({ - next: () => { - onStoreChange(); - }, - error: (e) => { - onStoreChange(); - throw e; - }, - complete: () => { - onStoreChange(); - }, - }); + const subscription = observable._subscribeToStore(onStoreChange); return () => { subscription.unsubscribe(); diff --git a/test/suspenseSubject-error-branch.test.tsx b/test/suspenseSubject-error-branch.test.tsx index 1119c44f..26365e2c 100644 --- a/test/suspenseSubject-error-branch.test.tsx +++ b/test/suspenseSubject-error-branch.test.tsx @@ -1,15 +1,14 @@ import '@testing-library/jest-dom/extend-expect'; import { cleanup, render, waitFor } from '@testing-library/react'; import * as React from 'react'; -import { Observable, timer, throwError } from 'rxjs'; +import { concat, EmptyError, lastValueFrom, Observable, of, timer, throwError } from 'rxjs'; import { mergeMap } from 'rxjs/operators'; -import { useObservable } from '../src/index'; +import { preloadObservable, useObservable } from '../src/index'; -// Guards the `if (this._error)` branch in SuspenseSubject._subscribe's `complete` handler. -// `catchError(() => EMPTY)` turns a source error into a completion; that branch turns it back into -// a notification instead of completing the React subscriber. Completing instead auto-unsubscribes -// the mounted subscriber, and a mounted hook with `initialData` then resubscribes in a loop. -describe('SuspenseSubject error-to-completion branch', () => { +// `catchError(() => EMPTY)` turns a source error into a completion. `_subscribeToStore` reports it to +// `useObservable` without completing, because completing unsubscribes the still-mounted hook, and its +// unmount then no longer evicts the errored entry. `_subscribe` completes a direct subscriber. +describe('SuspenseSubject source errors', () => { afterEach(() => cleanup()); class Boundary extends React.Component<{ children: React.ReactNode }, { error: Error | null }> { @@ -54,4 +53,71 @@ describe('SuspenseSubject error-to-completion branch', () => { consoleError.mockRestore(); } }, 10000); + + // A direct subscriber of a `preload*` result completes, as in 4.3.0. + it('completes a direct subscriber of a preloaded observable when the source errors', async () => { + const source$ = timer(50).pipe(mergeMap(() => throwError(() => new Error('permission-denied')))); + const preloaded$ = preloadObservable(source$, `error-branch-direct-${Math.random()}`); + + const events: string[] = []; + const observer = { + next: (value: unknown) => events.push(`next:${String(value)}`), + error: () => events.push('error'), + complete: () => events.push('complete'), + }; + + preloaded$.subscribe(observer); + await waitFor(() => expect(events).toEqual(['complete'])); + + // A subscriber that arrives after the error completes too. + events.length = 0; + preloaded$.subscribe(observer); + expect(events).toEqual(['complete']); + await expect(lastValueFrom(preloaded$)).rejects.toBeInstanceOf(EmptyError); + }); + + it('delivers a value to a direct subscriber before completing it on a later error', async () => { + const source$ = concat(of('first'), timer(50).pipe(mergeMap(() => throwError(() => new Error('permission-denied'))))); + const events: string[] = []; + preloadObservable(source$, `error-branch-value-${Math.random()}`).subscribe({ + next: (value) => events.push(`next:${String(value)}`), + error: () => events.push('error'), + complete: () => events.push('complete'), + }); + + await waitFor(() => expect(events).toEqual(['next:first', 'complete'])); + }); + + it('keeps the errored entry for a suspended reader when a direct subscriber completes', async () => { + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {}); + try { + let subscriptions = 0; + const source$ = new Observable((subscriber) => { + subscriptions++; + return timer(50) + .pipe(mergeMap(() => throwError(() => new Error('permission-denied')))) + .subscribe(subscriber); + }); + const observableId = `error-branch-render-${Math.random()}`; + + const Reader = () => { + void lastValueFrom(preloadObservable(source$, observableId)).catch(() => undefined); + const { data } = useObservable(observableId, source$, { suspense: true }); + return
{data}
; + }; + + const { getByTestId } = render( + + }> + + + + ); + + await waitFor(() => expect(getByTestId('err')).toHaveTextContent('permission-denied')); + expect(subscriptions).toBe(1); + } finally { + consoleError.mockRestore(); + } + }); });