fix(error-handling): fix error-handling issues with suspense

src/internal/BehaviorObservable.ts

@@ -1,5 +1,5 @@
 import { Observable } from "rxjs"
 
 export interface BehaviorObservable<T> extends Observable<T> {
-  getValue: () => T
+  getValue: () => any
 }

src/internal/react-enhancer.ts

@@ -1,25 +1,19 @@
-import { Observable, of, Subscription, Subject, race } from "rxjs"
-import { delay, takeUntil, take, filter, tap } from "rxjs/operators"
+import { Observable, of, Subscription } from "rxjs"
+import { delay, take, filter, tap } from "rxjs/operators"
 import { SUSPENSE } from "../SUSPENSE"
 import { BehaviorObservable } from "./BehaviorObservable"
 import { EMPTY_VALUE } from "./empty-value"
 import { noop } from "./noop"
 import { COMPLETE } from "./COMPLETE"
 
-const IS_SSR =
-  typeof window === "undefined" ||
-  typeof window.document === "undefined" ||
-  typeof window.document.createElement === "undefined"
-
 const reactEnhancer = <T>(
   source$: Observable<T>,
   delayTime: number,
 ): BehaviorObservable<T> => {
   let finalizeLastUnsubscription = noop
-  const onSubscribe = new Subject()
+  let latestValue = EMPTY_VALUE
   const result = new Observable<T>(subscriber => {
     let isActive = true
-    let latestValue = EMPTY_VALUE
     const subscription = source$.subscribe({
       next(value) {
         if (
@@ -34,7 +28,6 @@ const reactEnhancer = <T>(
         subscriber.error(e)
       },
     })
-    onSubscribe.next()
     finalizeLastUnsubscription()
     return () => {
       finalizeLastUnsubscription()
@@ -57,34 +50,67 @@ const reactEnhancer = <T>(
   }) as BehaviorObservable<T>
 
   let promise: any
+  let error = EMPTY_VALUE
+  let valueResult: { type: "v"; payload: any } | undefined
   const getValue = () => {
-    try {
-      return (source$ as BehaviorObservable<T>).getValue()
-    } catch (e) {
-      if (promise) throw promise
-
-      if (!IS_SSR && e !== SUSPENSE) {
-        source$
-          .pipe(takeUntil(race(onSubscribe, of(true).pipe(delay(60000)))))
-          .subscribe()
-        try {
-          return (source$ as BehaviorObservable<T>).getValue()
-        } catch (e) {}
-      }
+    if (error !== EMPTY_VALUE) {
+      throw error
+    }
+
+    try {
+      const latest = (source$ as BehaviorObservable<T>).getValue()
+      return valueResult && Object.is(valueResult.payload, latest)
+        ? valueResult
+        : (valueResult = { type: "v", payload: latest })
+    } catch (e) {
+      if (promise) return promise
+
+      let value = EMPTY_VALUE
+      let isSyncError = false
+      promise = {
+        type: "s",
+        payload: reactEnhancer(source$, delayTime)
+          .pipe(
+            filter(value => value !== (SUSPENSE as any)),
+            take(1),
+            tap({
+              next(v) {
+                value = v
+              },
+              error(e) {
+                error = e
+                setTimeout(() => {
+                  error = EMPTY_VALUE
+                }, 200)
+              },
+            }),
+          )
+          .toPromise()
+          .catch(e => {
+            if (isSyncError) return
+            throw e
+          })
+          .finally(() => {
+            promise = undefined
+            valueResult = undefined
+          }),
+      }
+
+      if (value !== EMPTY_VALUE) {
+        latestValue = value
+        return (valueResult = { type: "v", payload: value })
+      }
+
+      if (error !== EMPTY_VALUE) {
+        isSyncError = true
+        throw error
+      }
+
+      return promise
     }
-    promise = source$
-      .pipe(
-        filter(value => value !== (SUSPENSE as any)),
-        take(1),
-        tap(() => {
-          promise = undefined
-        }),
-      )
-      .toPromise()
-    throw promise
   }
 
-  result.getValue = getValue as () => T
+  result.getValue = getValue as () => T | Promise<T>
   return result
 }
 

src/internal/share-latest.ts

@@ -51,8 +51,12 @@ const shareLatest = <T>(
       innerSub.unsubscribe()
       if (refCount === 0) {
         currentValue = EMPTY_VALUE
+        if (subject !== undefined) {
+          teardown()
+        } else {
+          setTimeout(teardown, 200)
+        }
         subject = undefined
-        teardown()
         if (subscription) {
           subscription.unsubscribe()
           subscription = undefined

src/internal/useObservable.ts

@@ -2,23 +2,23 @@ import { useEffect, useReducer } from "react"
 import { BehaviorObservable } from "./BehaviorObservable"
 import { SUSPENSE } from "../SUSPENSE"
 
+const ERROR: "e" = "e"
+const VALUE: "v" = "v"
+const SUSP: "s" = "s"
+type Action = "e" | "v" | "s"
+
 const reducer = (
-  _: any,
-  { type, payload }: { type: "next" | "error"; payload: any },
+  _: { type: Action; payload: any },
+  action: { type: Action; payload: any },
 ) => {
-  if (type === "error") {
-    throw payload
-  }
-  return payload
-}
-const init = (source$: BehaviorObservable<any>) => {
-  try {
-    return source$.getValue()
-  } catch (e) {
-    return SUSPENSE
+  if (action.type === ERROR) {
+    throw action.payload
   }
+  return action
 }
 
+const init = (source$: BehaviorObservable<any>) => source$.getValue()
+
 export const useObservable = <O>(
   source$: BehaviorObservable<O>,
 ): Exclude<O, typeof SUSPENSE> => {
@@ -26,30 +26,32 @@ export const useObservable = <O>(
 
   useEffect(() => {
     try {
-      dispatch({
-        type: "next",
-        payload: source$.getValue(),
-      })
+      dispatch(source$.getValue())
     } catch (e) {
-      dispatch({
-        type: "next",
-        payload: SUSPENSE,
-      })
+      return dispatch({ type: ERROR, payload: e })
     }
     const subscription = source$.subscribe(
-      value =>
-        dispatch({
-          type: "next",
-          payload: value,
-        }),
+      value => {
+        if ((value as any) === SUSPENSE) {
+          dispatch(source$.getValue())
+        } else {
+          dispatch({
+            type: VALUE,
+            payload: value,
+          })
+        }
+      },
       error =>
         dispatch({
-          type: "error",
+          type: ERROR,
           payload: error,
         }),
     )
     return () => subscription.unsubscribe()
   }, [source$])
 
-  return state !== (SUSPENSE as any) ? (state as any) : source$.getValue()
+  if (state.type === SUSP) {
+    throw state.payload
+  }
+  return state.payload
 }

test/connectFactoryObservable.test.tsx

@@ -1,6 +1,15 @@
 import { connectFactoryObservable } from "../src"
 import { TestErrorBoundary } from "../test/TestErrorBoundary"
-import { from, of, defer, concat, BehaviorSubject, throwError } from "rxjs"
+import {
+  from,
+  of,
+  defer,
+  concat,
+  BehaviorSubject,
+  throwError,
+  Observable,
+  Subject,
+} from "rxjs"
 import { renderHook, act as actHook } from "@testing-library/react-hooks"
 import { switchMap, delay } from "rxjs/operators"
 import { FC, Suspense, useState } from "react"
@@ -223,6 +232,119 @@ describe("connectFactoryObservable", () => {
       )
     })
 
+    it("allows sync errors to be caught in error boundaries with suspense", () => {
+      const errStream = new Observable(observer =>
+        observer.error("controlled error"),
+      )
+      const [useError] = connectFactoryObservable((_: string) => errStream)
+
+      const ErrorComponent = () => {
+        const value = useError("foo")
+
+        return <>{value}</>
+      }
+
+      const errorCallback = jest.fn()
+      const { unmount } = render(
+        <TestErrorBoundary onError={errorCallback}>
+          <Suspense fallback={<div>Loading...</div>}>
+            <ErrorComponent />
+          </Suspense>
+        </TestErrorBoundary>,
+      )
+
+      expect(errorCallback).toHaveBeenCalledWith(
+        "controlled error",
+        expect.any(Object),
+      )
+      unmount()
+    })
+
+    it("allows async errors to be caught in error boundaries with suspense", async () => {
+      const errStream = new Subject()
+      const [useError] = connectFactoryObservable((_: string) => errStream)
+
+      const ErrorComponent = () => {
+        const value = useError("foo")
+
+        return <>{value}</>
+      }
+
+      const errorCallback = jest.fn()
+      const { unmount } = render(
+        <TestErrorBoundary onError={errorCallback}>
+          <Suspense fallback={<div>Loading...</div>}>
+            <ErrorComponent />
+          </Suspense>
+        </TestErrorBoundary>,
+      )
+
+      await componentAct(async () => {
+        errStream.error("controlled error")
+        await wait(0)
+      })
+
+      expect(errorCallback).toHaveBeenCalledWith(
+        "controlled error",
+        expect.any(Object),
+      )
+      unmount()
+    })
+
+    it(
+      "the errror-boundary can capture errors that are produced when changing the " +
+        "key of the hook to an observable that throws synchronously",
+      async () => {
+        const normal$ = new Subject<string>()
+        const errored$ = new Observable<string>(observer => {
+          observer.error("controlled error")
+        })
+
+        const [useOkKo] = connectFactoryObservable((ok: boolean) =>
+          ok ? normal$ : errored$,
+        )
+
+        const ErrorComponent = () => {
+          const [ok, setOk] = useState(true)
+          const value = useOkKo(ok)
+
+          return <span onClick={() => setOk(false)}>{value}</span>
+        }
+
+        const errorCallback = jest.fn()
+        const { unmount } = render(
+          <TestErrorBoundary onError={errorCallback}>
+            <Suspense fallback={<div>Loading...</div>}>
+              <ErrorComponent />
+            </Suspense>
+          </TestErrorBoundary>,
+        )
+
+        expect(screen.queryByText("ALL GOOD")).toBeNull()
+        expect(screen.queryByText("Loading...")).not.toBeNull()
+
+        await componentAct(async () => {
+          normal$.next("ALL GOOD")
+          await wait(50)
+        })
+
+        expect(screen.queryByText("ALL GOOD")).not.toBeNull()
+        expect(screen.queryByText("Loading...")).toBeNull()
+        expect(errorCallback).not.toHaveBeenCalled()
+
+        componentAct(() => {
+          fireEvent.click(screen.getByText(/GOOD/i))
+        })
+
+        expect(errorCallback).toHaveBeenCalledWith(
+          "controlled error",
+          expect.any(Object),
+        )
+
+        unmount()
+      },
+    )
+
     it("doesn't throw errors on components that will get unmounted on the next cycle", () => {
       const valueStream = new BehaviorSubject(1)
       const [useValue, value$] = connectFactoryObservable(() => valueStream)
@@ -258,6 +380,7 @@ describe("connectFactoryObservable", () => {
       expect(errorCallback).not.toHaveBeenCalled()
     })
   })
+
   describe("observable", () => {
     it("it completes when the source observable completes, regardless of mounted componentes being subscribed to the source", async () => {
       let diff = -1

test/connectObservable.test.tsx

@@ -6,7 +6,15 @@ import {
 } from "@testing-library/react"
 import { act, renderHook } from "@testing-library/react-hooks"
 import React, { Suspense, useEffect, FC } from "react"
-import { BehaviorSubject, defer, from, of, Subject, throwError } from "rxjs"
+import {
+  BehaviorSubject,
+  defer,
+  from,
+  of,
+  Subject,
+  throwError,
+  Observable,
+} from "rxjs"
 import { delay, scan, startWith, map, switchMap } from "rxjs/operators"
 import { connectObservable, SUSPENSE } from "../src"
 import { TestErrorBoundary } from "../test/TestErrorBoundary"
@@ -245,7 +253,6 @@ describe("connectObservable", () => {
 
     const ErrorComponent = () => {
       const value = useError()
-
       return <>{value}</>
     }
 
@@ -266,18 +273,19 @@ describe("connectObservable", () => {
     )
   })
 
-  it("allows errors to be caught in error boundaries with suspense", () => {
-    const errStream = new Subject()
+  it("allows sync errors to be caught in error boundaries with suspense", () => {
+    const errStream = new Observable(observer =>
+      observer.error("controlled error"),
+    )
     const [useError] = connectObservable(errStream)
 
     const ErrorComponent = () => {
       const value = useError()
-
       return <>{value}</>
     }
 
     const errorCallback = jest.fn()
-    render(
+    const { unmount } = render(
       <TestErrorBoundary onError={errorCallback}>
         <Suspense fallback={<div>Loading...</div>}>
           <ErrorComponent />
@@ -285,14 +293,105 @@ describe("connectObservable", () => {
       </TestErrorBoundary>,
     )
 
-    componentAct(() => {
+    expect(errorCallback).toHaveBeenCalledWith(
+      "controlled error",
+      expect.any(Object),
+    )
+    unmount()
+  })
+
+  it("allows async errors to be caught in error boundaries with suspense", async () => {
+    const errStream = new Subject()
+    const [useError] = connectObservable(errStream)
+
+    const ErrorComponent = () => {
+      const value = useError()
+      return <>{value}</>
+    }
+
+    const errorCallback = jest.fn()
+    const { unmount } = render(
+      <TestErrorBoundary onError={errorCallback}>
+        <Suspense fallback={<div>Loading...</div>}>
+          <ErrorComponent />
+        </Suspense>
+      </TestErrorBoundary>,
+    )
+
+    await componentAct(async () => {
       errStream.error("controlled error")
+      await wait(0)
     })
 
     expect(errorCallback).toHaveBeenCalledWith(
       "controlled error",
       expect.any(Object),
     )
+    unmount()
+  })
+
+  it("allows to retry the errored observable after a grace period of time", async () => {
+    let errStream = new Subject<string>()
+    const [useError] = connectObservable(
+      defer(() => {
+        return (errStream = new Subject<string>())
+      }),
+    )
+
+    const ErrorComponent = () => {
+      const value = useError()
+      return <>{value}</>
+    }
+
+    const errorCallback = jest.fn()
+    const { unmount } = render(
+      <TestErrorBoundary onError={errorCallback}>
+        <Suspense fallback={<div>Loading...</div>}>
+          <ErrorComponent />
+        </Suspense>
+      </TestErrorBoundary>,
+    )
+
+    expect(screen.queryByText("Loading...")).not.toBeNull()
+    expect(screen.queryByText("ALL GOOD")).toBeNull()
+
+    await componentAct(async () => {
+      errStream.error("controlled error")
+      await wait(0)
+    })
+
+    expect(screen.queryByText("Loading...")).toBeNull()
+    expect(screen.queryByText("ALL GOOD")).toBeNull()
+    expect(errorCallback).toHaveBeenCalledWith(
+      "controlled error",
+      expect.any(Object),
+    )
+    unmount()
+
+    errorCallback.mockReset()
+    await componentAct(async () => {
+      await wait(250)
+    })
+
+    render(
+      <TestErrorBoundary onError={errorCallback}>
+        <Suspense fallback={<div>Loading...</div>}>
+          <ErrorComponent />
+        </Suspense>
+      </TestErrorBoundary>,
+    )
+    expect(screen.queryByText("Loading...")).not.toBeNull()
+
+    await componentAct(async () => {
+      errStream.next("ALL GOOD")
+      await wait(50)
+    })
+
+    expect(errorCallback).not.toHaveBeenCalledWith(
+      "controlled error",
+      expect.any(Object),
+    )
+    expect(screen.queryByText("ALL GOOD")).not.toBeNull()
   })
 
   it("doesn't throw errors on components that will get unmounted on the next cycle", () => {