实现 take、put

https://github.com/redux-saga/redux-saga/blob/master/packages/core/src/internal/runSaga.js

redux-saga/effectTypes.js

  1. export const TAKE = 'TAKE'; //监听
  2. export const PUT = 'PUT'; //派发

redux-saga/effects.js

  1. import * as effectTypes from './effectTypes';
  3. /** 监听某个动作，生成指令对象的工厂函数
  4.  * 
  5.  * @param {*} actionType 
  6.  */
  7. export function take(actionType){
  8.   // 这是一个用来发送给saga中间件的指令对象
  9.   return {type: effectTypes.TAKE, actionType};
  10. }
  12. /** 派发一个真正的动作
  13.  * 
  14.  * @param {*} action 
  15.  */
  16. export function put(action){
  17.   return {type: effectTypes.PUT, action};
  18. }

redux-saga/index.js

  1. import runSaga from './runSaga';  //驱动根saga自动执行，类似co
  2. import stdChannel from './channel'; // 管道 channel 用于实现我们的发布订阅
  4. function createSagaMiddleware(){
  5.   const channel = stdChannel();
  6.   let boundRunSaga;
  8.   // 创建 saga 中间件
  9.   function sagaMiddleware({getState, dispatch}){
  10.     boundRunSaga = runSaga.bind(null, {channel, dispatch, getState});
  12.     return function(next){
  13.       return function(action){
  14.         const result = next(action);
  15.         //把action也给channel一份，如果有channel订阅的动作派发，就会触发相应的监听函数
  16.         channel.emit(action);
  17.         return result;
  18.       }
  19.     }
  20.   }
  22.   sagaMiddleware.run = (saga) => boundRunSaga(saga);
  23.   return sagaMiddleware;
  24. }
  26. export default createSagaMiddleware;

redux-saga/runSaga.js

  1. import * as effectTypes from './effectTypes';
  3. /** runSaga
  4.  * 
  5.  * @param {*} env {channel, dispatch, getState}
  6.  * @param {*} saga 可能是一个生成器，也可能是一个迭代器
  7.  */
  8. function runSaga(env, saga){
  9.   let {channel, dispatch, getState} = env;
  10.   let it = saga();
  12.   function next(val){
  13.     let {value: effect, done} = it.next(val);
  14.     if (!done){
  15.       // 根据不同的effect进行不同的处理
  16.       switch (effect.type){
  17.         case effectTypes.TAKE:
  18.           // effect = {type: 'TAKE', actionType: 'ASYNC_ADD'}
  19.           // 添加订阅，如果有人派发了 effect.actionType 这个动作类型，就执行next继续走
  20.           // 如果没有，就暂停，阻塞在这里
  21.           channel.once(effect.actionType, next);  
  22.           break;
  23.         case effectTypes.PUT:
  24.           // effect = {type: 'PUT', action: {type: 'ADD'}}
  25.           // put 不阻塞，派发action之后就继续向下执行
  26.           dispatch(effect.action);
  27.           next(effect.action);
  28.         default: 
  29.         break;
  30.       }
  31.     }
  32.   }
  33.   next();
  34. }
  36. export default runSaga;

redux-saga/channel.js

  1. function stdChannel(){
  2.   let listeners = [];  //缓存 监听函数
  4.   /** once（在源码中名字叫take） 订阅，保存监听函数
  5.    * 
  6.    * @param {*} actionType 等待的动作类型
  7.    * @param {*} listener 等到之后，执行的监听函数
  8.    */
  9.   function once(actionType, listener){
  10.     listener.actionType = actionType;
  11.     // 设置取消监听函数，用于单次监听（即只触发一次，触发后自动移除）
  12.     listener.cancel = () => {
  13.       listeners = listeners.filter(item => item !== listener);
  14.     }
  15.     listeners.push(listener);
  16.   }
  18.   /** emit（在源码中名字叫put） 发布，执行监听函数
  19.    * 
  20.    * @param {*} action 派发的动作
  21.    */
  22.   function emit(action){
  23.     listeners.forEach(listener => {
  24.       if (listener.actionType === action.type){
  25.         // 先取消，再执行，这样就实现了多次发布也只会触发一次监听函数。
  26.         // 类似于 eventEmitter.once (单次监听器，触发过一次后，就被自动移除了)
  27.         listener.cancel(); 
  28.         listener(action);
  29.       }
  30.     })
  31.   }
  33.   return {once, emit};
  34. }
  35. export default stdChannel;
  38. /** 使用示例
  39.   let channel = stdChannel(); //先实例化
  40.   // 订阅
  41.   channel.on('ASYNC_ADD', action => {
  42.     console.log('action=>', action);
  43.   })
  44.   // 发布。不管派发几次，都只会执行一次
  45.   channel.emit({type: 'ASYNC_ADD', payload: {a: 1} });
  46.   channel.emit({type: 'ASYNC_ADD', payload: {a: 2} });
  47.   // action=> { type: 'ASYNC_ADD', payload: { a: 1 } }
  49. */

