redux-saga源码解析-下

发表于 |

这篇文中是对 redux-saga源码解析 的补充，可以帮助大家更加全面的了解redux-saga。主要探索一下 channel, 和redux-saga中的取消任务
想要具体分析一下channel并不是因为多么复杂，而是工作中可能会用到

channel

在前一篇文章中说到其实redux-saga支持三种channel，分别是channel， eventChannel， multicastChannel。multicastChannel 在前一篇文章已经分析过。这里就看一下另外两个。

channel

接受一个buffer对象，返回 channel 对象
buffer的作用：当没有任务监听器时，先把事件缓存起来，如果后面有新的任务监听器建立，就立即执行任务。

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
export function channel(buffer = buffers.expanding()) {
  let closed = false
  let takers = []

  function put(input) {
    // 如果 channel 已经关闭，返回 
    if (closed) {
      return
    }
    // 当没有任务监听器时， 缓存在 buffer 里
    if (takers.length === 0) {
      return buffer.put(input)
    }

    // 正常执行任务
    const cb = takers.shift()
    cb(input)
  }

  function take(cb) {
    // 如果 channel 已经关闭并且buffer为空，则对任务发送END信号
    // 也可以看出来，即使 channel 关闭，buffer只要不为空，还是可以正常建立任务监听器的
    if (closed && buffer.isEmpty()) {
      cb(END)
    } else if (!buffer.isEmpty()) {
      // buffer不为空的话 直接执行
      cb(buffer.take())
    } else {
      // 正常建立任务监听器
      takers.push(cb)
      cb.cancel = () => remove(takers, cb)
    }
  }

  // 清空buffer里的任务
  function flush(cb) {
    //...
    if (closed && buffer.isEmpty()) {
      cb(END)
      return
    }
    cb(buffer.flush())
  }

  // 当关闭channel时，终止所有任务监听器
  function close() {
    //...
    if (!closed) {
      closed = true
      if (takers.length) {
        const arr = takers
        takers = []
        for (let i = 0, len = arr.length; i < len; i++) {
          const taker = arr[i]
          taker(END)
        }
      }
    }
  }

  return {
    take,
    put,
    flush,
    close,
  };
}

eventChannel

eventChannel 是对 channel的封装，主要区别是 channel.put 函数交给subscribe函数驱动

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
export function eventChannel(subscribe, buffer = buffers.none()) {
  let closed = false
  let unsubscribe

  const chan = channel(buffer)
  const close = () => {
    if (is.func(unsubscribe)) {
      unsubscribe()
    }
    chan.close()
  }

  unsubscribe = subscribe(input => {
    if (isEnd(input)) {
      close()
      closed = true
      return
    }
    chan.put(input)
  })

  if (!is.func(unsubscribe)) {
    throw new Error('in eventChannel: subscribe should return a function to unsubscribe')
  }

  unsubscribe = once(unsubscribe)

  if (closed) {
    unsubscribe()
  }

  return {
    take: chan.take,
    flush: chan.flush,
    close,
  }
}

example:

1
2
3
4
5
// 这个channel刚建立就被关闭了
eventChannel(emitter => {
  emitter(END)
  return () => {}
})

cancel

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
function* test(action) {
  try {
    yield call(delayUtil, 5000);
    yield put({ type: 'ADD_TODO', text: action.text  });
  } finally {
    if (yield cancelled()) {
      console.log('cancelled');
    }
  }
}

function* rootSaga() {
  const action = yield take('ADD_TODO_SAGA');
  const task = yield fork(test, action);

  yield call(delayUtil, 4000);
  yield cancel(task);
}

sagaMiddleware.run(rootSaga)

这里首先监听了ADD_TODO_SAGA事件，在4s之后又取消了任务，由于在子任务中延时了5s，所以 ADD_TODO 并没有被抛出。

cancel effect

1
2
3
4
5
function cancelSingleTask(taskToCancel) {
  if (taskToCancel.isRunning()) {
    taskToCancel.cancel()
  }
}

可以看到调用task的cancel方法。

1
2
3
4
5
6
function cancel() {
  if (task._isRunning && !task._isCancelled) {
    task._isCancelled = true
    taskQueue.cancelAll()
  }
}

实际执行的就是 taskQueue.cancelAll() 在这里取消了任务及其子任务。

cancelled effect

1
2
3
function runCancelledEffect(data, cb) {
  cb(Boolean(mainTask._isCancelled))
}

直接将_isCancelled作为结果

由于取消任务是在父线程上作的，所以应该通知到子线程上。
原理是当迭代器运行return方法时，iterator.return() 函数就会跳过try剩下的语句，直接进入finally代码块。这里改造一下之前的自动流程控制demo

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
const delay = (ms) => {
  return new Promise((res) => {
    setTimeout(res, ms);
  });
}

function *main() {
  try {
    console.log('begin');
    yield delay(1000);
    console.log('1s later');
    yield delay(2000);
    console.log('end');
  } finally {
    console.log('finally');
  }
}

function autoRun(gfunc) {
  const gen = gfunc();

  function next() {
    const res = gen.next();
    gen.return('cancel');
    if (res.done) return;
    res.value.then(next);
  }

  next();
}

autoRun(main);
// begin
// finish

可以看到直接进入了finally代码块。

总结

这边文章主要是对前一篇文章的补充，探索了一下我比较感兴趣的两个点，也希望读者看完之后会有所帮助。
接下来我会分析一下dva（基于 redux、redux-saga 和 react-router 的轻量级前端框架）