Initial commit (WIP)

.gitignore

@@ -0,0 +1,5 @@
+*.log
+.DS_Store
+node_modules
+.cache
+dist

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2020 Josep M Sobrepere
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1 @@
+# react-rxjs

jest.config.js

@@ -0,0 +1,7 @@
+module.exports = {
+  globals: {
+    "ts-jest": {
+      diagnostics: false,
+    },
+  },
+}

package.json

@@ -0,0 +1,51 @@
+{
+  "version": "0.0.1",
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/josepot/react-rxjs.git"
+  },
+  "license": "MIT",
+  "main": "dist/index.js",
+  "typings": "dist/index.d.ts",
+  "files": [
+    "dist"
+  ],
+  "scripts": {
+    "start": "tsdx watch",
+    "build": "tsdx build",
+    "test": "tsdx test --passWithNoTests",
+    "lint": "tsdx lint",
+    "prepare": "tsdx build"
+  },
+  "peerDependencies": {
+    "react": ">=16.8.0",
+    "rxjs": ">=6"
+  },
+  "husky": {
+    "hooks": {
+      "pre-commit": "tsdx lint"
+    }
+  },
+  "prettier": {
+    "printWidth": 80,
+    "semi": false,
+    "trailingComma": "all"
+  },
+  "name": "@josepot/react-rxjs",
+  "author": "Josep M Sobrepere",
+  "module": "dist/react-rxjs.esm.js",
+  "devDependencies": {
+    "@testing-library/react-hooks": "^3.2.1",
+    "@types/jest": "^25.2.1",
+    "@types/react": "^16.9.34",
+    "@types/react-dom": "^16.9.7",
+    "husky": "^4.2.5",
+    "react": "^16.13.1",
+    "react-dom": "^16.13.1",
+    "react-test-renderer": "^16.13.1",
+    "rxjs": "^6.5.5",
+    "tsdx": "^0.13.2",
+    "tslib": "^1.11.1",
+    "typescript": "^3.8.3"
+  }
+}

src/connectFactoryObservable.ts

@@ -0,0 +1,65 @@
+import { Observable } from "rxjs"
+import { useEffect, useState } from "react"
+import reactOperator from "./react-operator"
+
+export function connectFactoryObservable<
+  I,
+  A extends (number | string | boolean | null)[],
+  O
+>(
+  getObservable: (...args: A) => Observable<O>,
+  initialValue: I,
+  suspenseTime: number = 200,
+): [(...args: A) => O | I, (...args: A) => Observable<O>] {
+  const cache = new Map<string, Observable<O>>()
+
+  const getReactObservable$ = (...input: A): Observable<O> => {
+    const key = JSON.stringify(input)
+    const cachedVal = cache.get(key)
+
+    if (cachedVal !== undefined) {
+      return cachedVal
+    }
+
+    const reactObservable$ = reactOperator(
+      getObservable(...input),
+      initialValue,
+      () => {
+        cache.delete(key)
+      },
+    )
+
+    cache.set(key, reactObservable$)
+    return reactObservable$
+  }
+
+  return [
+    (...input: A) => {
+      const [value, setValue] = useState<I | O>(initialValue)
+
+      useEffect(() => {
+        let timeoutToken: NodeJS.Timeout | null = null
+        if (suspenseTime === 0) {
+          setValue(initialValue)
+        } else if (suspenseTime < Infinity) {
+          timeoutToken = setTimeout(() => {
+            timeoutToken = null
+            setValue(initialValue)
+          }, suspenseTime)
+        }
+
+        const subscription = getReactObservable$(...input).subscribe(value => {
+          setValue(value)
+          if (timeoutToken !== null) clearTimeout(timeoutToken)
+        })
+        return () => {
+          subscription.unsubscribe()
+          if (timeoutToken !== null) clearTimeout(timeoutToken)
+        }
+      }, input)
+
+      return value
+    },
+    getReactObservable$,
+  ]
+}

src/connectObservable.ts

@@ -0,0 +1,22 @@
+import { Observable } from "rxjs"
+import { useEffect, useState } from "react"
+import reactOperator from "./react-operator"
+
+export function connectObservable<O, IO>(
+  observable: Observable<O>,
+  initialValue: IO,
+) {
+  const reactObservable$ = reactOperator(observable, initialValue)
+
+  const useStaticObservable = () => {
+    const [value, setValue] = useState<O | IO>(
+      reactObservable$.getCurrentValue(),
+    )
+    useEffect(() => {
+      const subscription = reactObservable$.subscribe(setValue)
+      return () => subscription.unsubscribe()
+    }, [])
+    return value
+  }
+  return [useStaticObservable, reactObservable$] as const
+}

src/index.tsx

@@ -0,0 +1,3 @@
+export { connectObservable } from "./connectObservable"
+export { connectFactoryObservable } from "./connectFactoryObservable"
+export { ReactObservable } from "./react-operator"

src/react-operator.ts