支持产出iterator

redux-saga/runSaga.js

  1. import * as effectTypes from './effectTypes';
  3. /** runSaga
  4.  * 
  5.  * @param {*} env {channel, dispatch, getState}
  6.  * @param {*} saga 可能是一个Generator生成器，也可能是一个Generator迭代器
  7.  */
  8. function runSaga(env, saga){
  9.   let {channel, dispatch, getState} = env;
  10. + let it = typeof saga === 'function' ? saga() : saga;
  12.   function next(val){
  13.     let {value: effect, done} = it.next(val);
  15.     if (!done){
  16. +     // 如果 effect产出 是 Generator 迭代器
  17. +     if (typeof effect[Symbol.iterator] === 'function'){
  18. +       // 开始一个子进程执行 worker saga
  19. +       // 此处的“开启子进程”只是一个比喻，并不是真的开了一个子进程，只是它的表现很像。
  20. +       runSaga(env, effect); 
  21. +       next(); //调用next，让当前saga继续往下走（子saga内部是否阻塞对其无影响）
  22. +       break;
  23. +     } else {
  24.         // 根据不同的effect进行不同的处理
  25.         switch (effect.type){
  26.           case effectTypes.TAKE:
  27.             // effect = {type: 'TAKE', actionType: 'ASYNC_ADD'}
  28.             // 添加订阅，如果有人派发了 effect.actionType 这个动作类型，就执行next，继续走；
  29.             // 如果没有，就暂停，阻塞在这里
  30.             channel.once(effect.actionType, next);  
  31.             break;
  32.           case effectTypes.PUT:
  33.             // effect = {type: 'PUT', action: {type: 'ADD'}}
  34.             // put 派发action之后就继续向下执行
  35.             dispatch(effect.action);
  36.             next(effect.action);
  37.           default: 
  38.           break;
  39.         }
  40.       } 
  41.     }
  42.   }
  43.   next();
  44. }
  46. export default runSaga;

实现 takeEvery

PS: while 遇到yield 不会阻塞主进程

  1. function* gen(){
  2.   while (true){
  3.     yield 1;
  4.     yield 2;
  5.   }
  6. }
  7. setInterval(() => {
  8.   console.log(new Date());
  9. }, 1000);
  11. let it = gen();
  12. it.next();
  13. it.next();
  14. it.next();
  15. // 时间依然输出，说明 这里的while不会阻塞主进程。因为它遇到yield，会放弃执行权

redux-saga/effectTypes.js

  1. export const TAKE = 'TAKE'; //监听
  2. export const PUT = 'PUT'; //派发
  3. + export const FORK = 'FORK'; //fork出一个子进程

redux-saga/effects.js

  1. /** 开启一个子进程运行saga，为了不阻塞当前的进程
  2.  * 
  3.  * @param {*} saga Generator生成器
  4.  * @param {*} args 额外参数
  5.  */
  6. export function fork(saga, ...args){
  7.   return {type: effectTypes.FORK, saga, args};
  8. }
  10. /** 当监听到某个动作类型时，开启新的子进程执行saga
  11.  * 
  12.  * @param {*} actionType 动作类型
  13.  * @param {*} saga 
  14.  */
  15. export function takeEvery(actionType, saga){
  16.   function* takeEveryHelper(){
  17.     // while 不会阻塞主进程，因为它遇到 yield后，就会放弃执行权。
  18.     while (true){
  19.       const action = yield take(actionType); //监听动作派发
  20.       // yield fork(saga, action); //开启一个新的子进程执行saga
  21.       yield fork(function* (){
  22.         yield saga(action);
  23.       })
  24.     }
  25.   }
  26.   return fork(takeEveryHelper);
  27. }

