v0.2.0-alpha.3

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@josepot/react-rxjs",
-  "version": "0.2.0-alpha.2",
+  "version": "0.2.0-alpha.3",
   "lockfileVersion": 1,
   "requires": true,
   "dependencies": {

package.json

@@ -1,5 +1,5 @@
 {
-  "version": "0.2.0-alpha.2",
+  "version": "0.2.0-alpha.3",
   "sideEffects": false,
   "repository": {
     "type": "git",

src/connectGroupedObservable.tsx

@@ -1,16 +1,19 @@
-import React, { useMemo } from "react"
+import { useMemo } from "react"
 import { Observable, GroupedObservable } from "rxjs"
-import { map, filter, take, concatMap } from "rxjs/operators"
+import { map, filter, take, concatMap, shareReplay } from "rxjs/operators"
 import distinctShareReplay from "./operators/distinct-share-replay"
 import { FactoryObservableOptions, defaultFactoryOptions } from "./options"
 import useSharedReplayableObservable from "./useSharedReplayableObservable"
-import { useLayoutEffect } from "react"
 
 const connectGroupedObservable = <K, O, I>(
   source$: Observable<GroupedObservable<K, O>>,
   initialValue: I,
   _options?: FactoryObservableOptions<O>,
-): [(key: K) => I | O, React.FC, (key: K) => Observable<O>] => {
+): [
+  (key: K) => I | O,
+  () => () => void,
+  (key: K) => GroupedObservable<K, O>,
+] => {
   const options = {
     ...defaultFactoryOptions,
     ..._options,
@@ -26,18 +29,18 @@ const connectGroupedObservable = <K, O, I>(
       )
       return observables
     }),
-    distinctShareReplay(
-      () => false,
-      () => observables.clear(),
-    ),
+    shareReplay(1),
   )
 
-  const getObservableByKey = (key: K) =>
-    activeObservables$.pipe(
+  const getObservableByKey = (key: K) => {
+    const result = activeObservables$.pipe(
       filter(x => x.has(key)),
       take(1),
       concatMap(x => x.get(key)!),
-    )
+    ) as GroupedObservable<K, O>
+    result.key = key
+    return result
+  }
 
   const hook = (key: K) =>
     useSharedReplayableObservable(
@@ -46,15 +49,12 @@ const connectGroupedObservable = <K, O, I>(
       options,
     )
 
-  const GroupSubsriber: React.FC = ({ children }) => {
-    useLayoutEffect(() => {
-      const subscription = activeObservables$.subscribe()
-      return () => subscription.unsubscribe()
-    }, [])
-    return <>{children}</>
+  const getGroupSubscription = () => {
+    const subscription = activeObservables$.subscribe()
+    return () => subscription.unsubscribe()
   }
 
-  return [hook, GroupSubsriber, getObservableByKey]
+  return [hook, getGroupSubscription, getObservableByKey]
 }
 
 export default connectGroupedObservable

src/operators/react-optimizations.ts

@@ -1,9 +0,0 @@
-import { Observable } from "rxjs"
-import { debounceTime } from "rxjs/operators"
-import delayUnsubscription from "./delay-unsubscription"
-
-const reactOptimizations = (delayTime: number) => <T>(
-  source: Observable<T>,
-): Observable<T> => source.pipe(delayUnsubscription(delayTime), debounceTime(0))
-
-export default reactOptimizations

src/useSharedReplayableObservable.ts

@@ -1,7 +1,7 @@
 import { useState, useLayoutEffect } from "react"
-import { Observable, of, race, merge, NEVER } from "rxjs"
-import { take, delay } from "rxjs/operators"
-import reactOptimizations from "./operators/react-optimizations"
+import { Observable, of, race, concat } from "rxjs"
+import { delay } from "rxjs/operators"
+import delayUnsubscription from "./operators/delay-unsubscription"
 import { defaultFactoryOptions, ObservableOptions } from "./options"
 
 const useSharedReplayableObservable = <O, I>(
@@ -17,16 +17,16 @@ const useSharedReplayableObservable = <O, I>(
 
   useLayoutEffect(() => {
     const updates$ = sharedReplayableObservable$.pipe(
-      reactOptimizations(unsubscribeGraceTime),
-    )
-    const initialState$ = race(
-      suspenseTime === Infinity
-        ? NEVER
-        : of(initialValue).pipe(delay(suspenseTime)),
-      sharedReplayableObservable$.pipe(take(1)),
+      delayUnsubscription(unsubscribeGraceTime),
     )
 
-    const subscription = merge(updates$, initialState$).subscribe(setState)
+    const subscription = (suspenseTime === Infinity
+      ? updates$
+      : race(
+          concat(of(initialValue).pipe(delay(suspenseTime)), updates$),
+          updates$,
+        )
+    ).subscribe(setState)
 
     return () => subscription.unsubscribe()
   }, [sharedReplayableObservable$, suspenseTime, unsubscribeGraceTime])

test/connectFactoryObservable.test.tsx

@@ -1,8 +1,6 @@
 import { connectFactoryObservable } from "../src"
 import { NEVER, from, of, defer, concat } from "rxjs"
 import { renderHook, act } from "@testing-library/react-hooks"
-import { useEffect, useState } from "react"
-import { delay } from "rxjs/operators"
 
 const wait = (ms: number) => new Promise(res => setTimeout(res, ms))
 
@@ -13,9 +11,6 @@ describe("connectObservable", () => {
       "initialValue",
     )
     const { result } = renderHook(() => useSomething(5))
-    await act(async () => {
-      await wait(0)
-    })
 
     expect(result.current).toBe("initialValue")
   })
@@ -23,34 +18,9 @@ describe("connectObservable", () => {
   it("returns the latest emitted value", async () => {
     const [useNumber] = connectFactoryObservable((id: number) => of(id), 0)
     const { result } = renderHook(() => useNumber(1))
-    await act(async () => {
-      await wait(0)
-    })
     expect(result.current).toBe(1)
   })
 
-  it("batches the updates that happen on the same event-loop", async () => {
-    const observable$ = from([1, 2, 3, 4, 5])
-    const [useLatestNumber] = connectFactoryObservable(
-      (id: number) => concat(observable$, of(id).pipe(delay(1000))),
-      0,
-    )
-    const useLatestNumberTest = () => {
-      const latestNumber = useLatestNumber(6)
-      const [emittedValues, setEmittedValues] = useState<number[]>([])
-      useEffect(() => {
-        setEmittedValues(prev => [...prev, latestNumber])
-      }, [latestNumber])
-      return emittedValues
-    }
-
-    const { result } = renderHook(() => useLatestNumberTest())
-    await act(async () => {
-      await wait(0)
-    })
-    expect(result.current).toEqual([0, 5])
-  })
-
   it("shares the source subscription until the refCount has stayed at zero for the grace-period", async () => {
     let nInitCount = 0
     const observable$ = defer(() => {
@@ -66,17 +36,8 @@ describe("connectObservable", () => {
       },
     )
     const { unmount } = renderHook(() => useLatestNumber(6))
-    await act(async () => {
-      await wait(0)
-    })
     const { unmount: unmount2 } = renderHook(() => useLatestNumber(6))
-    await act(async () => {
-      await wait(0)
-    })
     const { unmount: unmount3 } = renderHook(() => useLatestNumber(6))
-    await act(async () => {
-      await wait(0)
-    })
     expect(nInitCount).toBe(1)
     unmount()
     unmount2()
@@ -86,9 +47,6 @@ describe("connectObservable", () => {
       await wait(90)
     })
     const { unmount: unmount4 } = renderHook(() => useLatestNumber(6))
-    await act(async () => {
-      await wait(0)
-    })
     expect(nInitCount).toBe(1)
     unmount4()
 
@@ -96,9 +54,6 @@ describe("connectObservable", () => {
       await wait(101)
     })
     renderHook(() => useLatestNumber(6))
-    await act(async () => {
-      await wait(0)
-    })
     expect(nInitCount).toBe(2)
   })
 })

test/connectObservable.test.tsx

@@ -1,7 +1,6 @@
 import { connectObservable } from "../src"
-import { NEVER, from, of, defer, Subject } from "rxjs"
+import { NEVER, from, of, defer } from "rxjs"
 import { renderHook, act } from "@testing-library/react-hooks"
-import { useState, useLayoutEffect } from "react"
 
 const wait = (ms: number) => new Promise(res => setTimeout(res, ms))
 
@@ -9,55 +8,16 @@ describe("connectObservable", () => {
   it("returns the initial value when the stream has not emitted anything", async () => {
     const [useSomething] = connectObservable(NEVER, "initialValue")
     const { result } = renderHook(() => useSomething())
-    await act(async () => {
-      await wait(0)
-    })
 
     expect(result.current).toBe("initialValue")
   })
 
-  it("returns the latest emitted value", async () => {
-    const [useNumber] = connectObservable(of(1), 0)
-    const { result } = renderHook(() => useNumber())
-    await act(async () => {
-      await wait(0)
-    })
-    expect(result.current).toBe(1)
-  })
-
   it("sets the initial state synchronously if it's available", async () => {
     const observable$ = of(1)
     const [useLatestNumber] = connectObservable(observable$, 0)
 
-    const { result, unmount } = renderHook(() => useLatestNumber())
+    const { result } = renderHook(() => useLatestNumber())
     expect(result.current).toEqual(1)
-    unmount()
-  })
-
-  it("batches synchronous updates", async () => {
-    const observable$ = new Subject<number>()
-    const [useLatestNumber] = connectObservable(observable$, 0)
-    const useLatestNumberTest = () => {
-      const latestNumber = useLatestNumber()
-      const [emittedValues, setEmittedValues] = useState<number[]>([])
-      useLayoutEffect(() => {
-        setEmittedValues(prev => [...prev, latestNumber])
-      }, [latestNumber])
-      return emittedValues
-    }
-
-    const { result } = renderHook(() => useLatestNumberTest())
-    expect(result.current).toEqual([0])
-    await act(async () => {
-      observable$.next(0)
-      observable$.next(1)
-      observable$.next(2)
-      observable$.next(3)
-      observable$.next(4)
-      observable$.next(5)
-      await wait(0)
-    })
-    expect(result.current).toEqual([0, 5])
   })
 
   it("shares the source subscription until the refCount has stayed at zero for the grace-period", async () => {
@@ -71,17 +31,8 @@ describe("connectObservable", () => {
       unsubscribeGraceTime: 100,
     })
     const { unmount } = renderHook(() => useLatestNumber())
-    await act(async () => {
-      await wait(0)
-    })
     const { unmount: unmount2 } = renderHook(() => useLatestNumber())
-    await act(async () => {
-      await wait(0)
-    })
     const { unmount: unmount3 } = renderHook(() => useLatestNumber())
-    await act(async () => {
-      await wait(0)
-    })
     expect(nInitCount).toBe(1)
     unmount()
     unmount2()
@@ -91,9 +42,6 @@ describe("connectObservable", () => {
       await wait(90)
     })
     const { unmount: unmount4 } = renderHook(() => useLatestNumber())
-    await act(async () => {
-      await wait(0)
-    })
     expect(nInitCount).toBe(1)
     unmount4()
 
@@ -101,9 +49,6 @@ describe("connectObservable", () => {
       await wait(101)
     })
     renderHook(() => useLatestNumber())
-    await act(async () => {
-      await wait(0)
-    })
     expect(nInitCount).toBe(2)
   })
 })