深入浅出ーRedux-sagaソース

目次

reduxミドルウェア作成

redux-sagaのミドルウェアのソースコード解析

概念解析

ミドルウェア編集部:エントランスファイル-->multicast Chanel-->ブザー

エントランスファイル

mlticast Chanel関数

ブザー関数

起動中間件:runSaga-->proc

ルンSaga

proc

Effect

redux中間文書作成
ミドルウェアはどう書きますか?

({dispatch, getState}) => next => action => {
    // write your code here
    next(action)
}

上の関数のパラメータはどういう意味ですか?

{dispatch, getState}: dispatch getState  redux      
next:        
action:    action 

具体的にはReduxソース解析を参照してください.
redux-sagaの中间件のソースコードの解析
概念解析
redux-sagaソースは三つの部分に分けられます.

中間部品部

中間部品

を起動する.

effects全体として、redux-sagaは一連のactionを傍受し、その後、actionに対してどのような操作をするかは、ユーザから入ってきたgenerator関数によって決定され、いくつかの列API(take\put\call\race\forkなど)を提供し、操作によって生成されるeffectを操作する

.
中間部品編纂部分:エントランスファイル-->multicartChanel-->ブザー
インポートファイル
＊入り口ファイルは標準的な中間部品の編纂方式で、チャンネル.put(*)を入れました.

//   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
}

multicast Chanel関数
multicast ChanelはstdChanelが実行した後の結果で、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
}

oasp関数

const queue = []
let semaphore = 0

function 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)
  }
}

あのputは何をしましたか?putはmulticartChanel関数が実行した結果、オブジェクトに戻り、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,
  }
}

ブザー関数
bufferは行列で、actionを保存します.

import { 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 = 4

const 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 = 0

  const 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 * limit

            arr = flush()

            length = arr.length
            pushIndex = arr.length
            popIndex = 0

            arr.length = doubledLimit
            limit = doubledLimit

            push(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)

起動中間件:runSaga-->proc
ルンサガ
ルンサガ:あなたが入ってきたサガ関数は、ゲナート関数です.

function runSaga(options, saga, ...args) {
  // saga      saga  
  const iterator = saga(...args)

  const {
    channel = stdChannel(),
    dispatch,
    getState,
    context = {},
    sagaMonitor,
    logger,
    effectMiddlewares,
    onError,
  } = options

  const 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()
  }
}

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
}

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)
    }
  }

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)
    }
  }

Effect
上の説明から、サガのミドルウェアは、二つのことをしました.

あなたが入ってきたgenerator関数に基づいて、nextは

を実行します.

あなたのゲナート関数には、put、call、raceなどの副作用があります.これらの操作によって他の操作が行われます.これらの操作は一つのことしかできません.つまり、effectとは何ですか?

export function race(effects) {
  return makeEffect(effectTypes.RACE, effects)
}

const makeEffect = (type, payload) => ({ [IO]: true, type, payload })

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)
      }
    })
  }