在Node.js TypeScript的fp-ts函数中如何正确使用await？

问题背景 #

我正在重构IoT设备的旧代码，希望用fp-ts更安全地处理错误，但缺乏函数式编程和fp-ts经验。设备需要定期推送本地临时数据库中的对象列表到后端，收到后端更新完成通知后清除本地对应条目。现在的核心问题是：在轮询逻辑里，调用await sleep(loopIntervalSleep)后递归调用函数的环节，如何把这个异步操作整合到fp-ts的pipe流程中？测试代码里的**await sleep(loopIntervalSleep)**是待处理的关键位置。

解决方案 #

方案1：将sleep包装为TaskEither #

首先要把异步的sleep操作转换成fp-ts的TaskEither类型，这样就能在pipe中用TE.flatMap串联任务。

修改后的完整代码：

import { pipe } from "fp-ts/lib/function"
import * as TE from "fp-ts/lib/TaskEither"
import sleep from 'sleep-promise'

const STATES = ['running', 'completed', 'failed']

const getUpdatesRandom = () => {
  const rr = Math.floor(Math.random() * 100)
  const res1 = async () => rr > 99 ? STATES[2] : rr > 70 ? STATES[1] : STATES[0]

  return TE.tryCatch(
    () => res1(),
    (err) => err instanceof Error
      ? err
      : new Error("获取更新状态时发生意外错误")
  )
}

// 把sleep包装成TaskRight，因为sleep不会抛出错误（除非被中断，此处忽略）
const sleepTE = (ms: number): TE.TaskEither<never, void> => 
  TE.rightTask(sleep(ms))

const checkStatus = async (loopIntervalSleep: number, numTimes: number) => {
  const checkStatusLoop = (times: number): TE.TaskEither<Error, string> => 
    pipe(
      getUpdatesRandom(),
      TE.flatMap((item) => {
        console.log('状态', item, times)
        switch (item) {
          case 'running':
            if (times > 0) {
              // 先执行sleep，再递归调用轮询函数
              return pipe(
                sleepTE(loopIntervalSleep),
                TE.flatMap(() => checkStatusLoop(times - 1))
              )
            } else {
              return TE.left(new Error('轮询次数超限'))
            }
          case 'completed':
            return TE.right('同步成功')
          default:
            return TE.left(new Error('调用失败'))
        }
      })
    )

  return checkStatusLoop(numTimes)()
}

const result = await checkStatus(5000, 5)
console.log(result)

方案2：简化递归逻辑的写法 #

如果想让代码更简洁，可以把轮询的分支逻辑进一步提炼，减少嵌套pipe：

// 仅修改checkStatusLoop部分，其余代码同方案1
const checkStatusLoop = (times: number): TE.TaskEither<Error, string> => 
  pipe(
    getUpdatesRandom(),
    TE.flatMap((item) => {
      console.log('状态', item, times)
      if (item === 'running') {
        return times > 0 
          ? pipe(sleepTE(loopIntervalSleep), TE.flatMap(() => checkStatusLoop(times - 1)))
          : TE.left(new Error('轮询次数超限'))
      } else if (item === 'completed') {
        return TE.right('同步成功')
      } else {
        return TE.left(new Error('调用失败'))
      }
    })
  )

关键说明 #

• 异步操作容器化：用TE.rightTask把普通Promise（sleep）转换成TaskEither，由于sleep本身不会抛出错误，所以错误类型设为never，明确表示这个任务不会失败。
• 流程串联逻辑：在需要等待后递归的场景，用TE.flatMap将sleep任务和下一轮轮询任务串联，整个流程保持TaskEither的类型一致性，所有异步操作都被包裹在fp-ts的容器中。
• 统一错误处理：轮询过程中的所有错误（包括状态获取异常、轮询超限、调用失败）都会被捕获到TaskEither的Left分支，避免了未处理的Promise拒绝，实现了你想要的安全错误处理目标。

内容的提问来源于stack exchange，提问作者colinbes