v0.9.0 rxjs-state integration

Co-authored-by: Víctor Oliva <olivarra1@gmail.com>

Author: josepot

package.json

@@ -44,8 +44,8 @@
     "@babel/preset-env": "^7.16.11",
     "@babel/preset-typescript": "^7.16.7",
     "@testing-library/jest-dom": "^5.16.2",
-    "@testing-library/react": "^12.1.4",
-    "@testing-library/react-hooks": "^7.0.2",
+    "@testing-library/react": "^13.0.0-alpha.6",
+    "@testing-library/react-hooks": "^8.0.0-alpha.1",
     "@types/jest": "^27.4.1",
     "@types/react": "^17.0.40",
     "@types/react-dom": "^17.0.13",
@@ -56,9 +56,9 @@
     "jest-marbles": "^3.0.2",
     "lint-staged": ">=11.2.0",
     "prettier": "^2.5.1",
-    "react": "^17.0.2",
-    "react-dom": "^17.0.2",
-    "react-test-renderer": "^17.0.2",
+    "react": "^18.0.0-rc.2",
+    "react-dom": "^18.0.0-rc.2",
+    "react-test-renderer": "^18.0.0-rc.2",
     "rxjs": "^7.5.5",
     "ts-jest": "^27.1.3",
     "tslib": "^2.3.1",

packages/core/package.json

