lodash 已死？Radash 库方法介绍及源码解析 —— 异步方法篇

import { all } from 'radash'
// 传入promise数组const [user] = await all([  api.users.create(...),  s3.buckets.create(...),  slack.customerSuccessChannel.sendMessage(...)])
// 传入对象const { user } = await all({  user: api.users.create(...),  bucket: s3.buckets.create(...),  message: slack.customerSuccessChannel.sendMessage(...)})

// 定义一个泛型异步函数 `all`。export async function all<  // 泛型约束 `T` 可以是一个 `Promise` 数组或一个 `Promise` 对象。  T extends Record<string, Promise<any>> | Promise<any>[]>(promises: T) {  // 根据 `promises` 是数组还是对象，将其转换成一个统一格式的数组 `entries`。  const entries = isArray(promises)    ? promises.map(p => [null, p] as [null, Promise<any>])    : Object.entries(promises)
  // 使用 `Promise.all` 等待所有 `Promise` 完成，并处理每个 `Promise` 的结果和异常。  const results = await Promise.all(    entries.map(([key, value]) =>      value        .then(result => ({ result, exc: null, key })) // 如果成功，记录结果。        .catch(exc => ({ result: null, exc, key })) // 如果失败，记录异常。    )  )
  // 筛选出所有出现异常的结果。  const exceptions = results.filter(r => r.exc)  // 如果有异常，抛出一个 `AggregateError`，包含所有异常。  if (exceptions.length > 0) {    throw new AggregateError(exceptions.map(e => e.exc))  }
  // 如果输入的 `promises` 是数组，返回一个包含所有结果的数组。  if (isArray(promises)) {    return results.map(r => r.result) as T extends Promise<any>[]      ? PromiseValues<T>      : unknown  }
  // 如果输入的 `promises` 是对象，将结果组合成一个新对象并返回。  return results.reduce(    (acc, item) => ({      ...acc,      [item.key!]: item.result // 使用断言 `item.key!`，因为我们知道 `key` 不会是 `null`。    }),    {} as { [K in keyof T]: Awaited<T[K]> } // 返回类型是一个对象，其键类型为 `T` 的键，值类型为 `T` 中 `Promise` 解析后的类型。  )}

import { defer } from 'radash'
await defer(async (cleanup) => {  const buildDir = await createBuildDir()
  cleanup(() => fs.unlink(buildDir))
  await build()})
await defer(async (register) => {  const org = await api.org.create()  register(async () => api.org.delete(org.id), { rethrow: true })
  const user = await api.user.create()  register(async () => api.users.delete(user.id), { rethrow: true })
  await executeTest(org, user)})

// 定义一个异步泛型函数 `defer`。export const defer = async <TResponse>(  // `func` 是一个接受注册函数 `register` 的异步函数。  func: (    register: (      // `register` 允许 `func` 注册一个回调函数 `fn`，该函数在 `func` 执行完成后调用。      // 可以通过 `options` 指定是否在回调函数中重新抛出错误。      fn: (error?: any) => any,      options?: { rethrow?: boolean }    ) => void  ) => Promise<TResponse>): Promise<TResponse> => {  // 初始化一个用于存放回调函数及其选项的数组 `callbacks`。  const callbacks: {    fn: (error?: any) => any    rethrow: boolean  }[] = []
  // 实现注册函数 `register`，它将回调函数及其选项添加到 `callbacks` 数组。  const register = (    fn: (error?: any) => any,    options?: { rethrow?: boolean }  ) =>    callbacks.push({      fn,      rethrow: options?.rethrow ?? false    })
  // 调用 `tryit` 函数执行 `func`，并传入 `register`。  // `tryit` 函数不在提供的代码片段中，但我们可以假设它是一个错误处理函数，返回一个包含错误和响应的元组。  const [err, response] = await tryit(func)(register)
  // 遍历 `callbacks` 数组，依次执行每个回调函数。  for (const { fn, rethrow } of callbacks) {    // 使用 `tryit` 函数调用回调，以捕获并处理任何抛出的错误。    const [rethrown] = await tryit(fn)(err)    // 如果回调函数中有错误被重新抛出，并且 `rethrow` 选项为 `true`，则重新抛出该错误。    if (rethrown && rethrow) throw rethrown  }
  // 如果 `func` 执行时有错误产生，重新抛出该错误。  if (err) throw err  // 如果 `func` 执行成功，返回响应结果。  return response}