@@ -0,0 +1,74 @@
+import { Observable, ReplaySubject, Subscription } from "rxjs"
+import { debounceTime } from "rxjs/operators"
+
+export interface ReactObservable<O, IO> extends Observable<O> {
+  getCurrentValue: () => O | IO
+}
+
+const batchUpdates: <T>(source: Observable<T>) => Observable<T> = debounceTime(
+  0,
+)
+
+const GRACE_PERIOD = 100
+const reactOperator = <T, I>(
+  source$: Observable<T>,
+  initialValue: I,
+  teardown?: () => void,
+): ReactObservable<T, I> => {
+  const batchedSource$ = batchUpdates(source$)
+  let subject: ReplaySubject<T> | undefined
+  let subscription: Subscription | undefined
+  let timeoutToken: NodeJS.Timeout | undefined = undefined
+  let refCount = 0
+  let hasError = false
+  let currentValue: T | I = initialValue
+
+  const observable$ = new Observable<T>(subscriber => {
+    if (timeoutToken !== undefined) {
+      clearTimeout(timeoutToken)
+    }
+    refCount++
+    if (!subject || hasError) {
+      hasError = false
+      subject = new ReplaySubject<T>(1)
+      subscription = batchedSource$.subscribe({
+        next(value) {
+          currentValue = value
+          subject!.next(value)
+        },
+        error(err) {
+          hasError = true
+          subject!.error(err)
+        },
+        complete() {
+          subscription = undefined
+          subject!.complete()
+        },
+      })
+    }
+
+    const innerSub = subject.subscribe(subscriber)
+    return () => {
+      refCount--
+      innerSub.unsubscribe()
+      if (refCount === 0) {
+        timeoutToken = setTimeout(() => {
+          timeoutToken = undefined
+          currentValue = initialValue
+          teardown && teardown()
+          if (subscription) {
+            subscription.unsubscribe()
+            subscription = undefined
+          }
+          subject = undefined
+        }, GRACE_PERIOD)
+      }
+    }
+  })
+
+  const result = observable$ as ReactObservable<T, I>
+  result.getCurrentValue = () => currentValue
+  return result
+}
+
+export default reactOperator

test/connectObservable.test.tsx

@@ -0,0 +1,70 @@
+import { connectObservable } from "../src"
+import { NEVER, from, of, defer } from "rxjs"
+import { renderHook, act } from "@testing-library/react-hooks"
+import { useEffect, useState } from "react"
+
+const wait = (ms: number) => new Promise(res => setTimeout(res, ms))
+
+describe("connectObservable", () => {
+  it("returns the initial value when the stream has not emitted anything", () => {
+    const [useSomething] = connectObservable(NEVER, "initialValue")
+    const { result } = renderHook(() => useSomething())
+
+    expect(result.current).toBe("initialValue")
+  })
+
+  it("returns the latest emitted value", () => {
+    const [useNumber] = connectObservable(of(1), 0)
+    const { result } = renderHook(() => useNumber())
+    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] = connectObservable(observable$, 0)
+    const useLatestNumberTest = () => {
+      const latestNumber = useLatestNumber()
+      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 remained zero for 100 milliseconds", async () => {
+    let nInitCount = 0
+    const observable$ = defer(() => {
+      nInitCount += 1
+      return from([1, 2, 3, 4, 5])
+    })
+
+    const [useLatestNumber] = connectObservable(observable$, 0)
+    const { unmount } = renderHook(() => useLatestNumber())
+    const { unmount: unmount2 } = renderHook(() => useLatestNumber())
+    const { unmount: unmount3 } = renderHook(() => useLatestNumber())
+    expect(nInitCount).toBe(1)
+    unmount()
+    unmount2()
+    unmount3()
+
+    await act(async () => {
+      await wait(90)
+    })
+    const { unmount: unmount4 } = renderHook(() => useLatestNumber())
+    expect(nInitCount).toBe(1)
+    unmount4()
+
+    await act(async () => {
+      await wait(101)
+    })
+    renderHook(() => useLatestNumber())
+    expect(nInitCount).toBe(2)
+  })
+})

tsconfig.json

@@ -0,0 +1,30 @@
+{
+  "include": ["src", "types", "test"],
+  "compilerOptions": {
+    "target": "es5",
+    "module": "esnext",
+    "lib": ["dom", "esnext"],
+    "importHelpers": true,
+    "declaration": true,
+    "sourceMap": true,
+    "rootDir": "./src",
+    "strict": true,
+    "noImplicitAny": true,
+    "strictNullChecks": true,
+    "strictFunctionTypes": true,
+    "strictPropertyInitialization": true,
+    "noImplicitThis": true,
+    "alwaysStrict": true,
+    "noUnusedLocals": true,
+    "noUnusedParameters": true,
+    "noImplicitReturns": true,
+    "noFallthroughCasesInSwitch": true,
+    "moduleResolution": "node",
+    "baseUrl": "./",
+    "paths": {
+      "*": ["src/*", "node_modules/*"]
+    },
+    "jsx": "react",
+    "esModuleInterop": true
+  }
+}