redux-saga/runSaga.js

  1. import * as effectTypes from './effectTypes';
  3. /** runSaga
  4.  * 
  5.  * @param {*} env {channel, dispatch, getState}
  6.  * @param {*} saga 可能是一个Generator生成器，也可能是一个Generator迭代器
  7. +* @param {*} args 额外参数
  8.  */
  9. + function runSaga(env, saga, ...args){
  10.   let {channel, dispatch, getState} = env;
  11. + let it = typeof saga === 'function' ? saga() : saga;
  13.   function next(val){
  14.     let {value: effect, done} = it.next(val);
  16.     if (!done){
  17.       // 如果 effect产出 是 Generator 迭代器
  18.       if (typeof effect[Symbol.iterator] === 'function'){
  19.         // 开始一个子进程执行 worker saga
  20.         // 此处的“开启子进程”只是一个比喻，并不是真的开了一个子进程，只是它的表现很像。
  21.         runSaga(env, effect); 
  22.         next(); //调用next，让当前saga继续往下走（子saga内部是否阻塞对其无影响）
  23.       } else {
  24.         // 根据不同的effect进行不同的处理
  25.         switch (effect.type){
  26.           case effectTypes.TAKE:
  27.             // effect = {type: 'TAKE', actionType: 'ASYNC_ADD'}
  28.             // 添加订阅，如果有人派发了 effect.actionType 这个动作类型，就执行next，继续走；
  29.             // 如果没有，就暂停，阻塞在这里
  30.             channel.once(effect.actionType, next);  
  31.             break;
  32.           case effectTypes.PUT:
  33.             // effect = {type: 'PUT', action: {type: 'ADD'}}
  34.             // put 派发action之后就继续向下执行
  35.             dispatch(effect.action);
  36.             next(effect.action);
  37.             break;
  38. +         case effectTypes.FORK:
  39. +           runSaga(env, effect.saga, ...effect.args);
  40. +           next();
  41. +           break;
  42.           default: 
  43.           break;
  44.         }
  45.       } 
  46.     }
  47.   }
  48.   next();
  49. }
  50. export default runSaga;

支持Promise

redux-saga/runSaga.js

  1. import * as effectTypes from './effectTypes';
  3. /** runSaga
  4.  * 
  5.  * @param {*} env {channel, dispatch, getState}
  6.  * @param {*} saga 可能是一个Generator生成器，也可能是一个Generator迭代器
  7.  * @param {*} args 额外参数
  8.  */
  9. function runSaga(env, saga, ...args){
  10.   let {channel, dispatch, getState} = env;
  11.   let it = typeof saga === 'function' ? saga() : saga;
  13.   function next(val){
  14.     let {value: effect, done} = it.next(val);
  16.     if (!done){
  17.       // 如果 effect产出 是 Generator 迭代器
  18.       if (typeof effect[Symbol.iterator] === 'function'){
  19.         // 开始一个子进程执行 worker saga
  20.         // 此处的“开启子进程”只是一个比喻，并不是真的开了一个子进程，只是它的表现很像。
  21.         runSaga(env, effect); 
  22.         next(); //调用next，让当前saga继续往下走（子saga内部是否阻塞对其无影响）
  23. +     } else if (typeof effect.then === 'function'){
  24. +       effect.then(next);
  25. +     } else{
  26.         // 根据不同的effect进行不同的处理
  27.         switch (effect.type){
  28.           case effectTypes.TAKE:
  29.             // effect = {type: 'TAKE', actionType: 'ASYNC_ADD'}
  30.             // 添加订阅，如果有人派发了 effect.actionType 这个动作类型，就执行next，继续走；
  31.             // 如果没有，就暂停，阻塞在这里
  32.             channel.once(effect.actionType, next);  
  33.             break;
  34.           case effectTypes.PUT:
  35.             // effect = {type: 'PUT', action: {type: 'ADD'}}
  36.             // put 不阻塞，派发action之后就继续向下执行
  37.             dispatch(effect.action);
  38.             next(effect.action);
  39.             break;
  40.           case effectTypes.FORK:
  41.             runSaga(env, effect.saga, ...effect.args);
  42.             next();
  43.             break;
  44.           default: 
  45.           break;
  46.         }
  47.       } 
  48.     }
  49.   }
  50.   next();
  51. }
  53. export default runSaga;

