相当于 rxjs 中可观察到的 redux 'getState'

Question

我正在玩 rxjs，目的是掌握 JS 中的反应式和函数式编程。

我有以下琐碎的应用程序：

import Rx from 'rxjs'

const domContainer = document.querySelector('[data-container]')

const clickElement = document.querySelector('[data-init-fetch]')
const loader = document.querySelector('.loader')

// only emit when all the data is returned
const fetchData = (first, last) => Rx.Observable.range(first, last)
  .flatMap(async n => {
    const res = await fetch(`https://swapi.co/api/people/${n}`)
    return await res.json()
  })
  .reduce((acc, item) => [...acc, item], [])

const createEventStream = ({
  domNode,
  createRequestDataAction,
  createReceiveDataAction,
}) => {

  const handleButtonClick = Rx.Observable.fromEvent(domNode, 'click')

  const request = fetchData(1, 10)
  // i'd like to be be able to 'getState' in here to use correct offset and 
  // fetch the next ten entries

  const setIsFetching = handleButtonClick
    .map(createRequestDataAction)

  // fetch data on request
  const requestData = handleButtonClick
    .switchMapTo(request)
    .map(createReceiveDataAction)

  return Rx.Observable.merge(
    requestData,
    setIsFetching,
  )
}

const initState = {
  offset: 1,
  isFetching: false,
  chars: {},
}

const eventStream = createEventStream({
  domNode: clickElement,
  createReceiveDataAction,
  createRequestDataAction,
})
  .scan(reducer, initState)
  .subscribe(
    renderMarkup,
  )

function reducer (
  state = initState,
  {
    type,
    payload,
  }
) {
  // do some reducery stuff in here and return new state
}

function renderMarkup(state) {
  // renders the state to the DOM
}

function createReceiveDataAction(data) {
  return {
    type: 'apiReceive',
    payload: {
      chars: data,
    },
  }
}

function createRequestDataAction() {
  return { type: 'apiRequest' }
  // this action will update offset += 10 in the reducer
}

我的问题是，如果我在调度 fetchData 操作时使用 redux（虽然我不在这里），我只需调用 getState() 从当前状态获取 offset 并在 @987654323 中使用它@动作。

有没有办法在上面的代码中使用 rxjs（并且没有 redux）来做到这一点？例如，我想在fetchData 调用中使用状态变量。

我是否需要创建另一个 Observable 来处理offset？还是我以完全错误的方式处理这个问题？

为清晰起见进行了编辑。

Answer 1

我不太清楚你想要达到什么目的。

我有几个可以帮助你的 cmets。

使用 Observable 代替 Promise

您可以转换此代码

const fetchData = (first, last) => Rx.Observable.range(first, last)
  .flatMap(async n => {
    const res = await fetch(`https://swapi.co/api/people/${n}`)
    return await res.json()
  })
  .reduce((acc, item) => [...acc, item], [])

到这里

const fetchData = (first, last) => Rx.Observable.range(first, last)
  .mergeMap(n => {   // mergeMap is the new name of flatMap
    const promiseCall = fetch(`https://swapi.co/api/people/${n}`)
    return Observable.fromPromise(promiseCall).map(res => res.json())
  })
  .reduce((acc, item) => [...acc, item], [])

您也可以查看this question 以找到其他方法来执行许多 http 请求并仅在所有请求都完成后返回。

在 eventStream 中存储订阅

如果您运行以下代码，您最终会将订阅存储在变量 eventStream 中。然后可以使用 Subcription 实例取消订阅。

const eventStream = createEventStream({
  domNode: clickElement,
  createReceiveDataAction,
  createRequestDataAction,
})
  .scan(reducer, initState)
  .subscribe(
    renderMarkup,
  )

在 fetchData 调用中使用状态信息

我假设您将状态信息存储在 state 变量引用的对象中。

如果这是正确的，我会按照这些思路进行操作

const createEventStream = ({
  domNode,
  createRequestDataAction,
  createReceiveDataAction,
}) => {

  Rx.Observable.fromEvent(domNode, 'click')
    .map(createRequestDataAction)
    .concat(fetchData(initState.offset, initState.offset + 10))
    .map(createReceiveDataAction)

}

您甚至可以考虑使用类似这样的方式删除 reduce 函数。

const createEventStream = ({
  domNode
}) => {

  Rx.Observable.fromEvent(domNode, 'click')
    .do(_ => state.fetching = true)
    .concat(fetchData(initState.offset, initState.offset + 10))
    .do(_ => initState.offset = initState.offset + 10)
    .do(_ => initState.fetching = false)

}

我无法测试我的建议，所以我不确定是否有错误。