import { guard } from 'radash'
const users = (await guard(fetchUsers)) ?? []
const isInvalidUserError = (err: any) => err.code === 'INVALID_ID'const user = (await guard(fetchUser, isInvalidUserError)) ?? DEFAULT_USER

// 定义一个泛型函数 `guard`。export const guard = <TFunction extends () => any>(  // 参数 `func` 是一个无参数的函数，它可能返回任何类型的值，包括 `Promise`。  func: TFunction,  // 可选参数 `shouldGuard` 是一个函数，它接受一个错误对象 `err`，  // 并返回一个布尔值，指示是否应该 "guard" 这个错误。  shouldGuard?: (err: any) => boolean// 函数的返回类型依赖于 `func` 的返回类型。如果 `func` 返回一个 `Promise`，// 则 `guard` 返回一个 `Promise`，该 `Promise` 解析为 `func` 的返回值或 `undefined`。// 如果 `func` 不返回 `Promise`，则 `guard` 返回 `func` 的返回值或 `undefined`。): ReturnType<TFunction> extends Promise<any>  ? Promise<Awaited<ReturnType<TFunction>> | undefined>  : ReturnType<TFunction> | undefined => {  // 定义一个内部函数 `_guard`，它接受一个错误对象 `err`。  const _guard = (err: any) => {    // 如果提供了 `shouldGuard` 函数并且该函数返回 `false`，    // 表示不应该 "guard" 这个错误，则重新抛出该错误。    if (shouldGuard && !shouldGuard(err)) throw err    // 否则，返回 `undefined`。    return undefined as any  }
  // 定义一个类型守卫函数 `isPromise`，它检查一个值是否为 `Promise`。  const isPromise = (result: any): result is Promise<any> =>    result instanceof Promise
  try {    // 尝试执行 `func` 并获取结果。    const result = func()    // 如果 `result` 是一个 `Promise`，使用 `catch` 方法应用 `_guard` 函数。    // 否则，直接返回 `result`。    return isPromise(result) ? result.catch(_guard) : result  } catch (err) {    // 如果在执行 `func` 时抛出错误，使用 `_guard` 函数处理该错误。    return _guard(err)  }}

import { map } from 'radash'
const userIds = [1, 2, 3, 4]
const users = await map(userIds, async (userId) => {  return await api.users.find(userId)})

// 定义一个异步函数 `map`。export const map = async <T, K>(  // 第一个参数 `array` 是一个具有只读属性的泛型数组。  array: readonly T[],  // 第二个参数 `asyncMapFunc` 是一个异步映射函数，它接受一个数组元素和它的索引，  // 返回一个 `Promise`，该 `Promise` 解析为新类型 `K` 的值。  asyncMapFunc: (item: T, index: number) => Promise<K>): Promise<K[]> => {  // 如果传入的数组 `array` 不存在，则返回一个空数组。  if (!array) return []  // 初始化一个空数组 `result`，用于存放映射后的新值。  let result = []  // 初始化一个索引计数器 `index`。  let index = 0  // 使用 `for...of` 循环遍历数组 `array` 的每个元素。  for (const value of array) {    // 对每个元素调用 `asyncMapFunc` 映射函数，并等待其 `Promise` 解析。    const newValue = await asyncMapFunc(value, index++)    // 将解析后的新值添加到 `result` 数组中。    result.push(newValue)  }  // 循环完成后，返回包含所有新值的数组 `result`。  return result}

// 定义一个异步泛型函数 `parallel`。export const parallel = async <T, K>(  // `limit` 是一个数字，指定了可以同时运行的异步任务的最大数量。  limit: number,  // `array` 是一个只读数组，包含将要被异步处理的元素。  array: readonly T[],  // `func` 是一个函数，将数组中的每个元素转换为一个异步操作（返回 Promise）。  func: (item: T) => Promise<K>): Promise<K[]> => {  // 将数组 `array` 转换为包含元素和它们索引的对象的数组 `work`。  const work = array.map((item, index) => ({    index,    item  }))  // 定义一个处理函数 `processor`，它将异步处理 `work` 数组中的元素。  const processor = async (res: (value: WorkItemResult<K>[]) => void) => {    const results: WorkItemResult<K>[] = []    while (true) {      // 从 `work` 数组的末尾取出一个元素进行处理。      const next = work.pop()      // 如果没有更多元素，调用回调函数 `res` 并传入结果数组 `results`。      if (!next) return res(results)      // 使用 `tryit` 函数执行 `func` 并处理结果或错误。      const [error, result] = await tryit(func)(next.item)      // 将结果或错误添加到 `results` 数组中。      results.push({        error,        result: result as K,        index: next.index      })    }  }  // 创建一个 `queues` 数组，它包含了 `limit` 个新的 Promise，每个 Promise 都由 `processor` 函数处理。  const queues = list(1, limit).map(() => new Promise(processor))  // 使用 `Promise.all` 等待所有的 `queues` 中的 Promise 完成。  const itemResults = (await Promise.all(queues)) as WorkItemResult<K>[][]  // 将所有的结果扁平化并根据索引排序，然后使用 `fork` 函数将结果分为错误和成功的结果。  const [errors, results] = fork(    sort(itemResults.flat(), r => r.index),    x => !!x.error  )  // 如果有任何错误，抛出一个 `AggregateError` 包含所有错误。  if (errors.length > 0) {    throw new AggregateError(errors.map(error => error.error))  }  // 返回一个数组，它包含了按原数组顺序排序的所有成功的结果。  return results.map(r => r.result)}

// 定义一个异步泛型函数 `parallel`。export const parallel = async <T, K>(  // `limit` 是一个数字，指定了可以同时运行的异步任务的最大数量。  limit: number,  // `array` 是一个只读数组，包含将要被异步处理的元素。  array: readonly T[],  // `func` 是一个函数，将数组中的每个元素转换为一个异步操作（返回 Promise）。  func: (item: T) => Promise<K>): Promise<K[]> => {  // 将数组 `array` 转换为包含元素和它们索引的对象的数组 `work`。  const work = array.map((item, index) => ({    index,    item  }))  // 定义一个处理函数 `processor`，它将异步处理 `work` 数组中的元素。  const processor = async (res: (value: WorkItemResult<K>[]) => void) => {    const results: WorkItemResult<K>[] = []    while (true) {      // 从 `work` 数组的末尾取出一个元素进行处理。      const next = work.pop()      // 如果没有更多元素，调用回调函数 `res` 并传入结果数组 `results`。      if (!next) return res(results)      // 使用 `tryit` 函数执行 `func` 并处理结果或错误。      const [error, result] = await tryit(func)(next.item)      // 将结果或错误添加到 `results` 数组中。      results.push({        error,        result: result as K,        index: next.index      })    }  }  // 创建一个 `queues` 数组，它包含了 `limit` 个新的 Promise，每个 Promise 都由 `processor` 函数处理。  const queues = list(1, limit).map(() => new Promise(processor))  // 使用 `Promise.all` 等待所有的 `queues` 中的 Promise 完成。  const itemResults = (await Promise.all(queues)) as WorkItemResult<K>[][]  // 将所有的结果扁平化并根据索引排序，然后使用 `fork` 函数将结果分为错误和成功的结果。  const [errors, results] = fork(    sort(itemResults.flat(), r => r.index),    x => !!x.error  )  // 如果有任何错误，抛出一个 `AggregateError` 包含所有错误。  if (errors.length > 0) {    throw new AggregateError(errors.map(error => error.error))  }  // 返回一个数组，它包含了按原数组顺序排序的所有成功的结果。  return results.map(r => r.result)}

import { reduce } from 'radash'
const userIds = [1, 2, 3, 4]
const users = await reduce(userIds, async (acc, userId) => {  const user = await api.users.find(userId)  return {    ...acc,    [userId]: user  }}, {})

// 定义一个异步泛型函数 `reduce`。export const reduce = async <T, K>(  // 第一个参数 `array` 是一个只读数组，包含将要被归约处理的元素。  array: readonly T[],  // 第二个参数 `asyncReducer` 是一个异步归约函数，它接受累加值 `acc`、当前元素 `item` 和它的索引 `index`，  // 并返回一个 `Promise`，该 `Promise` 解析为新的累加值。  asyncReducer: (acc: K, item: T, index: number) => Promise<K>,  // 第三个参数 `initValue` 是可选的初始值。  initValue?: K): Promise<K> => {  // 检查初始值是否提供了。  const initProvided = initValue !== undefined  // 如果没有提供初始值且数组为空，则抛出错误。  if (!initProvided && array?.length < 1) {    throw new Error('Cannot reduce empty array with no init value')  }  // 如果提供了初始值，使用整个数组；否则，从数组的第二个元素开始迭代。  const iter = initProvided ? array : array.slice(1)  // 初始化累加值 `value`。如果提供了初始值，使用它；否则使用数组的第一个元素。  let value: any = initProvided ? initValue : array[0]  // 使用 `for...of` 循环和 `entries` 方法遍历数组或其子数组。  for (const [i, item] of iter.entries()) {    // 对每个元素调用异步归约函数 `asyncReducer` 并等待其 `Promise` 解析。    value = await asyncReducer(value, item, i)  }  // 循环完成后，返回最终的累加值 `value`。  return value}

import { retry } from 'radash'
await retry({}, api.users.list)await retry({ times: 10 }, api.users.list)await retry({ times: 2, delay: 1000 }, api.users.list)
// exponential backoffawait retry({ backoff: i => 10**i }, api.users.list)

// 定义一个异步泛型函数 `retry`。export const retry = async <TResponse>(  // `options` 对象包含重试策略的选项。  options: {    times?: number // 重试次数，默认为 3。    delay?: number | null // 固定延迟时间，如果提供，则在重试之间等待这么多毫秒。    backoff?: (count: number) => number // 退避函数，可以根据重试次数来计算延迟时间。  },  // `func` 是要执行的异步函数，它可能会失败。  func: (exit: (err: any) => void) => Promise<TResponse>): Promise<TResponse> => {  // 从 `options` 中获取重试次数、固定延迟时间和退避函数。  const times = options?.times ?? 3  const delay = options?.delay  const backoff = options?.backoff ?? null
  // 使用 `range` 函数生成一个序列，并遍历这个序列进行重试。  for (const i of range(1, times)) {    // 尝试执行 `func` 函数，并捕获可能的错误 `err` 和结果 `result`。    const [err, result] = (await tryit(func)((err: any) => {      // 如果 `func` 失败，并使用 `exit` 函数退出，则抛出一个特殊的错误对象。      throw { _exited: err }    })) as [any, TResponse]
    // 如果没有错误，说明 `func` 成功执行，返回结果。    if (!err) return result
    // 如果有特殊的退出错误，重新抛出原始错误。    if (err._exited) throw err._exited
    // 如果是最后一次重试且仍然失败，抛出错误。    if (i === times) throw err
    // 如果设置了固定延迟时间，使用 `sleep` 函数等待。    if (delay) await sleep(delay)
    // 如果提供了退避函数，根据重试次数计算延迟时间并等待。    if (backoff) await sleep(backoff(i))  }
  // 如果代码执行到这里，说明逻辑上不应该到达的代码路径。  // 这是为了满足 TypeScript 的严格模式要求。  /* istanbul ignore next */  return undefined as unknown as TResponse}

import { sleep } from 'radash'
await sleep(2000) // => waits 2 seconds

// 定义一个名为 `sleep` 的函数。export const sleep = (milliseconds: number) => {  // 返回一个新的 Promise。  return new Promise(res =>     // 使用 `setTimeout` 函数设置一个定时器，它在 `milliseconds` 指定的毫秒数后执行。    setTimeout(      // 当定时器到时，调用 `res` 函数来解析这个 Promise。      res,      // 传递给 `setTimeout` 的毫秒数，它决定了延时的长度。      milliseconds    )  )}

import { tryit } from 'radash'
const findUser = tryit(api.users.find)
const [err, user] = await findUser(userId)

// 定义一个泛型高阶函数 `tryit`。export const tryit = <Args extends any[], Return>(  // `func` 是一个接受任意参数的函数，其返回值可以是任何类型，包括 `Promise`。  func: (...args: Args) => Return) => {  // 返回一个新函数，这个新函数接受与 `func` 相同的参数。  return (    ...args: Args  // 新函数的返回类型取决于 `func` 的返回类型是否是 `Promise`。  // 如果 `func` 返回 `Promise`，则返回一个 `Promise`，包含一个错误或函数返回值的元组。  // 如果 `func` 返回非 `Promise`，则直接返回错误或函数返回值的元组。  ): Return extends Promise<any>    ? Promise<[Error, undefined] | [undefined, Awaited<Return>]>    : [Error, undefined] | [undefined, Return] => {    try {      // 尝试执行 `func` 并获取结果。      const result = func(...args)      // 使用辅助函数 `isPromise` 检查 `result` 是否是 `Promise`。      if (isPromise(result)) {        // 如果是 `Promise`，使用 `then` 和 `catch` 方法处理结果或捕获错误。        return result          .then(value => [undefined, value]) // 成功时返回值的元组。          .catch(err => [err, undefined]) // 错误时返回错误的元组。      }      // 如果 `result` 不是 `Promise`，直接返回值的元组。      return [undefined, result]    } catch (err) {      // 如果执行 `func` 时捕获到同步错误，返回错误的元组。      return [err as any, undefined]    }  }}