version 2.2.0; 新增 StateObservable.ts; 使用 StateObservable 作为 novel 入参的类型

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@little-saga/rx-hooks",
-  "version": "2.1.0",
+  "version": "2.2.0",
   "description": "RxJS hooks for React and its friends.",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",

src/StateObservable.ts

@@ -0,0 +1,30 @@
+import { Observable, Subject } from 'rxjs'
+
+// StateObservable from redux-observable
+// https://github.com/redux-observable/redux-observable
+export default class StateObservable<S> extends Observable<S> {
+  value: S
+  private __notifier = new Subject<S>()
+
+  constructor(input$: Observable<S>, initialState: S) {
+    super(subscriber => {
+      const subscription = this.__notifier.subscribe(subscriber)
+      if (subscription && !subscription.closed) {
+        subscriber.next(this.value)
+      }
+      return subscription
+    })
+
+    this.value = initialState
+    input$.subscribe(value => {
+      // We only want to update state$ if it has actually changed since
+      // redux requires reducers use immutability patterns.
+      // This is basically what distinctUntilChanged() does but it's so simple
+      // we don't need to pull that code in
+      if (value !== this.value) {
+        this.value = value
+        this.__notifier.next(value)
+      }
+    })
+  }
+}

src/__tests__/novel.test.ts

@@ -2,13 +2,14 @@ import { act, renderHook } from '@testing-library/react-hooks'
 import { combineLatest, interval, isObservable, NEVER, Observable, Subscription } from 'rxjs'
 import { map, startWith, switchMap } from 'rxjs/operators'
 import { SubjectProxy } from '../helpers'
+import { StateObservable } from '../index'
 import { useNovel } from '../novel'
 import { applyMutatorAsReducer } from '../operators'
 
 describe('simpleCounterNovel', () => {
   function simpleCounterNovel(
-    input$: Observable<{ initCount: number }>,
-    state$: Observable<{ count: number }>,
+    input$: StateObservable<{ initCount: number }>,
+    state$: StateObservable<{ count: number }>,
   ) {
     const subscription = new Subscription()
     const actionProxy$ = new SubjectProxy<string>(subscription)

src/index.ts

@@ -2,3 +2,4 @@ export * from './equal-fns'
 export * from './helpers'
 export * from './novel'
 export * from './operators'
+export { default as StateObservable } from './StateObservable'

src/novel.ts

@@ -1,11 +1,12 @@
 import { useEffect, useMemo, useRef, useState } from 'react'
-import { BehaviorSubject, isObservable, Observable, Subscription } from 'rxjs'
+import { isObservable, Observable, Subject, Subscription } from 'rxjs'
+import StateObservable from './StateObservable'
 
 const NO_VALUE = Symbol('no-value')
 
 export type Novel<I, S extends object, D extends object, E> = (
-  input$: Observable<I>,
-  state$: Observable<S>,
+  input$: StateObservable<I>,
+  state$: StateObservable<S>,
 ) =>
   | Observable<S>
   | {
@@ -21,8 +22,8 @@ export function useNovel<I, S extends object, D extends object, E>(
   novel: Novel<I, S, D, E>,
 ) {
   const [state, setState] = useState(initialState)
-  const input$ = useMemo(() => new BehaviorSubject(input), [])
-  const state$ = useMemo(() => new BehaviorSubject(state), [])
+  const input$ = useMemo(() => new Subject<I>(), [])
+  const state$ = useMemo(() => new Subject<S>(), [])
 
   // 每次渲染之后更新 input$
   const mount = useRef(true)
@@ -37,56 +38,58 @@ export function useNovel<I, S extends object, D extends object, E>(
 
   const derivedValueRef = useRef<D>(null)
   const exportsRef = useRef<E>(null)
-  const ref = useRef<{
-    deriveSub: Subscription
-    stateSub: Subscription
-    teardown?(): void
-  }>({} as any)
+  const subscription = useMemo(() => new Subscription(), [])
 
   // 用 useMemo 来同步地执行 novel 与订阅 derived$，防止 derived 的初始值丢失
   useMemo(() => {
-    const output = novel(input$.asObservable(), state$.asObservable())
+    const output = novel(
+      new StateObservable(input$, input),
+      new StateObservable(state$, initialState),
+    )
     if (output == null) {
       return
     }
 
     if (isObservable(output)) {
-      ref.current.stateSub = output.subscribe(value => {
-        state$.next(value)
-        setState(value)
-      })
+      subscription.add(
+        output.subscribe(value => {
+          state$.next(value)
+          setState(value)
+        }),
+      )
     } else {
-      ref.current.stateSub = output.nextState?.subscribe(value => {
-        state$.next(value)
-        setState(value)
-      })
+      subscription.add(
+        output.nextState?.subscribe(value => {
+          state$.next(value)
+          setState(value)
+        }),
+      )
       if (output.derived) {
         let syncEmittedValue: D | typeof NO_VALUE = NO_VALUE
-        ref.current.deriveSub = output.derived.subscribe(value => {
-          derivedValueRef.current = value
-          /* istanbul ignore else */
-          if (process.env.NODE_ENV !== 'production') {
-            syncEmittedValue = value
-          }
-        })
+        subscription.add(
+          output.derived.subscribe(value => {
+            derivedValueRef.current = value
+            /* istanbul ignore else */
+            if (process.env.NODE_ENV !== 'production') {
+              syncEmittedValue = value
+            }
+          }),
+        )
         /* istanbul ignore else */
         if (process.env.NODE_ENV !== 'production') {
           if (syncEmittedValue === NO_VALUE) {
             throw new Error('derived$ must synchronously emit a value.')
           }
         }
       }
-      ref.current.teardown = output.teardown
+      subscription.add(output.teardown)
       exportsRef.current = output.exports
     }
   }, [])
 
   useEffect(() => {
     return () => {
-      state$.complete()
-      ref.current.teardown?.()
-      ref.current.deriveSub?.unsubscribe()
-      ref.current.stateSub?.unsubscribe()
+      subscription.unsubscribe()
     }
   }, [])