实现 call、delay

redux-saga/effectTypes.js

  1. export const TAKE = 'TAKE'; //监听
  2. export const PUT = 'PUT'; //派发
  3. export const FORK = 'FORK'; //fork出一个子进程
  4. + export const CALL = 'CALL'; //处理一个 返回值为Promise的函数fn

redux-saga/effects.js

  1. /** 处理一个 返回值为Promise的函数fn
  2.  * 
  3.  * @param {*} fn 函数
  4.  * @param  {...args} args 函数参数
  5.  */
  6. export function call(fn, ...args){
  7.   return {type: effectTypes.CALL, fn, args};
  8. }
  10. // 延迟
  11. function delayP(ms, val=true){
  12.   const promise = new Promise(resolve => {
  13.     setTimeout(resolve, ms, val);
  14.   })
  15.   return promise;
  16. }
  17. export const delay = call.bind(null, delayP);
  18. // delay(fn, 1000) => {type: 'CALL', fn:delayP, args} args=[1000]

redux-saga/runSaga.js

  1. import * as effectTypes from './effectTypes';
  3. /** runSaga
  4.  * 
  5.  * @param {*} env {channel, dispatch, getState}
  6.  * @param {*} saga 可能是一个Generator生成器，也可能是一个Generator迭代器
  7.  * @param {*} args 额外参数
  8.  */
  9. function runSaga(env, saga, ...args){
  10.   let {channel, dispatch, getState} = env;
  11.   let it = typeof saga === 'function' ? saga() : saga;
  13.   function next(val){
  14.     let {value: effect, done} = it.next(val);
  16.     if (!done){
  17.       // 如果 effect产出 是 Generator 迭代器
  18.       if (typeof effect[Symbol.iterator] === 'function'){
  19.         // 开始一个子进程执行 worker saga
  20.         // 此处的“开启子进程”只是一个比喻，并不是真的开了一个子进程，只是它的表现很像。
  21.         runSaga(env, effect); 
  22.         next(); //调用next，让当前saga继续往下走（子saga内部是否阻塞对其无影响）
  23.       } else if (typeof effect.then === 'function'){
  24.         effect.then(next);
  25.       } else{
  26.         // 根据不同的effect进行不同的处理
  27.         switch (effect.type){
  28.           case effectTypes.TAKE:
  29.             // effect = {type: 'TAKE', actionType: 'ASYNC_ADD'}
  30.             // 添加订阅，如果有人派发了 effect.actionType 这个动作类型，就执行next，继续走；
  31.             // 如果没有，就暂停，阻塞在这里
  32.             channel.once(effect.actionType, next);  
  33.             break;
  34.           case effectTypes.PUT:
  35.             // effect = {type: 'PUT', action: {type: 'ADD'}}
  36.             // put 不阻塞，派发action之后就继续向下执行
  37.             dispatch(effect.action);
  38.             next(effect.action);
  39.             break;
  40.           case effectTypes.FORK:
  41.             runSaga(env, effect.saga, ...effect.args);
  42.             next();
  43.             break;
  44. +         case effectTypes.CALL:
  45. +           effect.fn(...effect.args).then(next);
  46. +           break;
  47.           default: 
  48.           break;
  49.         }
  50.       } 
  51.     }
  52.   }
  53.   next();
  54. }
  56. export default runSaga;

实现 cps

