fix: useStateObservable (#253)

Author: josepot

packages/core/src/Subscribe.test.tsx

@@ -1,6 +1,8 @@
-import { render } from "@testing-library/react"
+import { state } from "@josepot/rxjs-state"
+import { render, screen } from "@testing-library/react"
 import React, { StrictMode, useState } from "react"
-import { defer, Observable, of } from "rxjs"
+import { defer, EMPTY, Observable, of, startWith } from "rxjs"
+import { useStateObservable } from "./useStateObservable"
 import { bind, Subscribe as OriginalSubscribe } from "./"
 
 const Subscribe = (props: any) => {
@@ -13,6 +15,24 @@ const Subscribe = (props: any) => {
 
 describe("Subscribe", () => {
   describe("Subscribe with source$", () => {
+    it("renders the sync emitted value on a StateObservable without default value", () => {
+      const test$ = state(EMPTY.pipe(startWith("there!")))
+      const useTest = () => useStateObservable(test$)
+
+      const Test: React.FC = () => <>Hello {useTest()}</>
+
+      const TestSubscribe: React.FC = () => (
+        <Subscribe>
+          <Test />
+        </Subscribe>
+      )
+
+      const { unmount } = render(<TestSubscribe />)
+
+      expect(screen.queryByText("Hello there!")).not.toBeNull()
+
+      unmount()
+    })
     it("subscribes to the provided observable and remains subscribed until it's unmounted", () => {
       let nSubscriptions = 0
       const [useNumber, number$] = bind(

packages/core/src/Subscribe.tsx

@@ -9,7 +9,7 @@ import React, {
 } from "react"
 import { Observable, Subscription } from "rxjs"
 
-const SubscriptionContext = createContext<Subscription>(null as any)
+const SubscriptionContext = createContext<Subscription | null>(null)
 const { Provider } = SubscriptionContext
 export const useSubscription = () => useContext(SubscriptionContext)
 

packages/core/src/bind/connectFactoryObservable.ts

@@ -1,9 +1,8 @@
-import { noop, Observable } from "rxjs"
-import { useObservable } from "../internal/useObservable"
+import { Observable } from "rxjs"
 import { SUSPENSE } from "../SUSPENSE"
 import { EMPTY_VALUE } from "../internal/empty-value"
-import { useSubscription } from "../Subscribe"
 import { state, StateObservable } from "@josepot/rxjs-state"
+import { useStateObservable } from "../useStateObservable"
 
 /**
  * Accepts: A factory function that returns an Observable.
@@ -39,9 +38,5 @@ export default function connectFactoryObservable<A extends [], O>(
       : [getObservable, defaultValue]
 
   const obs = state(...(args as [(...args: A) => Observable<O>]))
-  const useSub = defaultValue === EMPTY_VALUE ? useSubscription : noop
-  return [
-    (...input: A) => useObservable(obs(...input) as any, useSub() as any),
-    obs,
-  ]
+  return [(...input: A) => useStateObservable(obs(...input)), obs]
 }

packages/core/src/bind/connectObservable.ts

@@ -1,7 +1,6 @@
 import { EMPTY_VALUE } from "../internal/empty-value"
-import { noop, Observable } from "rxjs"
-import { useSubscription } from "../Subscribe"
-import { useObservable } from "../internal/useObservable"
+import { Observable } from "rxjs"
+import { useStateObservable } from "../useStateObservable"
 import { state } from "@josepot/rxjs-state"
 
 /**
@@ -23,13 +22,11 @@ export default function connectObservable<T>(
   observable: Observable<T>,
   defaultValue: T,
 ) {
-  const useSub = defaultValue === EMPTY_VALUE ? useSubscription : noop
   const sharedObservable$ =
     defaultValue === EMPTY_VALUE
       ? state(observable)
       : state(observable, defaultValue)
 
-  const useStaticObservable = () =>
-    useObservable(sharedObservable$ as any, useSub() as any)
+  const useStaticObservable = () => useStateObservable(sharedObservable$ as any)
   return [useStaticObservable, sharedObservable$] as const
 }

packages/core/src/internal/useObservable.ts

@@ -1,7 +1,73 @@
-import { useObservable } from "./internal/useObservable"
-import { SUSPENSE } from "./SUSPENSE"
-import type { StateObservable } from "@josepot/rxjs-state"
+import { useRef, useState } from "react"
+import { SUSPENSE, filterOutSuspense } from "./SUSPENSE"
+import { DefaultedStateObservable, StateObservable } from "@josepot/rxjs-state"
+import { EMPTY_VALUE } from "./internal/empty-value"
+import useSyncExternalStore from "./internal/useSyncExternalStore"
+import { useSubscription } from "./Subscribe"
 
-export const useStateObservable = <T>(
-  stateObservable: StateObservable<T>,
-): Exclude<T, typeof SUSPENSE> => useObservable(stateObservable)
+type VoidCb = () => void
+
+interface Ref<T> {
+  source$: StateObservable<T>
+  args: [(cb: VoidCb) => VoidCb, () => Exclude<T, typeof SUSPENSE>]
+}
+
+export const useStateObservable = <O>(
+  source$: StateObservable<O>,
+): Exclude<O, typeof SUSPENSE> => {
+  const subscription = useSubscription()
+  const [, setError] = useState()
+  const callbackRef = useRef<Ref<O>>()
+
+  if (!callbackRef.current) {
+    const getValue = (src: StateObservable<O>) => {
+      const result = src.getValue(filterOutSuspense)
+      if (result instanceof Promise) throw result
+      return result as any
+    }
+
+    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
+      return getValue(src)
+    }
+
+    callbackRef.current = {
+      source$: null as any,
+      args: [, gv] as any,
+    }
+  }
+
+  const ref = callbackRef.current
+  if (ref.source$ !== source$) {
+    ref.source$ = source$
+    ref.args[0] = (next: () => void) => {
+      const subscription = source$.subscribe({
+        next,
+        error: (e) => {
+          setError(() => {
+            throw e
+          })
+        },
+      })
+      return () => {
+        subscription.unsubscribe()
+      }
+    }
+  }
+
+  return useSyncExternalStore(...ref!.args)
+}