fix: Subscribe with async errors

Co-authored-by: Victor Oliva <olivarra1@gmail.com>

Author: josepot

packages/core/src/Subscribe.test.tsx

@@ -1,11 +1,13 @@
 import { state } from "@rxstate/core"
-import { render, screen } from "@testing-library/react"
+import { render, screen, act } from "@testing-library/react"
 import React, { StrictMode, useState, useEffect } from "react"
 import { defer, EMPTY, NEVER, Observable, of, startWith } from "rxjs"
 import { bind, RemoveSubscribe, Subscribe as OriginalSubscribe } from "./"
 import { TestErrorBoundary } from "./test-helpers/TestErrorBoundary"
 import { useStateObservable } from "./useStateObservable"
 
+const wait = (ms: number) => new Promise((res) => setTimeout(res, ms))
+
 const Subscribe = (props: any) => {
   return (
     <StrictMode>
@@ -303,6 +305,38 @@ describe("Subscribe", () => {
 
       expect(hasError).toBe(false)
     })
+
+    it("allows async errors to be caught in error boundaries with suspense, without using source$", async () => {
+      const [useError] = bind(
+        new Observable((obs) => {
+          setTimeout(() => obs.error("controlled error"), 10)
+        }),
+      )
+
+      const ErrorComponent = () => {
+        const value = useError()
+        return <>{value}</>
+      }
+
+      const errorCallback = jest.fn()
+      const { unmount } = render(
+        <TestErrorBoundary onError={errorCallback}>
+          <Subscribe fallback={<div>Loading...</div>}>
+            <ErrorComponent />
+          </Subscribe>
+        </TestErrorBoundary>,
+      )
+
+      await act(async () => {
+        await wait(100)
+      })
+
+      expect(errorCallback).toHaveBeenCalledWith(
+        "controlled error",
+        expect.any(Object),
+      )
+      unmount()
+    })
   })
 })
 

packages/core/src/Subscribe.tsx

@@ -8,8 +8,11 @@ import React, {
   useContext,
 } from "react"
 import { Observable, Subscription } from "rxjs"
+import type { StateObservable } from "@rxstate/core"
 
-const SubscriptionContext = createContext<Subscription | null>(null)
+const SubscriptionContext = createContext<
+  ((src: StateObservable<any>) => void) | null
+>(null)
 const { Provider } = SubscriptionContext
 export const useSubscription = () => useContext(SubscriptionContext)
 
@@ -41,9 +44,27 @@ export const Subscribe: React.FC<{
   source$?: Observable<any>
   fallback?: NonNullable<ReactNode> | null
 }> = ({ source$, children, fallback }) => {
-  const subscriptionRef = useRef<Subscription>()
+  const subscriptionRef = useRef<{
+    s: Subscription
+    u: (source: StateObservable<any>) => void
+  }>()
 
-  if (!subscriptionRef.current) subscriptionRef.current = new Subscription()
+  if (!subscriptionRef.current) {
+    const s = new Subscription()
+    subscriptionRef.current = {
+      s,
+      u: (src) => {
+        s.add(
+          src.subscribe({
+            error: (e) =>
+              setSubscribedSource(() => {
+                throw e
+              }),
+          }),
+        )
+      },
+    }
+  }
 
   const [subscribedSource, setSubscribedSource] = useState<
     Observable<any> | null | undefined
@@ -77,14 +98,14 @@ export const Subscribe: React.FC<{
 
   useEffect(() => {
     return () => {
-      subscriptionRef.current?.unsubscribe()
+      subscriptionRef.current?.s.unsubscribe()
       subscriptionRef.current = undefined
     }
   }, [])
 
   const actualChildren =
     subscribedSource === source$ ? (
-      <Provider value={subscriptionRef.current!}>{children}</Provider>
+      <Provider value={subscriptionRef.current!.u}>{children}</Provider>
     ) : fallback === undefined ? null : (
       <Throw />
     )

packages/core/src/useStateObservable.ts

@@ -1,7 +1,6 @@
 import { useRef, useState } from "react"
 import { SUSPENSE } from "./SUSPENSE"
 import { DefaultedStateObservable, StateObservable } from "@rxstate/core"
-import { EMPTY_VALUE } from "./internal/empty-value"
 import useSyncExternalStore from "./internal/useSyncExternalStore"
 import { useSubscription } from "./Subscribe"
 
@@ -31,20 +30,10 @@ export const useStateObservable = <O>(
 
     const gv: <T>() => Exclude<T, typeof SUSPENSE> = () => {
       const src = callbackRef.current!.source$ as DefaultedStateObservable<O>
-
-      if (src.getRefCount() > 0 || src.getDefaultValue) return getValue(src)
-
-      if (!subscription) throw new Error("Missing Subscribe!")
-
-      let error = EMPTY_VALUE
-      subscription.add(
-        src.subscribe({
-          error: (e) => {
-            error = e
-          },
-        }),
-      )
-      if (error !== EMPTY_VALUE) throw error
+      if (!src.getRefCount() && !src.getDefaultValue) {
+        if (!subscription) throw new Error("Missing Subscribe!")
+        subscription(src)
+      }
       return getValue(src)
     }