redux-saga/effectTypes.js

  1. export const TAKE = 'TAKE'; //监听
  2. export const PUT = 'PUT'; //派发
  3. export const FORK = 'FORK'; //fork出一个子进程
  4. export const CALL = 'CALL'; //处理一个 返回值为Promise的函数fn
  5. + export const CPS = 'CPS'; //处理一个 回调形式的函数

redux-saga/effects.js

  1. /** 处理一个 回调形式的函数
  2.  * 
  3.  * @param {*} fn 函数
  4.  * @param  {...args} args 函数参数
  5.  */
  6. export function cps(fn, ...args){
  7.   return {type: effectTypes.CPS, fn, args};
  8. }

redux-saga/runSaga.js

  1. import * as effectTypes from './effectTypes';
  3. /** runSaga
  4.  * 
  5.  * @param {*} env {channel, dispatch, getState}
  6.  * @param {*} saga 可能是一个Generator生成器，也可能是一个Generator迭代器
  7.  * @param {*} args 额外参数
  8.  */
  9. function runSaga(env, saga, ...args){
  10.   let {channel, dispatch, getState} = env;
  11.   let it = typeof saga === 'function' ? saga() : saga;
  13. + function next(val, isError){
  14. +   let result;
  15. +   if (isError){
  16. +     result = it.throw(val);    //立刻停止saga，表示出错了
  17. +   } else {
  18. +     it.next(val);
  19. +   }
  20. +   let {value: effect, done} = result;
  22.     if (!done){
  23.       // 如果 effect产出 是 Generator 迭代器
  24.       if (typeof effect[Symbol.iterator] === 'function'){
  25.         // 开始一个子进程执行 worker saga
  26.         // 此处的“开启子进程”只是一个比喻，并不是真的开了一个子进程，只是它的表现很像。
  27.         runSaga(env, effect); 
  28.         next(); //调用next，让当前saga继续往下走（子saga内部是否阻塞对其无影响）
  29.       } else if (typeof effect.then === 'function'){
  30.         effect.then(next);
  31.       } else{
  32.         // 根据不同的effect进行不同的处理
  33.         switch (effect.type){
  34.           case effectTypes.TAKE:
  35.             // effect = {type: 'TAKE', actionType: 'ASYNC_ADD'}
  36.             // 添加订阅，如果有人派发了 effect.actionType 这个动作类型，就执行next，继续走；
  37.             // 如果没有，就暂停，阻塞在这里
  38.             channel.once(effect.actionType, next);  
  39.             break;
  40.           case effectTypes.PUT:
  41.             // effect = {type: 'PUT', action: {type: 'ADD'}}
  42.             // put 不阻塞，派发action之后就继续向下执行
  43.             dispatch(effect.action);
  44.             next(effect.action);
  45.             break;
  46.           case effectTypes.FORK:
  47.             runSaga(env, effect.saga, ...effect.args);
  48.             next();
  49.             break;
  50.           case effectTypes.CALL:
  51.             effect.fn(...effect.args).then(next);
  52.             break;
  53. +         case effectTypes.CPS:
  54. +           effect.fn(...effect.args, (err, data) => {
  55. +             err ? next(err, true) : next(data);
  56. +           });
  57. +           break;
  58.           default: 
  59.           break;
  60.         }
  61.       } 
  62.     }
  63.   }
  64.   next();
  65. }
  67. export default runSaga;

实现 all

redux-saga/effectTypes.js

  1. export const TAKE = 'TAKE'; //监听
  2. export const PUT = 'PUT'; //派发
  3. export const FORK = 'FORK'; //fork出一个子进程
  4. export const CALL = 'CALL'; //处理一个 返回值为Promise的函数fn
  5. export const CPS = 'CPS'; //处理一个 回调形式的函数
  6. + export const ALL = 'ALL';

redux-saga/effects.js

  1. /**
  2.  * 
  3.  * @param {*} effects 数组，里面放置一个个的 监听saga迭代器
  4.  */
  5. export function all(effects){
  6.   return {type: effectTypes.ALL, effects};
  7. }

