【源码学习】Radash（四）Async

1
import { all } from 'radash'
2

3
const [user] = await all([
4
  api.users.create(...),
5
  s3.buckets.create(...),
6
  slack.customerSuccessChannel.sendMessage(...)
7
])

1
import { all } from 'radash'
2

3
const { user } = await all({
4
  user: api.users.create(...),
5
  bucket: s3.buckets.create(...),
6
  message: slack.customerSuccessChannel.sendMessage(...)
7
})

1
import { defer } from 'radash'
2

3
await defer(async (cleanup) => {
4
  const buildDir = await createBuildDir()
5

6
  cleanup(() => fs.unlink(buildDir))
7

8
  await build()
9
})
10

11
await defer(async (register) => {
12
  const org = await api.org.create()
13
  register(async () => api.org.delete(org.id), { rethrow: true })
14

15
  const user = await api.user.create()
16
  register(async () => api.users.delete(user.id), { rethrow: true })
17

18
  await executeTest(org, user)
19
})

1
const users = (await guard(fetchUsers)) ?? []
2

3
// 您也可以选择仅保护特定错误
4
const isInvalidUserError = (err: any) => err.code === 'INVALID_ID'
5
const user = (await guard(fetchUser, isInvalidUserError)) ?? DEFAULT_USER

1
// guard 函数，用于捕获可能发生的异常，并根据条件判断是否需要处理这些异常
2
export const guard = <TFunction extends () => any>(
3
  // 接受两个参数，一个是函数 func，
4
  func: TFunction,
5
  // 另一个是可选的 shouldGuard 函数，用于判断是否需要进行异常处理
6
  shouldGuard?: (err: any) => boolean
7
  // 这的代码看得头大，拆一下就好：都是返回类型，先不管，后面再回过头来看这部分 MARK
8
): ReturnType<TFunction> extends Promise<any>
9
  ? Promise<Awaited<ReturnType<TFunction>> | undefined>
10
  : ReturnType<TFunction> | undefined => {
11

12
  // 定义了一个内部函数 _guard
13
  // 用于根据 shouldGuard 函数的返回值决定是否抛出异常
14
  const _guard = (err: any) => {
15
    if (shouldGuard && !shouldGuard(err)) throw err
16
    return undefined as any
17
  }
18

19
  // 定义了一个函数 isPromise，用于判断一个值是否为 Promise 类型
20
  const isPromise = (result: any): result is Promise<any> =>
21
    result instanceof Promise
22

23
  try {
24
    // 尝试执行传入的函数 func，根据返回值是 Promise 还是普通值来决定是直接返回值还是通过 catch 处理可能的异步异常
25
    const result = func()
26
    return isPromise(result) ? result.catch(_guard) : result
27
  } catch (err) {
28
    return _guard(err)
29
  }
30
  // 我们再看MARK部分：
31
  // 这里对函数返回类型进行了判断，如果是 Promise 类型，则返回一个包裹 Awaited<ReturnType<TFunction>> 类型的 Promise，否则直接返回函数的返回类型或者 undefined
32
}

1
import { map } from 'radash'
2

3
const userIds = [1, 2, 3, 4]
4

5
const users = await map(userIds, async (userId) => {
6
  return await api.users.find(userId)
7
})

1
export const map = async <T, K>(
2
  // 接受只读数组 array
3
  array: readonly T[],
4
  // 接受异步映射函数 asyncMapFunc，该函数将每个数组元素转换为另一种类型
5
  asyncMapFunc: (item: T, index: number) => Promise<K>
6
): Promise<K[]> => {
7
  // 如果传入的数组为空，则直接返回一个空数组
8
  if (!array) return []
9
  // 初始化一个空数组 result，用于存储映射后的结果
10
  let result = []
11
  // 初始化一个变量 index，用于跟踪当前处理的元素的索引
12
  let index = 0
13

14
  // 遍历数组中的每个元素，for of 循环遍历的是数组的值
15
  // 对每个元素调用异步映射函数 asyncMapFunc，并将映射后的结果存储到 result 数组中
16
  for (const value of array) {
17
    const newValue = await asyncMapFunc(value, index++)
18
    // 这里可结合例子来看，
19
    // newValue 是 asyncMapFunc 返回的结果，因为 asyncMapFunc 返回的是 Promise，所以这里需要使用 await 等待 Promise 完成
20
    // index在函数内定义了，有默认值，这里不需额外处理，这里直接存储到 result 数组中
21
    result.push(newValue)
22
  }
23
  // 返回映射后的结果数组 result
24
  return result
25
}