@@ -1,5 +1,5 @@
 {
-  "version": "0.8.7",
+  "version": "0.9.0",
   "repository": {
     "type": "git",
     "url": "git+https://github.com/re-rxjs/react-rxjs.git"
@@ -37,7 +37,7 @@
   },
   "peerDependencies": {
     "react": ">=16.8.0",
-    "rxjs": ">=6"
+    "rxjs": ">=7"
   },
   "prettier": {
     "printWidth": 80,
@@ -50,6 +50,7 @@
     "Victor Oliva (https://github.com/voliva)"
   ],
   "dependencies": {
+    "@josepot/rxjs-state": "^0.1.0",
     "use-sync-external-store": "^1.0.0-rc.0"
   },
   "devDependencies": {

packages/core/src/SUSPENSE.ts

@@ -4,3 +4,7 @@
  * leverage React Suspense API while we are waiting for that value.
  */
 export const SUSPENSE = Symbol("SUSPENSE")
+
+export const filterOutSuspense = <T>(
+  value: T,
+): value is Exclude<T, typeof SUSPENSE> => value !== (SUSPENSE as any)

packages/core/src/Subscribe.tsx

@@ -42,9 +42,7 @@ export const Subscribe: React.FC<{
 }> = ({ source$, children, fallback }) => {
   const subscriptionRef = useRef<Subscription>()
 
-  if (!subscriptionRef.current) {
-    subscriptionRef.current = new Subscription()
-  }
+  if (!subscriptionRef.current) subscriptionRef.current = new Subscription()
 
   const [subscribedSource, setSubscribedSource] = useState<
     Observable<any> | null | undefined
@@ -54,31 +52,25 @@ export const Subscribe: React.FC<{
     if (source$ === undefined) {
       setSubscribedSource(source$)
     } else {
-      let result: any
       try {
-        ;(source$ as any).gV()
-        result = source$
-      } catch (e: any) {
-        result = e.then ? source$ : null
-      }
-      if (result) {
-        setSubscribedSource(result)
-      }
+        ;(source$ as any).getValue()
+        setSubscribedSource(source$)
+      } catch (e: any) {}
     }
   }
 
   useEffect(() => {
-    const subscription =
-      source$ &&
-      source$.subscribe({
-        error: (e) =>
-          setSubscribedSource(() => {
-            throw e
-          }),
-      })
     setSubscribedSource(source$)
+    if (!source$) return
+
+    const subscription = source$.subscribe({
+      error: (e) =>
+        setSubscribedSource(() => {
+          throw e
+        }),
+    })
     return () => {
-      subscription && subscription.unsubscribe()
+      subscription.unsubscribe()
     }
   }, [source$])
 

packages/core/src/bind/connectFactoryObservable.ts

@@ -1,10 +1,9 @@
 import { noop, Observable } from "rxjs"
-import shareLatest from "../internal/share-latest"
-import { BehaviorObservable } from "../internal/BehaviorObservable"
 import { useObservable } from "../internal/useObservable"
 import { SUSPENSE } from "../SUSPENSE"
 import { EMPTY_VALUE } from "../internal/empty-value"
 import { useSubscription } from "../Subscribe"
+import { state, StateObservable } from "@josepot/rxjs-state"
 
 /**
  * Accepts: A factory function that returns an Observable.
@@ -30,106 +29,19 @@ export default function connectFactoryObservable<A extends [], O>(
   defaultValue: O | ((...args: A) => O),
 ): [
   (...args: A) => Exclude<O, typeof SUSPENSE>,
-  (...args: A) => Observable<O>,
+  (...args: A) => StateObservable<O>,
 ] {
-  const cache = new NestedMap<A, BehaviorObservable<O>>()
-  const getDefaultValue = (
-    typeof defaultValue === "function" ? defaultValue : () => defaultValue
-  ) as (...args: A) => O
-
-  const getSharedObservables$ = (input: A): BehaviorObservable<O> => {
-    for (let i = input.length - 1; input[i] === undefined && i > -1; i--) {
-      input.splice(-1)
-    }
-    const keys = [input.length, ...input] as any as A
-    const cachedVal = cache.get(keys)
-
-    if (cachedVal !== undefined) {
-      return cachedVal
-    }
-
-    const sharedObservable$ = shareLatest(
-      new Observable<O>((observer) =>
-        getObservable(...input).subscribe(observer),
-      ),
-      getDefaultValue(...input),
-      false,
-      () => {
-        cache.delete(keys)
-      },
-    )
-
-    const publicShared$ = new Observable<O>((subscriber) => {
-      const inCache = cache.get(keys)
-      let source$: BehaviorObservable<O> = sharedObservable$
-
-      if (!inCache) {
-        cache.set(keys, result)
-      } else if (inCache !== publicShared$) {
-        source$ = inCache
-        publicShared$.gV = source$.gV
-      }
-
-      return source$.subscribe(subscriber)
-    }) as BehaviorObservable<O>
-    publicShared$.gV = sharedObservable$.gV
-
-    const result: BehaviorObservable<O> = publicShared$
-
-    cache.set(keys, result)
-    return result
-  }
-
+  const args:
+    | [(...args: A) => Observable<O>]
+    | [(...args: A) => Observable<O>, O | ((...args: A) => O)] =
+    defaultValue === EMPTY_VALUE
+      ? [getObservable]
+      : [getObservable, defaultValue]
+
+  const obs = state(...(args as [(...args: A) => Observable<O>]))
   const useSub = defaultValue === EMPTY_VALUE ? useSubscription : noop
   return [
-    (...input: A) =>
-      useObservable(getSharedObservables$(input), useSub() as any),
-    (...input: A) => getSharedObservables$(input),
+    (...input: A) => useObservable(obs(...input) as any, useSub() as any),
+    obs,
   ]
 }
-
-class NestedMap<K extends [], V extends Object> {
-  private root: Map<K, any>
-  constructor() {
-    this.root = new Map()
-  }
-
-  get(keys: K[]): V | undefined {
-    let current: any = this.root
-    for (let i = 0; i < keys.length; i++) {
-      current = current.get(keys[i])
-      if (!current) return undefined
-    }
-    return current
-  }
-
-  set(keys: K[], value: V): void {
-    let current: Map<K, any> = this.root
-    let i
-    for (i = 0; i < keys.length - 1; i++) {
-      let nextCurrent = current.get(keys[i])
-      if (!nextCurrent) {
-        nextCurrent = new Map<K, any>()
-        current.set(keys[i], nextCurrent)
-      }
-      current = nextCurrent
-    }
-    current.set(keys[i], value)
-  }
-
-  delete(keys: K[]): void {
-    const maps: Map<K, any>[] = [this.root]
-    let current: Map<K, any> = this.root
-
-    for (let i = 0; i < keys.length - 1; i++) {
-      maps.push((current = current.get(keys[i])))
-    }
-
-    let mapIdx = maps.length - 1
-    maps[mapIdx].delete(keys[mapIdx])
-
-    while (--mapIdx > -1 && maps[mapIdx].get(keys[mapIdx]).size === 0) {
-      maps[mapIdx].delete(keys[mapIdx])
-    }
-  }
-}

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

@@ -25,7 +25,7 @@ import {
   catchError,
   switchMapTo,
 } from "rxjs/operators"
-import { bind, SUSPENSE, Subscribe } from "../"
+import { bind, SUSPENSE, Subscribe, useStateObservable } from "../"
 import { TestErrorBoundary } from "../test-helpers/TestErrorBoundary"
 
 const wait = (ms: number) => new Promise((res) => setTimeout(res, ms))
@@ -63,9 +63,11 @@ describe("connectObservable", () => {
 
   it("suspends the component when the observable hasn't emitted yet.", async () => {
     const source$ = of(1).pipe(delay(100))
-    const [useDelayedNumber, delayedNumber$] = bind(source$)
+    const [, delayedNumber$] = bind(source$)
     const sub = delayedNumber$.subscribe()
-    const Result: React.FC = () => <div>Result {useDelayedNumber()}</div>
+    const Result: React.FC = () => (
+      <div>Result {useStateObservable(delayedNumber$)}</div>
+    )
     const TestSuspense: React.FC = () => {
       return (
         <Suspense fallback={<span>Waiting</span>}>

packages/core/src/bind/connectObservable.ts

@@ -1,8 +1,8 @@
 import { EMPTY_VALUE } from "../internal/empty-value"
 import { noop, Observable } from "rxjs"
 import { useSubscription } from "../Subscribe"
-import shareLatest from "../internal/share-latest"
 import { useObservable } from "../internal/useObservable"
+import { state } from "@josepot/rxjs-state"
 
 /**
  * Accepts: An Observable.
@@ -23,9 +23,13 @@ export default function connectObservable<T>(
   observable: Observable<T>,
   defaultValue: T,
 ) {
-  const sharedObservable$ = shareLatest<T>(observable, defaultValue, false)
   const useSub = defaultValue === EMPTY_VALUE ? useSubscription : noop
+  const sharedObservable$ =
+    defaultValue === EMPTY_VALUE
+      ? state(observable)
+      : state(observable, defaultValue)
+
   const useStaticObservable = () =>
-    useObservable(sharedObservable$, useSub() as any)
+    useObservable(sharedObservable$ as any, useSub() as any)
   return [useStaticObservable, sharedObservable$] as const
 }

packages/core/src/bind/index.ts

@@ -3,6 +3,7 @@ import { SUSPENSE } from "../SUSPENSE"
 import connectFactoryObservable from "./connectFactoryObservable"
 import connectObservable from "./connectObservable"
 import { EMPTY_VALUE } from "../internal/empty-value"
+import { StateObservable } from "@josepot/rxjs-state"
 
 /**
  * Binds an observable to React
@@ -23,7 +24,7 @@ import { EMPTY_VALUE } from "../internal/empty-value"
 export function bind<T>(
   observable: Observable<T>,
   defaultValue?: T,
-): [() => Exclude<T, typeof SUSPENSE>, Observable<T>]
+): [() => Exclude<T, typeof SUSPENSE>, StateObservable<T>]
 
 /**
  * Binds a factory observable to React
@@ -48,7 +49,10 @@ export function bind<T>(
 export function bind<A extends unknown[], O>(
   getObservable: (...args: A) => Observable<O>,
   defaultValue?: O | ((...args: A) => O),
-): [(...args: A) => Exclude<O, typeof SUSPENSE>, (...args: A) => Observable<O>]
+): [
+  (...args: A) => Exclude<O, typeof SUSPENSE>,
+  (...args: A) => StateObservable<O>,
+]
 
 export function bind(observable: any, defaultValue: any) {
   return (

packages/core/src/index.tsx

@@ -1,4 +1,6 @@
-export { bind } from "./bind"
+export * from "@josepot/rxjs-state"
 export { shareLatest } from "./shareLatest"
+export { useStateObservable } from "./useStateObservable"
+export { bind } from "./bind"
 export { SUSPENSE } from "./SUSPENSE"
 export { Subscribe } from "./Subscribe"