feat: simplify EffectObservables to just Suspense

Author: josepot

package-lock.json

@@ -8,6 +8,9 @@ module.exports = {
   globals: {
     "ts-jest": {
       babelConfig: true,
+      diagnostics: {
+        warnOnly: true,
+      },
     },
   },
 }

packages/core/jest.config.js

@@ -1,5 +1,5 @@
 {
-  "version": "0.10.0-rc.3",
+  "version": "0.10.0-rc.7",
   "repository": {
     "type": "git",
     "url": "git+https://github.com/re-rxjs/react-rxjs.git"
@@ -50,7 +50,7 @@
     "Victor Oliva (https://github.com/voliva)"
   ],
   "dependencies": {
-    "@rx-state/core": "0.1.0-rc.3",
+    "@rx-state/core": "0.1.0-rc.8",
     "use-sync-external-store": "^1.0.0"
   },
   "devDependencies": {

packages/core/package.json

@@ -1,7 +1,13 @@
-import { state } from "@rx-state/core"
+import {
+  EmptyObservableError,
+  NoSubscribersError,
+  sinkSuspense,
+  state,
+  SUSPENSE,
+} from "@rx-state/core"
 import { act, render, screen } from "@testing-library/react"
 import React, { StrictMode, useEffect, useState } from "react"
-import { defer, EMPTY, NEVER, Observable, of, startWith } from "rxjs"
+import { defer, EMPTY, NEVER, Observable, of, startWith, Subject } from "rxjs"
 import { bind, RemoveSubscribe, Subscribe as OriginalSubscribe } from "./"
 import { TestErrorBoundary } from "./test-helpers/TestErrorBoundary"
 import { useStateObservable } from "./useStateObservable"
@@ -160,6 +166,23 @@ describe("Subscribe", () => {
       expect(getByTestId("value").textContent).toBe("2")
       instanceTwoSubs.unsubscribe()
     })
+
+    it("lifts the effects of the source$ prop", () => {
+      const subject$ = new Subject<number | SUSPENSE>()
+      const test$ = state(subject$.pipe(sinkSuspense()))
+
+      const { unmount } = render(<Subscribe source$={test$} />)
+
+      expect(test$.getRefCount()).toBe(1)
+
+      act(() => subject$.next(SUSPENSE))
+      expect(test$.getRefCount()).toBe(1)
+
+      act(() => subject$.next(1))
+      expect(test$.getRefCount()).toBe(1)
+
+      unmount()
+    })
   })
   describe("Subscribe without source$", () => {
     it("subscribes to the provided observable and remains subscribed until it's unmounted", () => {
@@ -337,6 +360,76 @@ describe("Subscribe", () => {
       )
       unmount()
     })
+
+    it("propagates the EmptyObservable error if a stream completes synchronously", async () => {
+      const globalErrors = jest.spyOn(console, "error")
+      globalErrors.mockImplementation()
+
+      const [useEmpty] = bind(() => EMPTY)
+
+      const ErrorComponent = () => {
+        useEmpty()
+        return null
+      }
+
+      const errorCallback = jest.fn()
+      const { unmount } = render(
+        <TestErrorBoundary onError={errorCallback}>
+          <Subscribe fallback={<div>Loading...</div>}>
+            <ErrorComponent />
+          </Subscribe>
+        </TestErrorBoundary>,
+      )
+
+      // Can't have NoSubscribersError
+      // Can't have "Cannot update component (`%s`) while rendering a different component"
+      globalErrors.mock.calls.forEach(([errorMessage]) => {
+        expect(errorMessage).not.toContain(NoSubscribersError.name)
+        expect(errorMessage).not.toContain(
+          "Cannot update a component (`%s`) while rendering a different component",
+        )
+      })
+      globalErrors.mockRestore()
+
+      // Must have EmptyObservableError
+      expect(errorCallback.mock.calls.length).toBe(1)
+      expect(errorCallback.mock.calls[0][0]).toBeInstanceOf(
+        EmptyObservableError,
+      )
+
+      unmount()
+    })
+
+    it("lifts the effects of observables passed through context", () => {
+      const subject$ = new Subject<number | SUSPENSE>()
+      let innerSubs = 0
+      const test$ = state(
+        defer(() => {
+          innerSubs++
+          return subject$
+        }).pipe(sinkSuspense()),
+      )
+
+      const Child = () => <>{useStateObservable(test$)}</>
+
+      const { unmount } = render(
+        <Subscribe>
+          <Child />
+        </Subscribe>,
+      )
+
+      expect(test$.getRefCount()).toBe(1)
+
+      act(() => subject$.next(SUSPENSE))
+      expect(test$.getRefCount()).toBe(1)
+
+      act(() => subject$.next(1))
+      expect(test$.getRefCount()).toBe(1)
+
+      expect(innerSubs).toBe(1)
+
+      unmount()
+    })
   })
 })
 

packages/core/src/Subscribe.test.tsx

@@ -8,7 +8,8 @@ import React, {
   useContext,
 } from "react"
 import { Observable, Subscription } from "rxjs"
-import type { StateObservable } from "@rx-state/core"
+import { liftSuspense, StateObservable } from "@rx-state/core"
+import { EMPTY_VALUE } from "./internal/empty-value"
 
 const SubscriptionContext = createContext<
   ((src: StateObservable<any>) => void) | null
@@ -54,14 +55,26 @@ export const Subscribe: React.FC<{
     subscriptionRef.current = {
       s,
       u: (src) => {
+        let error = EMPTY_VALUE
+        let synchronous = true
         s.add(
-          src.subscribe({
-            error: (e) =>
+          liftSuspense()(src).subscribe({
+            error: (e) => {
+              if (synchronous) {
+                // Can't setState of this component when another one is rendering.
+                error = e
+                return
+              }
               setSubscribedSource(() => {
                 throw e
-              }),
+              })
+            },
           }),
         )
+        synchronous = false
+        if (error !== EMPTY_VALUE) {
+          throw error
+        }
       },
     }
   }
@@ -85,7 +98,7 @@ export const Subscribe: React.FC<{
     setSubscribedSource(source$)
     if (!source$) return
 
-    const subscription = source$.subscribe({
+    const subscription = liftSuspense()(source$).subscribe({
       error: (e) =>
         setSubscribedSource(() => {
           throw e

packages/core/src/Subscribe.tsx

@@ -852,7 +852,7 @@ describe("connectFactoryObservable", () => {
         const [, me$] = bind(
           (key: number): Observable<number> => {
             nSubscriptions++
-            return me$(key).pipe(
+            return defer(() => me$(key)).pipe(
               take(1),
               map((x) => x * 2),
             )