1
import { parallel, try as tryit } from 'radash'
2

3
const userIds1 = [1, 2, 3, 4, 5, 6, 7, 8, 9]
4

5
const users1 = await parallel(3, userIds1, async (userId) => {
6
  return await api.users1.find(userId)
7
})
8

9

10
// 错误处理
11
// 当所有工作完成后，并行将检查错误。如果发生任何错误，它们都将被抛出到一个 AggregateError 中，该 AggregateError 具有一个错误属性，该属性是抛出的所有错误。
12
const userIds = [1, 2, 3]
13

14
const [err, users] = await tryit(parallel)(3, userIds, async (userId) => {
15
  throw new Error(`No, I don\'t want to find user ${userId}`)
16
})
17

18
console.log(err) // => AggregateError
19
console.log(err.errors) // => [Error, Error, Error]
20
console.log(err.errors[1].message) // => No, I don't want to find user 2

1
import { reduce } from 'radash'
2

3
const userIds = [1, 2, 3, 4]
4

5
const users = await reduce(userIds, async (acc, userId) => {
6
  const user = await api.users.find(userId)
7
  return {
8
    ...acc,
9
    [userId]: user
10
  }
11
}, {})

1
export const reduce = async <T, K>(
2
  array: readonly T[],
3
  asyncReducer: (acc: K, item: T, index: number) => Promise<K>,
4
  initValue?: K
5
): Promise<K> => {
6
  // 检查是否提供了初始值，如果提供了， initProvided 为 true
7
  const initProvided = initValue !== undefined
8

9
  // 如果没有提供初始值，且数组为空，则抛出错误，因为无法对空数组进行相关操作
10
  if (!initProvided && array?.length < 1) {
11
    throw new Error('Cannot reduce empty array with no init value')
12
  }
13

14
  // 如果提供了初始值，则 iter 为原数组，否则 iter 为去掉第一个元素的数组
15
  const iter = initProvided ? array : array.slice(1)
16

17
  // 初始化 reduce 的结果值 value，如果提供了初始值，则为初始值，否则为数组的第一个元素
18
  let value: any = initProvided ? initValue : array[0]
19

20
  // 遍历 iter 中的每个元素，对 value 和当前元素应用异步 reduce 函数，更新 value 的值
21
  for (const [i, item] of iter.entries()) {
22
    value = await asyncReducer(value, item, i)
23
  }
24

25
  // 返回最终经过异步 reduce 后得到的结果值
26
  return value
27
}

1
import { sleep } from 'radash'
2

3
await sleep(2000) // => waits 2 seconds

1
// 表示要延迟的毫秒数
2
export const sleep = (milliseconds: number) => {
3
  // 返回一个 Promise 对象，这个 Promise 会在指定的毫秒数后自动 resolve（正常结束）。
4
  // setTimeout 函数用于在指定的毫秒数后执行一个函数，这里的 res 就是 Promise 的 resolve 函数，
5
  // 设置了等待的时间后自动触发 resolve，从而结束延迟状态。
6
  return new Promise(res => setTimeout(res, milliseconds))
7
}

1
import { retry } from 'radash'
2

3
await retry({}, api.users.list)
4
await retry({ times: 10 }, api.users.list)
5
await retry({ times: 2, delay: 1000 }, api.users.list)
6

7
// exponential backoff
8
await retry({ backoff: i => 10**i }, api.users.list)

