概念

effect

只要是被 yield 的值，都可以被称为 effect
业务代码往往充当 effect 的生产者，生成 effect
effect的类型主要有promise、iterator、take、put ，合理地组合不同类型的 effect 可以表达非常复杂的异步逻辑
effect 状态分为运行中、已完成（正常结束或是抛出错误结束都算完成）、被取消。

saga-middleware

yield 语句将 effect 传给 saga-middleware，saga-middleware 充当 effect 的消费者
获取 effect，根据 effect 的类型解释该 effect，然后将结果返回给生产者

模型

「请求-响应」模型，业务代码生产 effect 以发起「请求」，而 saga-middleware 负责消费并将响应返回给业务代码

saga

作为参数传入函数 proc 或 sagaMiddleware.run 执行的生成器函数
saga 实例指的是调用 saga 函数得到的迭代器对象

task

task 指的是 saga 实例运行状态的描述对象

  1. interface Task {
  2.   cancel(): void
  3.   toPromise(): Promise<any>
  4.   result: any
  5.   error: Error
  6.   isRunning: boolean
  7.   isCancelled: boolean
  8.   isAborted: boolean
  9. }

fork effect

saga 实例在运行的时候会多次 yield fork effect，那么一个 parent-saga 实例就会有多个 child-saga
rootSaga 通过 sagaMiddleware.run() 开始运行 fork 得到若干个 child-saga => saga 树

saga 树

迭代器自身的语句执行完成
所有的 child-saga 进入完成状态
迭代器自身执行时抛出了异常/其中一个 child-saga 抛出了错误=>saga 实例会中断并抛出错误
当一个节点发生错误时，错误会沿着树向根节点向上传播，直到某个节点捕获该错误
取消一个节点时，该节点对应的整个子树都将被取消

effect-runner

使用递归函数来消费迭代器 特点可以传入effect参数，可以同步或者异步调用next消费effect
通过驱动函数next消费effect