redux-saga/runSaga.js

  1. import * as effectTypes from './effectTypes';
  3. /** runSaga
  4.  * 
  5.  * @param {*} env {channel, dispatch, getState}
  6.  * @param {*} saga 可能是一个Generator生成器，也可能是一个Generator迭代器
  7.  * @param {*} args 额外参数
  8.  */
  9. function runSaga(env, saga, ...args){
  10.   let {channel, dispatch, getState} = env;
  11.   let it = typeof saga === 'function' ? saga() : saga;
  13.   function next(val, isError){
  14.     let result = isError ? it.throw(val) : it.next(val);
  15.     let {value: effect, done} = result;
  17.     if (!done){
  18.       // 如果 effect产出 是 Generator 迭代器
  19.       if (typeof effect[Symbol.iterator] === 'function'){
  20.         // 开始一个子进程执行 worker saga
  21.         // 此处的“开启子进程”只是一个比喻，并不是真的开了一个子进程，只是它的表现很像。
  22.         runSaga(env, effect); 
  23.         next(); //调用next，让当前saga继续往下走（子saga内部是否阻塞对其无影响）
  24.       } else if (typeof effect.then === 'function'){
  25.         effect.then(next);
  26.       } else{
  27.         // 根据不同的effect进行不同的处理
  28.         switch (effect.type){
  29.           case effectTypes.TAKE:
  30.             // effect = {type: 'TAKE', actionType: 'ASYNC_ADD'}
  31.             // 添加订阅，如果有人派发了 effect.actionType 这个动作类型，就执行next，继续走；
  32.             // 如果没有，就暂停，阻塞在这里
  33.             channel.once(effect.actionType, next);  
  34.             break;
  35.           case effectTypes.PUT:
  36.             // effect = {type: 'PUT', action: {type: 'ADD'}}
  37.             // put 不阻塞，派发action之后就继续向下执行
  38.             dispatch(effect.action);
  39.             next(effect.action);
  40.             break;
  41.           case effectTypes.FORK:
  42.             runSaga(env, effect.saga, ...effect.args);
  43.             next();
  44.             break;
  45.           case effectTypes.CALL:
  46.             effect.fn(...effect.args).then(next);
  47.             break;
  48.           case effectTypes.CPS:
  49.             effect.fn(...effect.args, (err, data) => {
  50.               err ? next(err, true) : next(data);
  51.             });
  52.             break;
  53. +         case effectTypes.ALL:
  54. +           let effects = effect.effects;
  55. +           let result = [];  //存放结果的数组
  56. +           let completeCount = 0;  //完成任务的数量
  58. +           effects.forEach((it, index) => {
  59. +             runSaga(env, it, (itResult) => {
  60. +               result[index] = itResult;
  61. +               if (++completeCount === effects.length){
  62. +                 next(result); //都完了之后，可以让当前的saga继续执行
  63. +               }
  64. +             });
  65. +           })
  66. +           break;
  67.           default: 
  68.           break;
  69.         }
  70.       } 
  71. +   } else {
  72. +     if (args && typeof args[0] === 'function'){
  73. +       args[0](effect);
  74. +     }
  75. +   }
  76.   }
  77.   next();
  78. }
  80. export default runSaga;

实现 cancel

如果父saga取消，子saga什么状态？现在这个版本里是保持原状 但其实应该是你把父saga取消了，所有的子saga也应该取消会比较好。

redux-saga/symbols.js

  1. export const TASK_CANCEL = Symbol('TASK_CANCEL');

redux-saga/effectTypes.js

  1. export const TAKE = 'TAKE'; //监听
  2. export const PUT = 'PUT'; //派发
  3. export const FORK = 'FORK'; //fork出一个子进程
  4. export const CALL = 'CALL'; //处理一个 返回值为Promise的函数fn
  5. export const CPS = 'CPS'; //处理一个 回调形式的函数
  6. export const ALL = 'ALL';
  7. export const CANCEL = 'CANCEL'; //取消一个saga任务

redux-saga/effects.js

  1. export function cancel(task){
  2.   return {type: effectTypes.CANCEL, task};
  3. }