1
export const retry = async <TResponse>(
2
  // options 是一个包含重试次数、延迟和退避函数的对象
3
  options: {
4
    times?: number
5
    delay?: number | null
6
    backoff?: (count: number) => number
7
  },
8
  // func 是包含退出错误函数的异步操作
9
  func: (exit: (err: any) => void) => Promise<TResponse>
10
): Promise<TResponse> => {
11

12
  // 获取重试次数，如果未提供，默认为3次
13
  const times = options?.times ?? 3
14

15
  // 获取延迟时间，如果未提供则为 undefined。这里可能会问上面不是定了 number | null吗？
16
  // 这里为可选项，没传，number、null是undefined的子类型，所以是 undefined
17
  const delay = options?.delay
18

19
  // 获取 backoff 函数，如果未提供则为 null
20
  const backoff = options?.backoff ?? null
21

22
  // 从1到设定的重试次数遍历
23
  for (const i of range(1, times)) {
24
    // 调用 tryit 函数以尝试执行传入的异步操作函数，并在发生错误时抛出一个包含错误信息的对象
25
    const [err, result] = (await tryit(func)((err: any) => {
26
      throw { _exited: err }
27
    })) as [any, TResponse]
28

29
    // 如果没有错误发生，直接返回结果
30
    if (!err) return result
31

32
    // 如果出错原因是程序退出，则抛出退出错误
33
    if (err._exited) throw err._exited
34

35
    // 如果当前重试次数达到设定的最大次数，则直接抛出错误
36
    if (i === times) throw err
37

38
    // 如果设置了延迟时间，等待相应的延迟
39
    if (delay) await sleep(delay)
40

41
    // 如果设置了退避函数，根据当前重试次数调整等待时间
42
    if (backoff) await sleep(backoff(i))
43
  }
44

45
  // 这段注释代码通常不应该执行到，用于满足函数的严格模式要求
46
  return undefined as unknown as TResponse
47
}

1
import { tryit } from 'radash'
2

3
const userId = 'xxx'
4
const [err, user] = await tryit(api.users.find)(userId)
5

6
// 也可以运用柯里化
7
const findUser = tryit(api.users.find)
8
const [err, user] = await findUser(userId)

1
export const tryit = <Args extends any[], Return>(
2
  // 定义了一个名为 tryit 的函数，它接受一个函数 func，该函数可以接受任意类型的参数并返回一个值
3
  func: (...args: Args) => Return
4
) => {
5
  return (
6
    // 这里的 ...args 表示接受任意数量的参数，参数的类型由 Args 泛型指定
7
    ...args: Args
8
  ): Return extends Promise<any>
9
    ? Promise<[Error, undefined] | [undefined, Awaited<Return>]>
10
    : [Error, undefined] | [undefined, Return] => {
11
    try {
12
      // 调用传入的函数 func 执行操作，并获取其返回结果
13
      const result = func(...args)
14

15
      // 判断函数返回的结果是否是一个 Promise 对象
16
      if (isPromise(result)) {
17
        // 如果结果是一个 Promise 对象，则返回一个 Promise，将成功时的值和失败时的错误进行封装
18
        return result
19
          .then(value => [undefined, value])
20
          .catch(err => [err, undefined]) as Return extends Promise<any>
21
          ? Promise<[Error, undefined] | [undefined, Awaited<Return>]>
22
          : [Error, undefined] | [undefined, Return]
23
      }
24

25
      // 如果结果不是 Promise 对象，则直接返回执行结果
26
      return [undefined, result] as Return extends Promise<any>
27
        ? Promise<[Error, undefined] | [undefined, Awaited<Return>]>
28
        : [Error, undefined] | [undefined, Return]
29
    } catch (err) {
30
      // 如果在执行中捕获到错误，返回一个包含错误信息的数组
31
      return [err as any, undefined] as Return extends Promise<any>
32
        ? Promise<[Error, undefined] | [undefined, Awaited<Return>]>
33
        : [Error, undefined] | [undefined, Return]
34
    }
35
  }
36
}