fix(core/subscribe): stop rendering children when source$ switches

packages/core/src/Subscribe.test.tsx

@@ -1,7 +1,7 @@
-import React from "react"
 import { render } from "@testing-library/react"
-import { Observable } from "rxjs"
-import { Subscribe, bind } from "./"
+import React, { useLayoutEffect, useState } from "react"
+import { defer, Observable, of } from "rxjs"
+import { bind, Subscribe } from "./"
 
 describe("Subscribe", () => {
   it("subscribes to the provided observable and remains subscribed until it's unmounted", () => {
@@ -31,4 +31,83 @@ describe("Subscribe", () => {
     unmount()
     expect(nSubscriptions).toBe(0)
   })
+
+  it("doesn't render its content until it has subscribed to a new source", () => {
+    let nSubscriptions = 0
+    let errored = false
+    const [useInstance, instance$] = bind((id: number) => {
+      if (id === 0) {
+        return of(0)
+      }
+      return defer(() => {
+        nSubscriptions++
+        return of(1)
+      })
+    })
+
+    const Child = ({ id }: { id: number }) => {
+      const value = useInstance(id)
+
+      if (id !== 0 && nSubscriptions === 0) {
+        errored = true
+      }
+
+      return <>{value}</>
+    }
+    const { rerender } = render(
+      <Subscribe source$={instance$(0)}>
+        <Child id={0} />
+      </Subscribe>,
+    )
+    expect(nSubscriptions).toBe(0)
+    expect(errored).toBe(false)
+
+    rerender(
+      <Subscribe source$={instance$(1)}>
+        <Child id={1} />
+      </Subscribe>,
+    )
+    expect(nSubscriptions).toBe(1)
+    expect(errored).toBe(false)
+  })
+
+  it("prevents the issue of stale data when switching keys", () => {
+    const [useInstance, instance$] = bind((id: number) => of(id))
+
+    const Child = ({
+      id,
+      initialValue,
+    }: {
+      id: number
+      initialValue: number
+    }) => {
+      const [value] = useState(initialValue)
+
+      return (
+        <>
+          <div data-testid="id">{id}</div>
+          <div data-testid="value">{value}</div>
+        </>
+      )
+    }
+
+    const Parent = ({ id }: { id: number }) => {
+      const value = useInstance(id)
+
+      return <Child key={id} id={id} initialValue={value} />
+    }
+    const { rerender, getByTestId } = render(
+      <Subscribe source$={instance$(0)}>
+        <Parent id={0} />
+      </Subscribe>,
+    )
+
+    rerender(
+      <Subscribe source$={instance$(1)}>
+        <Parent id={1} />
+      </Subscribe>,
+    )
+    expect(getByTestId("id").textContent).toBe("1")
+    expect(getByTestId("value").textContent).toBe("1")
+  })
 })

packages/core/src/Subscribe.tsx

@@ -6,6 +6,15 @@ const Throw = () => {
   throw p
 }
 
+const initFn = (source$: Observable<any>) => {
+  try {
+    ;(source$ as any).gV()
+    return source$
+  } catch (e) {
+    return e.then ? source$ : null
+  }
+}
+
 /**
  * A React Component that creates a subscription to the provided observable once
  * the component mounts and it unsubscribes when the component unmounts.
@@ -19,25 +28,31 @@ export const Subscribe: React.FC<{
   source$: Observable<any>
   fallback?: NonNullable<ReactNode> | null
 }> = ({ source$, children, fallback }) => {
-  const [mounted, setMounted] = useState(() => {
-    try {
-      ;(source$ as any).gV()
-      return 1
-    } catch (e) {
-      return e.then ? 1 : 0
+  const [subscribedSource, setSubscribedSource] = useState(() =>
+    initFn(source$),
+  )
+  if (subscribedSource && subscribedSource !== source$) {
+    const result = initFn(source$)
+    if (result) {
+      setSubscribedSource(result)
     }
-  })
+  }
+
   useLayoutEffect(() => {
     const subscription = source$.subscribe(noop, (e) =>
-      setMounted(() => {
+      setSubscribedSource(() => {
         throw e
       }),
     )
-    setMounted(1)
+    setSubscribedSource(source$)
     return () => {
       subscription.unsubscribe()
     }
   }, [source$])
   const fBack = fallback || null
-  return <Suspense fallback={fBack}>{mounted ? children : <Throw />}</Suspense>
+  return (
+    <Suspense fallback={fBack}>
+      {subscribedSource === source$ ? children : <Throw />}
+    </Suspense>
+  )
 }

packages/core/src/bind/connectFactoryObservable.test.tsx

@@ -9,14 +9,22 @@ import {
   merge,
 } from "rxjs"
 import { renderHook, act as actHook } from "@testing-library/react-hooks"
-import { delay, take, catchError, map, switchMapTo } from "rxjs/operators"
+import {
+  delay,
+  take,
+  catchError,
+  map,
+  switchMapTo,
+  first,
+} from "rxjs/operators"
 import { FC, useState } from "react"
 import React from "react"
 import {
   act as componentAct,
   fireEvent,
   screen,
   render,
+  act,
 } from "@testing-library/react"
 import { bind, Subscribe } from "../"
 import { TestErrorBoundary } from "../test-helpers/TestErrorBoundary"
@@ -183,6 +191,30 @@ describe("connectFactoryObservable", () => {
       subs.unsubscribe()
     })
 
+    it("immediately switches the state to the new observable", () => {
+      const [useNumber, getNumber$] = bind((x: number) => of(x))
+      const subs = merge(
+        getNumber$(0),
+        getNumber$(1),
+        getNumber$(2),
+      ).subscribe()
+
+      const Form = ({ id }: { id: number }) => {
+        const value = useNumber(id)
+
+        return <input role="input" key={id} defaultValue={value} />
+      }
+
+      const { rerender, getByRole } = render(<Form id={0} />)
+      expect((getByRole("input") as HTMLInputElement).value).toBe("0")
+
+      act(() => rerender(<Form id={1} />))
+      expect((getByRole("input") as HTMLInputElement).value).toBe("1")
+
+      act(() => rerender(<Form id={2} />))
+      expect((getByRole("input") as HTMLInputElement).value).toBe("2")
+    })
+
     it("handles optional args correctly", () => {
       const [, getNumber$] = bind((x: number, y?: number) => of(x + (y ?? 0)))
 
@@ -217,7 +249,8 @@ describe("connectFactoryObservable", () => {
       expect(screen.queryByText("Result")).toBeNull()
       expect(screen.queryByText("Waiting")).not.toBeNull()
       await componentAct(async () => {
-        await wait(60)
+        await getDelayedNumber$(0).pipe(first()).toPromise()
+        await wait(0)
       })
       expect(screen.queryByText("Result 0")).not.toBeNull()
       expect(screen.queryByText("Waiting")).toBeNull()

packages/core/src/bind/connectFactoryObservable.ts

@@ -74,8 +74,7 @@ export default function connectFactoryObservable<A extends [], O>(
   }
 
   return [
-    (...input: A) =>
-      useObservable(getSharedObservables$(input), input, defaultValue),
+    (...input: A) => useObservable(getSharedObservables$(input), defaultValue),
     (...input: A) => getSharedObservables$(input),
   ]
 }

packages/core/src/bind/connectObservable.test.tsx

@@ -359,6 +359,33 @@ describe("connectObservable", () => {
     unmount()
   })
 
+  it("allows sync errors to be caught in error boundaries when there is a default value", () => {
+    const errStream = new Observable((observer) =>
+      observer.error("controlled error"),
+    )
+    const [useError, errStream$] = bind(errStream, 0)
+
+    const ErrorComponent = () => {
+      const value = useError()
+      return <>{value}</>
+    }
+
+    const errorCallback = jest.fn()
+    const { unmount } = render(
+      <TestErrorBoundary onError={errorCallback}>
+        <Subscribe source$={errStream$} fallback={<div>Loading...</div>}>
+          <ErrorComponent />
+        </Subscribe>
+      </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, errStream$] = bind(errStream)

packages/core/src/bind/connectObservable.ts

@@ -17,13 +17,12 @@ import { useObservable } from "../internal/useObservable"
  * subscription, then the hook will leverage React Suspense while it's waiting
  * for the first value.
  */
-const emptyArr: Array<any> = []
 export default function connectObservable<T>(
   observable: Observable<T>,
   defaultValue: T,
 ) {
   const sharedObservable$ = shareLatest<T>(observable, false, defaultValue)
   const useStaticObservable = () =>
-    useObservable(sharedObservable$, emptyArr, defaultValue)
+    useObservable(sharedObservable$, defaultValue)
   return [useStaticObservable, sharedObservable$] as const
 }

packages/core/src/internal/useObservable.ts

@@ -5,11 +5,17 @@ import { BehaviorObservable } from "../internal/BehaviorObservable"
 
 export const useObservable = <O>(
   source$: BehaviorObservable<O>,
-  keys: Array<any>,
   defaultValue: O,
 ): Exclude<O, typeof SUSPENSE> => {
-  const [state, setState] = useState(source$.gV)
-  const prevStateRef = useRef<O | (() => O)>(state)
+  const [state, setState] = useState<[O, BehaviorObservable<O>]>(() => [
+    source$.gV(),
+    source$,
+  ])
+  const prevStateRef = useRef<O | (() => O)>(state[0])
+
+  if (source$ !== state[1]) {
+    setState([source$.gV(), source$])
+  }
 
   useEffect(() => {
     const { gV } = source$
@@ -28,25 +34,25 @@ export const useObservable = <O>(
     }, onError)
     if (err !== EMPTY_VALUE) return
 
-    const set = (value: O | (() => O)) => {
-      if (!Object.is(prevStateRef.current, value))
-        setState((prevStateRef.current = value))
+    const set = (value: O) => {
+      if (!Object.is(prevStateRef.current, value)) {
+        setState([(prevStateRef.current = value), source$])
+      }
     }
 
-    if (syncVal === EMPTY_VALUE) {
-      set(defaultValue === EMPTY_VALUE ? gV : defaultValue)
-    }
+    if (syncVal === EMPTY_VALUE) set(defaultValue)
 
     const t = subscription
     subscription = source$.subscribe((value: O | typeof SUSPENSE) => {
-      set(value === SUSPENSE ? gV : value)
+      if (value !== SUSPENSE) set(value)
+      else setState(gV as any)
     }, onError)
     t.unsubscribe()
 
     return () => {
       subscription.unsubscribe()
     }
-  }, keys)
+  }, [source$])
 
-  return state as Exclude<O, typeof SUSPENSE>
+  return state[0] as Exclude<O, typeof SUSPENSE>
 }