@react-rxjs/core@v0.6.0

packages/core/package.json

@@ -1,5 +1,5 @@
 {
-  "version": "0.5.0",
+  "version": "0.6.0",
   "repository": {
     "type": "git",
     "url": "git+https://github.com/re-rxjs/react-rxjs.git"

packages/core/src/Subscribe.test.tsx

@@ -1,23 +1,26 @@
 import React from "react"
 import { render } from "@testing-library/react"
-import { defer, Subject } from "rxjs"
-import { share, finalize } from "rxjs/operators"
-import { Subscribe } from "./"
+import { Observable } from "rxjs"
+import { Subscribe, bind } from "./"
 
 describe("Subscribe", () => {
   it("subscribes to the provided observable and remains subscribed until it's unmounted", () => {
     let nSubscriptions = 0
-    const source$ = defer(() => {
-      nSubscriptions++
-      return new Subject()
-    }).pipe(
-      finalize(() => {
-        nSubscriptions--
+    const [useNumber, number$] = bind(
+      new Observable<number>(() => {
+        nSubscriptions++
+        return () => {
+          nSubscriptions--
+        }
       }),
-      share(),
     )
 
-    const TestSubscribe: React.FC = () => <Subscribe source$={source$} />
+    const Number: React.FC = () => <>{useNumber()}</>
+    const TestSubscribe: React.FC = () => (
+      <Subscribe source$={number$}>
+        <Number />
+      </Subscribe>
+    )
 
     expect(nSubscriptions).toBe(0)
 

packages/core/src/Subscribe.tsx

@@ -1,5 +1,10 @@
-import React, { useState, useEffect } from "react"
-import { Observable } from "rxjs"
+import React, { useState, Suspense, useLayoutEffect, ReactNode } from "react"
+import { Observable, noop } from "rxjs"
+
+const p = Promise.resolve()
+const Throw = () => {
+  throw p
+}
 
 /**
  * A React Component that creates a subscription to the provided observable once
@@ -12,15 +17,27 @@ import { Observable } from "rxjs"
  */
 export const Subscribe: React.FC<{
   source$: Observable<any>
-  fallback?: null | JSX.Element
+  fallback?: NonNullable<ReactNode> | null
 }> = ({ source$, children, fallback }) => {
-  const [mounted, setMounted] = useState(0)
-  useEffect(() => {
-    const subscription = source$.subscribe()
+  const [mounted, setMounted] = useState(() => {
+    try {
+      ;(source$ as any).gV()
+      return 1
+    } catch (e) {
+      return e.then ? 1 : 0
+    }
+  })
+  useLayoutEffect(() => {
+    const subscription = source$.subscribe(noop, (e) =>
+      setMounted(() => {
+        throw e
+      }),
+    )
     setMounted(1)
     return () => {
       subscription.unsubscribe()
     }
   }, [source$])
-  return <>{mounted ? children : fallback}</>
+  const fBack = fallback || null
+  return <Suspense fallback={fBack}>{mounted ? children : <Throw />}</Suspense>
 }

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

@@ -3,22 +3,22 @@ import {
   of,
   defer,
   concat,
-  BehaviorSubject,
   throwError,
   Observable,
   Subject,
+  merge,
 } from "rxjs"
 import { renderHook, act as actHook } from "@testing-library/react-hooks"
-import { switchMap, delay, take, catchError, map } from "rxjs/operators"
-import { FC, Suspense, useState } from "react"
+import { delay, take, catchError, map, switchMapTo } from "rxjs/operators"
+import { FC, useState } from "react"
 import React from "react"
 import {
   act as componentAct,
   fireEvent,
   screen,
   render,
 } from "@testing-library/react"
-import { bind } from "../"
+import { bind, Subscribe } from "../"
 import { TestErrorBoundary } from "../test-helpers/TestErrorBoundary"
 
 const wait = (ms: number) => new Promise((res) => setTimeout(res, ms))
@@ -42,8 +42,8 @@ describe("connectFactoryObservable", () => {
   })
   describe("hook", () => {
     it("returns the latest emitted value", async () => {
-      const valueStream = new BehaviorSubject(1)
-      const [useNumber] = bind(() => valueStream)
+      const valueStream = new Subject<number>()
+      const [useNumber] = bind(() => valueStream, 1)
       const { result } = renderHook(() => useNumber())
       expect(result.current).toBe(1)
 
@@ -56,13 +56,15 @@ describe("connectFactoryObservable", () => {
     it("suspends the component when the observable hasn't emitted yet.", async () => {
       const source$ = of(1).pipe(delay(100))
       const [useDelayedNumber, getDelayedNumber$] = bind(() => source$)
-      const subs = getDelayedNumber$().subscribe()
       const Result: React.FC = () => <div>Result {useDelayedNumber()}</div>
       const TestSuspense: React.FC = () => {
         return (
-          <Suspense fallback={<span>Waiting</span>}>
+          <Subscribe
+            source$={getDelayedNumber$()}
+            fallback={<span>Waiting</span>}
+          >
             <Result />
-          </Suspense>
+          </Subscribe>
         )
       }
 
@@ -75,7 +77,51 @@ describe("connectFactoryObservable", () => {
 
       expect(screen.queryByText("Result 1")).not.toBeNull()
       expect(screen.queryByText("Waiting")).toBeNull()
-      subs.unsubscribe()
+    })
+
+    it("synchronously mounts the emitted value if the observable emits synchronously", () => {
+      const source$ = of(1)
+      const [useDelayedNumber, getDelayedNumber$] = bind(() => source$)
+      const Result: React.FC = () => <div>Result {useDelayedNumber()}</div>
+      const TestSuspense: React.FC = () => {
+        return (
+          <Subscribe
+            source$={getDelayedNumber$()}
+            fallback={<span>Waiting</span>}
+          >
+            <Result />
+          </Subscribe>
+        )
+      }
+
+      render(<TestSuspense />)
+
+      expect(screen.queryByText("Result 1")).not.toBeNull()
+      expect(screen.queryByText("Waiting")).toBeNull()
+    })
+
+    it("doesn't mount the fallback element if the subscription is already active", () => {
+      const source$ = new Subject<number>()
+      const [useDelayedNumber, getDelayedNumber$] = bind(() => source$)
+      const Result: React.FC = () => <div>Result {useDelayedNumber()}</div>
+      const TestSuspense: React.FC = () => {
+        return (
+          <Subscribe
+            source$={getDelayedNumber$()}
+            fallback={<span>Waiting</span>}
+          >
+            <Result />
+          </Subscribe>
+        )
+      }
+
+      const subscription = getDelayedNumber$().subscribe()
+      source$.next(1)
+      render(<TestSuspense />)
+
+      expect(screen.queryByText("Result 1")).not.toBeNull()
+      expect(screen.queryByText("Waiting")).toBeNull()
+      subscription.unsubscribe()
     })
 
     it("shares the multicasted subscription with all of the components that use the same parameters", async () => {
@@ -114,7 +160,12 @@ describe("connectFactoryObservable", () => {
     })
 
     it("returns the value of next new Observable when the arguments change", () => {
-      const [useNumber] = bind((x: number) => of(x))
+      const [useNumber, getNumber$] = bind((x: number) => of(x))
+      const subs = merge(
+        getNumber$(0),
+        getNumber$(1),
+        getNumber$(2),
+      ).subscribe()
       const { result, rerender } = renderHook(({ input }) => useNumber(input), {
         initialProps: { input: 0 },
       })
@@ -129,6 +180,7 @@ describe("connectFactoryObservable", () => {
         rerender({ input: 2 })
       })
       expect(result.current).toBe(2)
+      subs.unsubscribe()
     })
 
     it("handles optional args correctly", () => {
@@ -149,9 +201,12 @@ describe("connectFactoryObservable", () => {
         const [input, setInput] = useState(0)
         return (
           <>
-            <Suspense fallback={<span>Waiting</span>}>
+            <Subscribe
+              source$={getDelayedNumber$(input)}
+              fallback={<span>Waiting</span>}
+            >
               <Result input={input} />
-            </Suspense>
+            </Subscribe>
             <button onClick={() => setInput((x) => x + 1)}>increase</button>
           </>
         )
@@ -223,8 +278,8 @@ describe("connectFactoryObservable", () => {
     })
 
     it("allows errors to be caught in error boundaries", () => {
-      const errStream = new BehaviorSubject(1)
-      const [useError] = bind(() => errStream)
+      const errStream = new Subject()
+      const [useError] = bind(() => errStream, 1)
 
       const ErrorComponent = () => {
         const value = useError()
@@ -253,7 +308,7 @@ describe("connectFactoryObservable", () => {
       const errStream = new Observable((observer) =>
         observer.error("controlled error"),
       )
-      const [useError] = bind((_: string) => errStream)
+      const [useError, getErrStream$] = bind((_: string) => errStream)
 
       const ErrorComponent = () => {
         const value = useError("foo")
@@ -264,9 +319,12 @@ describe("connectFactoryObservable", () => {
       const errorCallback = jest.fn()
       const { unmount } = render(
         <TestErrorBoundary onError={errorCallback}>
-          <Suspense fallback={<div>Loading...</div>}>
+          <Subscribe
+            source$={getErrStream$("foo")}
+            fallback={<div>Loading...</div>}
+          >
             <ErrorComponent />
-          </Suspense>
+          </Subscribe>
         </TestErrorBoundary>,
       )
 
@@ -279,7 +337,7 @@ describe("connectFactoryObservable", () => {
 
     it("allows async errors to be caught in error boundaries with suspense", async () => {
       const errStream = new Subject()
-      const [useError] = bind((_: string) => errStream)
+      const [useError, getErrStream$] = bind((_: string) => errStream)
 
       const ErrorComponent = () => {
         const value = useError("foo")
@@ -290,9 +348,12 @@ describe("connectFactoryObservable", () => {
       const errorCallback = jest.fn()
       const { unmount } = render(
         <TestErrorBoundary onError={errorCallback}>
-          <Suspense fallback={<div>Loading...</div>}>
+          <Subscribe
+            source$={getErrStream$("foo")}
+            fallback={<div>Loading...</div>}
+          >
             <ErrorComponent />
-          </Suspense>
+          </Subscribe>
         </TestErrorBoundary>,
       )
 
@@ -325,19 +386,24 @@ describe("connectFactoryObservable", () => {
           .pipe(catchError(() => []))
           .subscribe()
 
+        const Ok: React.FC<{ ok: boolean }> = ({ ok }) => <>{useOkKo(ok)}</>
+
         const ErrorComponent = () => {
           const [ok, setOk] = useState(true)
-          const value = useOkKo(ok)
 
-          return <span onClick={() => setOk(false)}>{value}</span>
+          return (
+            <Subscribe source$={getObs$(ok)} fallback={<div>Loading...</div>}>
+              <span onClick={() => setOk(false)}>
+                <Ok ok={ok} />
+              </span>
+            </Subscribe>
+          )
         }
 
         const errorCallback = jest.fn()
         const { unmount } = render(
           <TestErrorBoundary onError={errorCallback}>
-            <Suspense fallback={<div>Loading...</div>}>
-              <ErrorComponent />
-            </Suspense>
+            <ErrorComponent />
           </TestErrorBoundary>,
         )
 
@@ -367,12 +433,11 @@ describe("connectFactoryObservable", () => {
     )
 
     it("doesn't throw errors on components that will get unmounted on the next cycle", () => {
-      const valueStream = new BehaviorSubject(1)
-      const [useValue, value$] = bind(() => valueStream)
-      const [useError] = bind(() =>
-        value$().pipe(
-          switchMap((v) => (v === 1 ? of(v) : throwError("error"))),
-        ),
+      const valueStream = new Subject<number>()
+      const [useValue, value$] = bind(() => valueStream, 1)
+      const [useError] = bind(
+        () => value$().pipe(switchMapTo(throwError("error"))),
+        1,
       )
 
       const ErrorComponent: FC = () => {
@@ -403,30 +468,6 @@ describe("connectFactoryObservable", () => {
       expect(errorCallback).not.toHaveBeenCalled()
     })
 
-    it("does not resubscribe to an observable that emits synchronously and that does not have a top-level subscription after a re-render", () => {
-      let nTopSubscriptions = 0
-
-      const [useNTopSubscriptions] = bind((id: number) =>
-        defer(() => {
-          return of(++nTopSubscriptions + id)
-        }),
-      )
-
-      const { result, rerender, unmount } = renderHook(() =>
-        useNTopSubscriptions(0),
-      )
-
-      expect(result.current).toBe(2)
-
-      actHook(() => {
-        rerender()
-      })
-      expect(result.current).toBe(2)
-      expect(nTopSubscriptions).toBe(2)
-
-      unmount()
-    })
-
     it("if the observable hasn't emitted and a defaultValue is provided, it does not start suspense", () => {
       const number$ = new Subject<number>()
       const [useNumber] = bind(