本文共 11022 字,大约阅读时间需要 36 分钟。
redux中间件编写
redux-saga中间件源码解析 概念解析 中间件编写部分:入口文件 ----> multicastChannel ----> buffer 入口文件 multicastChannel函数 buffer函数 启动中间件:runSaga ----> proc runSaga proc Effect redux中间件编写 如何编写一个中间件呢?({dispatch, getState}) => next => action => {
// write your code here next(action) } 1 2 3 4 上面的函数参数什么含义呢?{dispatch, getState}: dispatch和getState就是redux中常用的函数
next: 就是下个中间件 action: 那就是action咯 1 2 3 具体可以参考Redux源码解析redux-saga中间件源码解析
概念解析 redux-saga源码分为三部分:中间件部分
启动中间件 effects 整体来说,redux-saga做了这样一件事: 监听一系列action,然后至于对action做什么操作,是由用户传入的generator函数决定的,同时提供了一些列API(take\put\call\race\fork等),用于对操作产生的effect进行操作 中间件编写部分:入口文件 ----> multicastChannel ----> buffer 入口文件 **入口文件是一个标准的中间件编写方式,加入了 channel.put(action) **// 这是redux-saga中间件的入口函数,是按照中间件的基本编写方式来写的
// context、options默认是空的,分析的时候可以忽略 function sagaMiddlewareFactory({ context = {}, ...options } = {}) { const { sagaMonitor, logger, onError, effectMiddlewares } = options let boundRunSaga// 下面就是中间件基本的编写方式
function sagaMiddleware({ getState, dispatch }) { const channel = stdChannel() // identity 是一个函数 identity = v => v channel.put = (options.emitter || identity)(channel.put)boundRunSaga = runSaga.bind(null, {
context, channel, dispatch, getState, sagaMonitor, logger, onError, effectMiddlewares, })return next => action => {
if (sagaMonitor && sagaMonitor.actionDispatched) { sagaMonitor.actionDispatched(action) } const result = next(action) channel.put(action) return result } } // 负责启动中间件的函数,下一小节讲述 sagaMiddleware.run = (...args) => { return boundRunSaga(...args) }sagaMiddleware.setContext = props => {
assignWithSymbols(context, props) }return sagaMiddleware
} 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 multicastChannel函数 multicastChannel是stdChannel执行之后的结果,然后把action传入function stdChannel() {
const chan = multicastChannel() const { put } = chan chan.put = input => { // SAGA_ACTION :一个字符串,模版字符串 `@@redux-saga/${name}` if (input[SAGA_ACTION]) { put(input) return } // asap是一个调度策略,存放了一个quene,然后每次只允许一个任务执行 asap(() => { put(input) }) } return chan } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 oasp函数const queue = []
let semaphore = 0function exec(task) {
try { suspend() task() } finally { release() } }export function asap(task) {
queue.push(task)if (!semaphore) {
suspend() flush() } }export function suspend() {
semaphore++ }function release() {
semaphore-- } // while循环,将队列中执行完成,直到为空 export function flush() { release()let task
while (!semaphore && (task = queue.shift()) !== undefined) { exec(task) } }1
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 那put是做了什么呢?put是multicastChannel函数执行的结果,返回了一个对象,对象包含put、take方法 take方法:将回调函数存入nextTakers put方法:执行相应的回调函数function multicastChannel() {
let closed = false// 在状态管理中,经常碰到current和next的操作,为了保持一致性
// 一个代表当前状态(任务队列), // 一个代表下一个状态(任务队列), // 初始状态两个是一致的 let currentTakers = [] let nextTakers = currentTakers // 下面函数做的操作是,将当前的队列,复制给下一个队列 const ensureCanMutateNextTakers = () => { if (nextTakers !== currentTakers) { return } nextTakers = currentTakers.slice() }const close = () => {
closed = true
const takers = (currentTakers = nextTakers) nextTakers = [] takers.forEach(taker => { // END是一个对象,END = { type: CHANNEL_END_TYPE } taker(END) }) }return {
[MULTICAST]: true, put(input) {if (closed) {
return } // isEND是一个函数,判断是不是已经结束了 // isEnd = a => a && a.type === CHANNEL_END_TYPE if (isEnd(input)) { close() return }const takers = (currentTakers = nextTakers)
for (let i = 0, len = takers.length; i < len; i++) {
const taker = takers[i]if (taker[MATCH](input)) {
taker.cancel() taker(input) } } }, take(cb, matcher = matchers.wildcard) { if (closed) { cb(END) return } cb[MATCH] = matcher ensureCanMutateNextTakers() nextTakers.push(cb)cb.cancel = once(() => {
ensureCanMutateNextTakers() remove(nextTakers, cb) }) }, close, } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 buffer函数 buffer就是一个数组,用来存储actionimport { kTrue, noop } from './utils'
const BUFFER_OVERFLOW = "Channel's Buffer overflow!"
const ON_OVERFLOW_THROW = 1
const ON_OVERFLOW_DROP = 2 const ON_OVERFLOW_SLIDE = 3 const ON_OVERFLOW_EXPAND = 4const zeroBuffer = { isEmpty: kTrue, put: noop, take: noop }
function ringBuffer(limit = 10, overflowAction) {
let arr = new Array(limit) let length = 0 let pushIndex = 0 let popIndex = 0const push = it => {
arr[pushIndex] = it pushIndex = (pushIndex + 1) % limit length++ }const take = () => {
if (length != 0) { let it = arr[popIndex] arr[popIndex] = null length-- popIndex = (popIndex + 1) % limit return it } }const flush = () => {
let items = [] while (length) { items.push(take()) } return items }return {
isEmpty: () => length == 0, put: it => { if (length < limit) { push(it) } else { let doubledLimit switch (overflowAction) { case ON_OVERFLOW_THROW: throw new Error(BUFFER_OVERFLOW) case ON_OVERFLOW_SLIDE: arr[pushIndex] = it pushIndex = (pushIndex + 1) % limit popIndex = pushIndex break case ON_OVERFLOW_EXPAND: doubledLimit = 2 * limitarr = flush()
length = arr.length
pushIndex = arr.length popIndex = 0arr.length = doubledLimit
limit = doubledLimitpush(it)
break default: // DROP } } }, take, flush, } }export const none = () => zeroBuffer
export const fixed = limit => ringBuffer(limit, ON_OVERFLOW_THROW) export const dropping = limit => ringBuffer(limit, ON_OVERFLOW_DROP) export const sliding = limit => ringBuffer(limit, ON_OVERFLOW_SLIDE) export const expanding = initialSize => ringBuffer(initialSize, ON_OVERFLOW_EXPAND)1
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 启动中间件:runSaga ----> proc runSaga runSaga:你传入的saga函数是一个generator函数function runSaga(options, saga, ...args) {
// saga就是传过来的saga函数 const iterator = saga(...args)const {
channel = stdChannel(), dispatch, getState, context = {}, sagaMonitor, logger, effectMiddlewares, onError, } = optionsconst effectId = nextSagaId()
// 日志 const log = logger || _log const logError = err => { log('error', err) if (err && err.sagaStack) { log('error', err.sagaStack) } } // 是否有effectMiddlewares const middleware = effectMiddlewares && compose(...effectMiddlewares) const finalizeRunEffect = runEffect => { if (is.func(middleware)) { return function finalRunEffect(effect, effectId, currCb) { const plainRunEffect = eff => runEffect(eff, effectId, currCb) return middleware(plainRunEffect)(effect) } } else { return runEffect } }const env = {
stdChannel: channel, dispatch: wrapSagaDispatch(dispatch), getState, sagaMonitor, logError, onError, finalizeRunEffect, }try {
suspend() // 这一行是最终执行的 const task = proc(env, iterator, context, effectId, getMetaInfo(saga), null)if (sagaMonitor) {
sagaMonitor.effectResolved(effectId, task) }return task
} finally { flush() } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 proc proc是一个很长的函数,里面包含了很多子函数,但是总结起来就做了两件事:执行你传入的generator函数
返回一个task描述 function proc(env, iterator, parentContext, parentEffectId, meta, cont) { const taskContext = Object.create(parentContext) const finalRunEffect = env.finalizeRunEffect(runEffect) const task = newTask(parentEffectId, meta, cont) const mainTask = { meta, cancel: cancelMain, _isRunning: true, _isCancelled: false }const taskQueue = forkQueue(
mainTask, function onAbort() { cancelledDueToErrorTasks.push(...taskQueue.getTaskNames()) }, end, ) next()return task
} 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 next是一个自动执行函数function next(arg, isErr) {
if (!mainTask._isRunning) { throw new Error('Trying to resume an already finished generator') }try {
let result if (isErr) { result = iterator.throw(arg) } else if (shouldCancel(arg)) { mainTask._isCancelled = true next.cancel() result = is.func(iterator.return) ? iterator.return(TASK_CANCEL) : { done: true, value: TASK_CANCEL } } else if (shouldTerminate(arg)) { result = is.func(iterator.return) ? iterator.return() : { done: true } } else { result = iterator.next(arg) }if (!result.done) {
digestEffect(result.value, parentEffectId, '', next) } else { mainTask._isRunning = false mainTask.cont(result.value) } } catch (error) { if (mainTask._isCancelled) { env.logError(error) } mainTask._isRunning = false mainTask.cont(error, true) } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 runEffect函数:根据不同的effect函数么,执行不同的操作function runEffect(effect, effectId, currCb) {
if (is.promise(effect)) { resolvePromise(effect, currCb) } else if (is.iterator(effect)) { resolveIterator(effect, effectId, meta, currCb) } else if (effect && effect[IO]) { const { type, payload } = effect if (type === effectTypes.TAKE) runTakeEffect(payload, currCb) else if (type === effectTypes.PUT) runPutEffect(payload, currCb) else if (type === effectTypes.ALL) runAllEffect(payload, effectId, currCb) else if (type === effectTypes.RACE) runRaceEffect(payload, effectId, currCb) else if (type === effectTypes.CALL) runCallEffect(payload, effectId, currCb) else if (type === effectTypes.CPS) runCPSEffect(payload, currCb) else if (type === effectTypes.FORK) runForkEffect(payload, effectId, currCb) else if (type === effectTypes.JOIN) runJoinEffect(payload, currCb) else if (type === effectTypes.CANCEL) runCancelEffect(payload, currCb) else if (type === effectTypes.SELECT) runSelectEffect(payload, currCb) else if (type === effectTypes.ACTION_CHANNEL) runChannelEffect(payload, currCb) else if (type === effectTypes.FLUSH) runFlushEffect(payload, currCb) else if (type === effectTypes.CANCELLED) runCancelledEffect(payload, currCb) else if (type === effectTypes.GET_CONTEXT) runGetContextEffect(payload, currCb) else if (type === effectTypes.SET_CONTEXT) runSetContextEffect(payload, currCb) else currCb(effect) } else { // anything else returned as is currCb(effect) } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 Effect 从上面的描述来看,saga中间件,做了两件事情:根据你传入的generator函数,next执行
你的generator函数中,存在一些副作用操作,比如put、call、race等,然后根据这些操作再进行其他的操作,而这些操作只做了一件事情,就是生成effect,那什么是effect呢? export function race(effects) { return makeEffect(effectTypes.RACE, effects) } 1 2 3 const makeEffect = (type, payload) => ({ [IO]: true, type, payload }) 1 Effect就是一个个对象,而这些对象的解释则交给上面的runEffect来进行 以put—effect为例解释来看function runPutEffect({ channel, action, resolve }, cb) {
asap(() => { let result try { // dispatch相应的action result = (channel ? channel.put : env.dispatch)(action) } catch (error) { cb(error, true) return } if (resolve && is.promise(result)) { resolvePromise(result, cb) } else { cb(result) } }) } --------------------- 作者:daysRoc 来源:CSDN 原文:https://blog.csdn.net/Donspeng/article/details/83060243 版权声明:本文为博主原创文章,转载请附上博文链接!