Add getServerSnapshot, fix loop on SSR Subscribe

package.json

@@ -44,6 +44,7 @@
     "@babel/preset-env": "^7.22.7",
     "@babel/preset-typescript": "^7.22.5",
     "@testing-library/react": "^14.0.0",
+    "@types/node": "^20.4.7",
     "@types/react": "^18.2.14",
     "@types/react-dom": "^18.2.6",
     "@vitest/coverage-v8": "^0.33.0",

packages/core/src/Subscribe.test.tsx

@@ -7,9 +7,20 @@ import {
 } 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, Subject } from "rxjs"
-import { describe, it, expect, vi } from "vitest"
-import { bind, RemoveSubscribe, Subscribe as OriginalSubscribe } from "./"
+import { renderToPipeableStream } from "react-dom/server"
+import {
+  defer,
+  EMPTY,
+  lastValueFrom,
+  NEVER,
+  Observable,
+  of,
+  startWith,
+  Subject,
+} from "rxjs"
+import { describe, expect, it, vi } from "vitest"
+import { bind, Subscribe as OriginalSubscribe, RemoveSubscribe } from "./"
+import { pipeableStreamToObservable } from "./test-helpers/pipeableStreamToObservable"
 import { TestErrorBoundary } from "./test-helpers/TestErrorBoundary"
 import { useStateObservable } from "./useStateObservable"
 
@@ -432,6 +443,22 @@ describe("Subscribe", () => {
       unmount()
     })
   })
+
+  describe("On SSR", () => {
+    // Testing-library doesn't support SSR yet https://github.com/testing-library/react-testing-library/issues/561
+
+    it("Renders the fallback", async () => {
+      const stream = renderToPipeableStream(
+        <Subscribe fallback={<div>Loading</div>}>
+          <div>Content</div>
+        </Subscribe>,
+      )
+      const result = await lastValueFrom(pipeableStreamToObservable(stream))
+
+      expect(result).toContain("<div>Loading</div>")
+      expect(result).not.toContain("<div>Content</div>")
+    })
+  })
 })
 
 describe("RemoveSubscribe", () => {

packages/core/src/Subscribe.tsx

@@ -125,6 +125,8 @@ export const Subscribe: React.FC<{
 
   return fallback === undefined ? (
     actualChildren
+  ) : subscribedSource === null ? (
+    fallback
   ) : (
     <Suspense fallback={fallback}>{actualChildren}</Suspense>
   )

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

@@ -10,6 +10,7 @@ import {
   defer,
   EMPTY,
   from,
+  lastValueFrom,
   merge,
   NEVER,
   Observable,
@@ -34,6 +35,8 @@ import {
   useStateObservable,
 } from "../"
 import { TestErrorBoundary } from "../test-helpers/TestErrorBoundary"
+import { renderToPipeableStream } from "react-dom/server"
+import { pipeableStreamToObservable } from "../test-helpers/pipeableStreamToObservable"
 
 const wait = (ms: number) => new Promise((res) => setTimeout(res, ms))
 
@@ -939,4 +942,49 @@ describe("connectObservable", () => {
     })
     expect(queryByText("Result 10")).not.toBeNull()
   })
+
+  describe("The hook on SSR", () => {
+    // Testing-library doesn't support SSR yet https://github.com/testing-library/react-testing-library/issues/561
+
+    it("returns the value if the state observable has a subscription", async () => {
+      const [useState, state$] = bind(of(5))
+      state$.subscribe()
+      const Component = () => {
+        const value = useState()
+        return <div>Value: {value}</div>
+      }
+      const stream = renderToPipeableStream(<Component />)
+      const result = await lastValueFrom(pipeableStreamToObservable(stream))
+
+      // Sigh...
+      expect(result).toEqual("<div>Value: <!-- -->5</div>")
+    })
+
+    it("throws Missing Subscribe if the state observable doesn't have a subscription nor a default value", async () => {
+      const [useState] = bind(of(5))
+      const Component = () => {
+        const value = useState()
+        return <div>Value: {value}</div>
+      }
+      const stream = renderToPipeableStream(<Component />)
+      try {
+        await lastValueFrom(pipeableStreamToObservable(stream))
+      } catch (ex: any) {
+        expect(ex.message).to.equal("Missing Subscribe!")
+      }
+      expect.assertions(1)
+    })
+
+    it("returns the default value if the observable didn't emit yet", async () => {
+      const [useState] = bind(of(5), 3)
+      const Component = () => {
+        const value = useState()
+        return <div>Value: {value}</div>
+      }
+      const stream = renderToPipeableStream(<Component />)
+      const result = await lastValueFrom(pipeableStreamToObservable(stream))
+
+      expect(result).toEqual("<div>Value: <!-- -->3</div>")
+    })
+  })
 })

packages/core/src/test-helpers/pipeableStreamToObservable.ts

@@ -0,0 +1,41 @@
+import { PipeableStream } from "react-dom/server"
+import { Observable, scan } from "rxjs"
+import { PassThrough } from "stream"
+
+export function pipeableStreamToObservable(
+  stream: PipeableStream,
+): Observable<string> {
+  return new Observable((subscriber) => {
+    const passthrough = new PassThrough()
+    const sub = readStream$<string>(passthrough)
+      .pipe(scan((acc, v) => acc + v, ""))
+      .subscribe(subscriber)
+
+    stream.pipe(passthrough)
+
+    return () => {
+      sub.unsubscribe()
+    }
+  })
+}
+
+function readStream$<T>(stream: NodeJS.ReadableStream) {
+  return new Observable<T>((subscriber) => {
+    const dataHandler = (data: T) => subscriber.next(data)
+    stream.addListener("data", dataHandler)
+
+    const errorHandler = (error: any) => subscriber.error(error)
+    stream.addListener("error", errorHandler)
+
+    const closeHandler = () => subscriber.complete()
+    stream.addListener("close", closeHandler)
+    stream.addListener("end", closeHandler)
+
+    return () => {
+      stream.removeListener("data", dataHandler)
+      stream.removeListener("error", errorHandler)
+      stream.removeListener("close", closeHandler)
+      stream.removeListener("end", closeHandler)
+    }
+  })
+}

packages/core/src/useStateObservable.ts

@@ -14,7 +14,11 @@ type VoidCb = () => void
 
 interface Ref<T> {
   source$: StateObservable<T>
-  args: [(cb: VoidCb) => VoidCb, () => Exclude<T, SUSPENSE>]
+  args: [
+    (cb: VoidCb) => VoidCb,
+    () => Exclude<T, typeof SUSPENSE>,
+    () => Exclude<T, typeof SUSPENSE>,
+  ]
 }
 
 export const useStateObservable = <O>(
@@ -46,7 +50,7 @@ export const useStateObservable = <O>(
 
     callbackRef.current = {
       source$: null as any,
-      args: [, gv] as any,
+      args: [, gv, gv] as any,
     }
   }