管理 RXJS 可观察对象中的承诺

Question

我已经研究过 SO 并发现了许多类似的问题/答案，但在我对如何使用此堆栈的基本理解中，我可能遗漏了一些东西。

我正在与 RXJS/obervables 一起开发一个 react native 项目。在某些时候我做文件下载，这部分不是问题。预先存在的axios-rxjs 和react-native-file-system 的组合让我到达我想要的地方。问题是我不确定如何在没有 async/await 的情况下干净地处理它，我知道这是一种反模式。

我想把这个工作代码转换成一个干净的 obervable 风格的流程。

我已经实现了一个 Epic 来执行我想要的操作：

const myDownloadEpic = (
  action$,
  state$
) =>
  action$.pipe(
    ofType(myDownloadActionType), // catches the relevant action
    map(action => action.payload),
    mergeMap(payload =>
      downloadManager       // a class with the relevant utils to get files
        .download(payload), // this axios call returns my Blob file as Observable<Blob> 
        .pipe(
          mergeMap(async response => {

            // a promise is returned by RNFS to read the contents of a folder
            const files = await RNFS.readDir(RELEVANT_TARGET_PATH) 
...
            // a promise returned from a custom function that converts my blob to bas64
            const data = await convertToBase64(response) 

            // another promise returned by RNFS to write to the file system
            await RNFS.writeFile(FULL_PATH, data, 'base64');

...
          })
        )
    )
   )

我尝试将其拆分为多个管道，例如，我尝试将 READ 操作拆分为前一个管道，但它最终看起来非常冗长。有没有一种干净简单的方法来“持有”直到承诺完成，以便我可以根据他们的结果做出决定？

在这种情况下什么会被认为是更清洁的？

Answer 1

你可以试试这样的。应该和上面写的差不多。

const myDownloadEpic = (
  action$,
  state$
) => action$.pipe(
  ofType(myDownloadActionType),
  map(action => action.payload),

  mergeMap(payload => downloadManager.download(payload)),

  mergeMap(response => from(RNFS.readDir(RELEVANT_TARGET_PATH)).pipe(
    map(files => ({response, files}))
  )),

  mergeMap(values => from(convertToBase64(values.response)).pipe(
    map(data => ({...values, data}))
  )),

  mergeMap(({response, files, data}) => RNFS.writeFile(FULL_PATH, data, 'base64'))
);

Answer 2

from() 运算符可以将 promise 转换为 observable，该 observable 将发出 promise 值然后完成。

如果您需要等到所有承诺都解决后，我推荐forkJoin()，因为在所有可观察对象完成之前它不会发出值。

最后，为了使代码更简洁，我还建议声明单独的变量/函数来为每个 Promise 定义你的 observables。

const files$ = from(RNFS.readDir(RELEVANT_TARGET_PATH));
const getData = (response: unknown) => from(convertToBase64(response)).pipe(
  mergeMap(data=>
    from(RNFS.writeFile(FULL_PATH, data, 'base64')).pipe(
      mapTo(data)
    )
  )
);

const myDownloadEpic = (action$, state$) =>
  action$.pipe(
    ofType(myDownloadActionType),
    map(({ payload }) => payload),
    mergeMap(payload => downloadManager.download(payload)),
    mergeMap(response =>
      forkJoin({
        files: files$,
        data: getData(response)
      })
    ),
    map(({ files, data }) => {
      // check something with `files` and `data`
    })
  );

我假设RNFS.writeFile() 的响应是void，所以我在订阅getData() 时将其作为效果。 mapTo() 运算符会忽略源 observable 发出的任何值，并返回您放入参数中的任何值。