redux-saga/runSaga.js

  1. import * as effectTypes from './effectTypes';
  2. + import {TASK_CANCEL} from './symbols';
  4. /** runSaga
  5.  * 
  6.  * @param {*} env {channel, dispatch, getState}
  7.  * @param {*} saga 可能是一个Generator生成器，也可能是一个Generator迭代器
  8.  * @param {*} args 额外参数
  9.  */
  10. function runSaga(env, saga, ...args){
  11. + let subTasks = [];  //子saga的task任务对象存放数组
  12. + // 每当执行runSaga的时候，会创建一个task任务对象
  13. + let task = {cancel: () => next(TASK_CANCEL)};
  14.   let {channel, dispatch, getState} = env;
  15.   let it = typeof saga === 'function' ? saga() : saga;
  17.   function next(val, isError){
  18.     let result;
  19.     if (isError){
  20.       result = it.throw(val); //立刻停止saga，表示出错了
  21.  +  } else if (val === TASK_CANCEL){
  22.  +       subTasks.forEach(subTask => subTask.cancel());  //如果父saga取消了，它的所有子saga也得取消
  23.  +    result = it.return(val);  //立刻停止saga，并返回 {value: val, done: true}
  24.     } else {
  25.       result = it.next(val);
  26.     }
  28.     let {value: effect, done} = result;
  30.     if (!done){
  31.       // 如果 effect产出 是 Generator 迭代器
  32.       if (typeof effect[Symbol.iterator] === 'function'){
  33.         // 开始一个子进程执行 worker saga
  34.         // 此处的“开启子进程”只是一个比喻，并不是真的开了一个子进程，只是它的表现很像。
  35. +       let iteratorTask = runSaga(env, effect);
  36. +       subTasks.push(iteratorTask);
  37.         next(); //调用next，让当前saga继续往下走（子saga内部是否阻塞对其无影响）
  38.       } else if (typeof effect.then === 'function'){
  39.         effect.then(next);
  40.       } else{
  41.         // 根据不同的effect进行不同的处理
  42.         switch (effect.type){
  43.           case effectTypes.TAKE:
  44.             // effect = {type: 'TAKE', actionType: 'ASYNC_ADD'}
  45.             // 添加订阅，如果有人派发了 effect.actionType 这个动作类型，就执行next，继续走；
  46.             // 如果没有，就暂停，阻塞在这里
  47.             channel.once(effect.actionType, next);  
  48.             break;
  49.           case effectTypes.PUT:
  50.             // effect = {type: 'PUT', action: {type: 'ADD'}}
  51.             // put 不阻塞，派发action之后就继续向下执行
  52.             dispatch(effect.action);
  53.             next(effect.action);
  54.             break;
  55.           case effectTypes.FORK:
  56. +           let forkTask = runSaga(env, effect.saga, ...effect.args);
  57. +           subTasks.push(forkTask);
  58. +           next(forkTask);
  59.             break;
  60.           case effectTypes.CALL:
  61.             effect.fn(...effect.args).then(next);
  62.             break;
  63.           case effectTypes.CPS:
  64.             effect.fn(...effect.args, (err, data) => {
  65.               err ? next(err, true) : next(data);
  66.             });
  67.             break;
  68.           case effectTypes.ALL:
  69.             let effects = effect.effects;
  70.             let result = [];  //存放结果的数组
  71.             let completeCount = 0;  //完成任务的数量
  73.             effects.forEach((it, index) => {
  74. +             let subTask = runSaga(env, it, (itResult) => {
  75.                 result[index] = itResult;
  76.                 if (++completeCount === effects.length){
  77.                   next(result); //都完了之后，可以让当前的saga继续执行
  78.                 }
  79.               });
  80. +             subTasks.push(subTask);
  81.             })
  82.             break;
  83. +         case effectTypes.CANCEL:
  84. +           effect.task.cancel();
  85. +           next();
  86. +           break;
  87.           default:
  88.           break;
  89.         }
  90.       } 
  91.     } else {
  92.       if (args && typeof args[0] === 'function'){
  93.         args[0](effect);
  94.       }
  95.     }
  96.   }
  97.   next();
  98. + return task;
  99. }
  101. export default runSaga;