代码分析

  1. const iterator = range2(1, 10)
  2. // 驱动函数
  3. function next(arg, isErr) {
  4.   let result
  5.   if (isErr) {
  6.     result = iterator.throw(arg)
  7.   } else {
  8.     result = iterator.next(arg)
  9.   }
  10.   const { done, value } = result
  11.   if (done) {
  12.     return
  13.   }
  14.   // 根据effect的类型执行相应的处理逻辑
  15.   if (value[0] === "promise") {
  16.     const promise = value[1]
  17.     promise.then(
  18.       resolvedValue => next(resolvedValue),
  19.       error => next(error, true)
  20.     )
  21.   } else if (value[0] === "delay") {
  22.     const timeout = value[1]
  23.     setTimeout(() => next(`${timeout}ms elapsed`), timeout)
  24.   } else if (value[0] === "ping") {
  25.     next("pong")
  26.   } else {
  27.     iterator.throw(new Error("无法识别的 effect"))
  28.   }
  29. }
  30. next()
  32. // cancellation
  33. // effects 是可取消的
  35. // effect规范化处理
  36. // 每一个 effect 在运行之前都会通过函数 digestEffect 的处理。该函数用变量 effectSettled 记录了一个 effect 是否已经完成
  37. function digestEffect(rawEffect, cb) {
  38.   let effectSettled = false
  39.   // 通过callback函数及时得到effect执行状态，调用该函数，表明一旦完成或是被取消，就不能再调用cb改变状态
  40.   function currCb(res, isErr) {
  41.     if (effectSettled) {
  42.       return
  43.     }
  44.     effectSettled = true
  45.     cb.cancel = noop
  46.     cb(res, isErr)
  47.   }
  48.   currCb.cancel = noop
  49.   // 通过cb注册cancel函数，改变标识符的值，可以达到同样的目的，不再调用cb改变状态，实现effect取消功能
  50.   cb.cancel = () => {
  51.     if (effectSettled) {
  52.       return
  53.     }
  54.     effectSettled = true
  55.     try {
  56.       currCb.cancel()
  57.     } catch (err) {
  58.       console.error(err)
  59.     }
  60.     currCb.cancel = noop
  61.   }
  62.   // normalizeEffect用来内省识别effect类型，不需要开发者手动去标识effect类型
  63.   runEffect(normalizeEffect(rawEffect), currCb)
  64. }
  66. //************************************************ */
  67. // redux-saga
  68. const TASK_CANCEL = Symbol("TASK_CANCEL")
  69. const CANCEL = Symbol("CANCEL")
  71. // 所有 saga 实例都是通过该函数启动的，返回 task
  72. // 建立对象之间的后继关系（cont）和取消关系（cancellation）
  73. function proc(iterator, parentContext, cont) {
  75.   // 初始化当前 task 的 context
  76.   const ctx = Object.create(parentContext)
  78.   // mainTask 用来跟踪当前迭代器的语句执行状态
  79.   const mainTask = {
  80.     // cont: **will be set when passed to ForkQueue**
  81.     isRunning: true,
  82.     isCancelled: false,
  83.     cancel() {
  84.       if (mainTask.isRunning && !mainTask.isCancelled) {
  85.         mainTask.isCancelled = true
  86.         next(TASK_CANCEL)
  87.       }
  88.     },
  89.   } 
  90.   // 实际上task并不是这么构造的，不过在初步实现中，暂时先这样吧
  91.   // const task = {
  92.   //   cancel: () => next(TASK_CANCEL),
  93.   // }
  95.   // 创建 ForkQueue 对象和 Task 对象，这两个类的代码在后面会写出来
  96.   const taskQueue = new ForkQueue(mainTask)
  97.   const task = new Task(taskQueue)
  99.   // 设置后继关系
  100.   taskQueue.cont = task.end
  101.   task.cont = cont
  103.    // 设置取消关系
  104.    cont.cancel = task.cancel
  105.    next()
  107.   return task
  109.   // 在图中驱动函数只和 mainTask 有联系
  110.   // 然后我们也可以发现下面 next 函数的代码中，也只调用了 mainTask 的接口
  111.   // 即 next 函数中的代码不会引用 task 和 taskQueue 对象
  112.   function next(arg, isErr) {
  113.     try {
  114.       let result
  115.       if (isErr) {
  116.         result = iterator.throw(arg)
  117.       } else if (arg === TASK_CANCEL) {
  118.         // next.cancel 由当前正在执行的 effectRunner 所设置
  119.         next.cancel()
  120.         // 跳转到迭代器的 finally block，执行清理逻辑
  121.         result = iterator.return(TASK_CANCEL)
  122.       } else {
  123.         result = iterator.next(arg)
  124.       }
  126.       if (!result.done) {
  127.         digestEffect(result.value, next)
  128.       } else {
  129.         // 迭代器执行完毕，调用cont将结果返回给上层
  130.         mainTask.isRunning = false
  131.         mainTask.cont(result.value)
  132.       }
  133.     } catch (error) {
  134.       if (!mainTask.isRunning) {
  135.         throw error
  136.       }
  137.       if (mainTask.isCancelled) {
  138.         // 在执行 cancel 逻辑时发生错误，在 3.4 其他问题与细节 中说明
  139.         console.error(error)
  140.       }
  141.       mainTask.isRunning = false
  142.       mainTask.cont(error, true)
  143.     }
  144.   }
  146.   // function digestEffect(rawEffect, cb) { /* ...... */ }
  148.   // 执行effect，根据effect的类型调用不同的effectRunner
  149.   function runEffect(effect, currCb) {
  150.     const effectType = effect[0]
  151.     if (effectType === "promise") {
  152.       resolvePromise(effect, ctx, currCb)
  153.     } else if (effectType === "iterator") {
  154.       resolveIterator(iterator, ctx, currCb)
  155.     } else {
  156.       // 拓展这里的 if-else 便可以拓展新的effect类型
  157.       throw new Error("Unknown effect type")
  158.     }
  159.   }
  161.   function resolvePromise([effectType, promise], ctx, cb) {
  162.     const cancelPromise = promise[CANCEL]
  163.     if (is.func(cancelPromise)) {
  164.       // 设置promise的cancel逻辑
  165.       cb.cancel = cancelPromise
  166.     }
  167.     promise.then(cb, error => cb(error, true))
  168.   }
  170.   function resolveIterator([effectType, iterator], ctx, cb) {
  171.     proc(iterator, ctx, cb)
  172.   }
  173. }
  176. // https://zhuanlan.zhihu.com/p/37356948
  179. // fork effect
  180. function runForkEffect([effectType, fn, ...args], ctx, cb) {
  181.   const iterator = createTaskIterator(fn, args)
  182.   try {
  183.     suspend() // 见 3.4 scheduler
  184.     const subTask = proc(iterator, ctx, noop)
  185.     if (subTask.isRunning) {
  186.       task.taskQueue.addTask(subTask)
  187.       cb(subTask)
  188.     } else if (subTask.error) {
  189.       task.taskQueue.abort(subTask.error)
  190.     } else {
  191.       cb(subTask)
  192.     }
  193.   } finally {
  194.     flush() // 见 3.4 scheduler
  195.   }
  196. }
  198. interface ForkQueue {
  199.   // MainTask 该参数代表当前迭代器自身代码的执行状态
  200.   constructor(mainTask: MainTask)
  202.   // 在 ForkQueue 被构造之后进行设置。当所有的 child-task 以及 mainTask 都完成时，我们需要调用 forkQueue.cont 来通知其 parent-saga
  203.   // cont: Callback   这是一个私有的字段
  205.   // 添加新的 child-task ,赋值task.cont函数，task.cont调用实例方法cont返回结果值
  206.   addTask(task: Task): void
  207.   cancelAll(): void
  209.   // 取消所有的 child-task，还会调用 forkQueue.cont 向 parent-task 通知错误
  210.   abort(err: Error): void
  211. }
  213. // task context
  214. // 通过原型链的方式访问父节点的上下文
  215. // 利用 context 机制，并使用该机制实现「effect 类型拓展」、「连接 redux store」等功能
  217. // /src/core/Task.js
  218. class Task {
  219.   isRunning = true
  220.   isCancelled = false
  221.   isAborted = false
  222.   result = undefined
  223.   error = undefined
  224.   joiners = []
  226.   // cont will be set after calling constructor()
  227.   cont = undefined
  229.   constructor(taskQueue) {
  230.     this.taskQueue = taskQueue
  231.   }
  233.   // 调用 cancel 函数来取消该 Task，这将取消所有当前正在执行的 child-task 和 mainTask
  234.   // cancellation 会向下传播，意味着该 Task 对应的 saga-tree 子树都将会被取消
  235.   // 同时 cancellation 也会传递给该 Task 的所有 joiners
  236.   cancel = () => {
  237.     // 如果该 Task 已经完成或是已经被取消，则跳过
  238.     if (this.isRunning && !this.isCancelled) {
  239.       this.isCancelled = true
  240.       this.taskQueue.cancelAll()
  241.       // 将 TASK_CANCEL 传递给所有 joiners
  242.       this.end(TASK_CANCEL)
  243.     }
  244.   }
  246.   // 结束当前 Task
  247.   // 设置 Task 的 result/error，然后调用 task.cont，最后将结果传递给 joiners
  248.   // 当该 Task 的 child-task 和 mainTask 都完成时（即 fork-queue 完成时），该函数将被调用
  249.   end = (result, isErr) => {
  250.     this.isRunning = false
  251.     if (!isErr) {
  252.       this.result = result
  253.     } else {
  254.       this.error = result
  255.       this.isAborted = true
  256.     }
  258.     this.cont(result, isErr)
  259.     this.joiners.forEach(j => j.cb(result, isErr))
  260.     this.joiners = null
  261.   }
  263.   toPromise() {
  264.     // 获取 task 对应的 promise 对象，这里省略了代码
  265.   }
  266. }
  269. function createSagaMiddleware(cont) {
  270.   function middleware({ dispatch, getState }) {
  271.     let channelPut
  272.     const env = new Env(cont)
  273.       .use(commonEffects)
  274.       .use(channelEffects)
  275.       .use(ctx => {
  276.         // 记录「真实」的 channel.put
  277.         channelPut = ctx.channel.put
  279.         // 使用 dispatch 替换掉 channel 上的 put 方法
  280.         ctx.channel.put = action => {
  281.           action[SAGA_ACTION] = true
  282.           dispatch(action)
  283.         }
  285.         // 使用 def 方法来定义 select 类型的 effect-runner
  286.         def(ctx, 'select', ([_effectType, selector = identity, ...args], _ctx, cb) =>
  287.           cb(selector(getState(), ...args)),
  288.         )
  289.       })
  291.     // 当 middleware 函数执行时，说明 store 正在创建
  292.     // 此时我们给 middleware.run 设置正确的函数
  293.     middleware.run = (...args) => env.run(...args)
  295.     return next => action => {
  296.       const result = next(action) // hit reducers
  297.       // 下面的 if-else 主要是为了保证 channelPut(action) 恰好被包裹在一层 asap 中
  298.       // asap 的介绍见 3.4
  299.       if (action[SAGA_ACTION]) {
  300.         // SAGA_ACTION 字段为 true 表示该 action 来自 saga
  301.         // 而在 saga 中，我们在 put 的时候已经使用了函数asap
  302.         // 所以在这里就不需要再次调用 asap 了
  303.         channelPut(action)
  304.       } else {
  305.         // 表示该 action 来自 store.dispatch
  306.         // 例如某个 React 组件的 onClick 中调用了 dispatch 方法
  307.         asap(() => channelPut(action))
  308.       }
  309.       return result
  310.     }
  311.   }
  313.   middleware.run = (...args) => {
  314.     throw new Error('运行 Saga 函数之前，必须使用 applyMiddleware 将 Saga 中间件加载到 Store 中')
  315.   }
  317.   return middleware
  318. }