30行代码实现Redux Callbag副作用隔离

背景

在过去很长一段时间里，因为我们经历了React，是React带领我们接触了JS函数式编程，之后Redux带领我们了解了elm思想，函数式编程就这样如春笋般的迅猛发展，但是，因为JS不是正统的函数式编程语言，在处理副作用问题上始终没有一个标准的解决方案来处理我们编程过程中的副作用隔离问题，这使得函数式编程并不纯，代码可维护性还是很难保证，比如：

我们在React应用中使用了redux来解决业务组件间的状态通讯问题，但是，在action派发过程中往往会存在很多副作用过程，也就是我们常说的异步IO，如果是简单的异步IO还好，用redux-thunk还能cover住，但是只要遇到异步IO的组合，那么，我们的action函数会变得越发复杂，比如以下代码：

//纯函数action，负责状态同步更新
const syncAction1 = payload=>({
    type:'ACTION_1',
    payload
})
 
const syncAction2 = payload=>({
    type:'ACTION_2',
    payload
})
 
...
 
//复合action，负责处理异步IO的分发
 
const asyncAction1 = payload=>dispatch=>{
    fetchData()
    .then((data)=>dispatch(syncAction1(data)))
    .then((data)=>{
        ....dispatch(syncAction2(data))
    })
}

这样的代码其实是存在问题的，比如每次要解决异步IO问题都需要声明定义一个复合action来处理,各种脏乱差的代码都在这里，action本应该是足够纯粹简洁的，这样也造成了action的创建过程变得非常繁琐，没法更高效的创建action。

所以，有没有更优雅的解决方案来帮助我们隔离副作用的同时，保证action的简单纯粹性呢？当然有，比如redux-saga/redux-observable，前者是基于generator来解决的，后者是基于rxjs来解决的，但是各有个的问题，比如借助了generator，因为yield的机制，generator没法优雅的实现无阻塞操作，必须得借助一个fork函数才能实现，而且它里面又引入了进程管理的概念，本来，我们是想解决副作用隔离的问题，结果redux-saga又带来了一个进程管理的问题给你，这复杂度立马提升了一个层次；然后是redux-observable，因为是使用rxjs来解决副作用隔离，Reactive Programming(RP)非常适合异步组合操作，所以在副作用隔离上的表现是非常优秀的，但是问题就是rxjs体积实在太大，因为它要实现“优雅”的链式调用，导致它必须以一个core library的思路来承载rxjs，最后就会导致rxjs越来越大，这对于前端来说必然是不能忍受的，这就是为什么rx理念在java或者其他领域推广的很好，反而在js领域推广的并不是那么好的原因。

首先，callbag是Reactive Programming(RP)，但是，它仅仅只是一个RP实现规范，具体规范可以看callbag，恰好就因为它是一个实现规范，所以它是No core library的，任何人都可以轻松实现一个callbag来扩展callbag体系的功能，快速实现高效轻量的RP解决方案，可以看看下面的callbag用例：

const {forEach, fromIter, take, map, pipe} = require('callbag-basics');
 
function* range(from, to) {
  let i = from;
  while (i <= to) {
    yield i;
    i++;
  }
}
 
pipe(
  fromIter(range(40, 99)), // 40, 41, 42, 43, 44, 45, 46, ...
  take(5), // 40, 41, 42, 43, 44
  map(x => x / 4), // 10, 10.25, 10.5, 10.75, 11
  forEach(x => console.log(x))
);
 
// 10
// 10.25
// 10.5
// 10.75
// 11

在没有任何ES语法糖的情况下，用pipe函数来连接所有的callbag source，其实pipe函数的实现也是非常简单的，有点类似underscore中的compose函数，lodash中的flow函数，pipe函数内部相当于是把函数的嵌套调用打扁

pipe(
  fromIter(range(40, 99)), // 40, 41, 42, 43, 44, 45, 46, ...
  take(5), // 40, 41, 42, 43, 44
  map(x => x / 4), // 10, 10.25, 10.5, 10.75, 11
  forEach(x => console.log(x))
)
 
//this is pipe
 
function pipe(...cbs) {
  let res = cbs[0];
  for (let i = 1, n = cbs.length; i < n; i++) res = cbs[i](res);
  return res;
}
 
//to
forEach(x => console.log(x))(
    map(x => x / 4)(
        take(5)(
          fromIter(range(40, 99))
        )
    )
)

所以每个callbag-xxx(比如callbag-map)都是一个高阶函数，类似于React的HOC实践一样，第一阶函数参数是callbag的外部入参，第二阶参数是由上一个callbag所产出的source处理函数，第三阶则是一个返回同样source类型的callbag标准函数，所以一个基本的callbag会是

const callbagBase = (options)=>source=>{
    return (type,sink)=>{//这是一个source函数
 
    }
}

上面代码其实描述的是callbag中的Listenable source，它通常都是三阶函数，因为它会接收上一个callbag所产出的source，所以它是一个三阶函数，callbag中还有一种类型是Puller source，通常为两阶函数(比如fromObs/fromEvent/fromPromise etc.)

const callbagPuller = (options)=>{
    return (type,sink)=>{
 
    }
}

当然，Listener source和Puller source最大的差别还在于内部处理逻辑

//实现一个callbag-map
const callbagListener = (fn)=>prev_source=>{
    return (type,next_source)=>{//这是一个source函数,给下一个callbag消费，注意，只有type为0的时候,参数next_source才是source函数
        if(type !== 0) return //判断消费者指令类型(0:启动 1:接收数据 2:停止/异常)0时执行后面逻辑
        prev_source(0,(type,data)=>{//与前一个callbag握手，传递的匿名函数也是一个source函数，给前一个callbag消费，用于接收消息
            if(type === 0) next_source(0,data) //type为0的时候data为上一个callbag的监听source函数，往后传递
            if(type === 1) next_source(1,fn(data)) //type为1的时候data为一个传输数据，往后传递
            if(type === 2) next_source(2,data) //type为2的时候data为一个error信息，往后传递
        })
    }
}
 
//实现一个callbag-interval
const callbagPuller = (options)=>{
    return (type,next_source)=>{//这是一个source函数,给下一个callbag消费，注意，只有type为0的时候参数next_source才是source函数
        if(type !== 0) return //判断消费者指令类型(0:启动 1:接收数据 2:停止/异常)0时执行后面逻辑
        let i = 0,id = null
        next_source(0,(type)=>{//与后一个callbag握手，如果后一个callbag往前传递消息了就可以接收，并作相关处理
           if(type === 2){
               clearInterval(id)
           }
        })
        id = setInterval(()=>{
             next_source(1,i++) //往下一个source函数持续传递数据
        },1000)
 
 
    }
}

看了上面实现，你应该大致能够理解callbag的核心设计思路了：

• callbag有两种类型，一个是puller，一个是listener，相当于一个是数据流的产出者，一个是数据流的消费者
• callbag的调用链中，当前callbag可以对下一个callbag发起数据消息，也可以对下一个callbag发起消息监听，当前callbag也可以对上一个callbag发起监听，继续往下一个callbag发送数据消息
• callbag的调用链中，sink作为最后一个链元素，它是作为整条链的启动器，因为倒数第二个链元素返回的还是一个source函数，需要一个最终消费者，所以sink就作为最后一个消费者而存在，在比如callbag-foreach/callbag-subscribe都是sink

这就是callbag，在函数调用链中实现了观察者模式，从而打造了一个reactive programming，代码极度轻量，非常容易理解。

什么是pipeline语法？

pipeline语法是ECMA script的最新提案，目前已经在babel7中尝鲜了，如果您想尝鲜可以按照如下方式安装依赖

npm install --save-dev @babel/cli @babel/core @babel/preset-env @babel/plugin-proposal-pipeline-operator

.babelrc定义

{
    "presets": ["@babel/preset-env"],
    "plugins": ["@babel/plugin-proposal-pipeline-operator"]
}

那么，pipeline语法到底是怎样的？

// old
 
c(b(a(data)))
 
// new
 
data
|> a
|> b
|> c

是不是很爽？它就是把嵌套函数调用链变成了如此优美的用法，对于上面的例子，如果a,b,c都是高阶函数的话，就变成了

// old
 
c()(b()(a()(data)))
 
// new
data
|> a()
|> b()
|> c()

所以，回到callbag，我们就可以放弃pipe函数了，直接

const {forEach, fromIter, take, map, pipe} = require('callbag-basics');
 
function* range(from, to) {
  let i = from;
  while (i <= to) {
    yield i;
    i++;
  }
}
 
//old
 
pipe(
  fromIter(range(40, 99)), // 40, 41, 42, 43, 44, 45, 46, ...
  take(5), // 40, 41, 42, 43, 44
  map(x => x / 4), // 10, 10.25, 10.5, 10.75, 11
  forEach(x => console.log(x))
)
 
//new
 
fromIter(range(40, 99)) // 40, 41, 42, 43, 44, 45, 46, ...
|>  take(5) // 40, 41, 42, 43, 44
|>  map(x => x / 4) // 10, 10.25, 10.5, 10.75, 11
|>  forEach(x => console.log(x))

简直帅呆了，有木有！

快速实现redux-callbag

好了，总算到了redux话题了，想必您看了上面的内容，对callbag也有比较深的了解了，主要是它真的不难，那么我们就来快速实现一个redux-callbag中间件吧。

import mitt from 'mitt'
 
export default (...epicses) => {
    return (store) => {
        const emitter = mitt()
        const actions = (start, sink) => {
            if (start !== 0) return
            const handler = (ev) => sink(1, ev)
            sink(0, (t) => {
                if (t === 2) emitter.off('action', handler)
            })
            emitter.on('action', handler)
        }
 
        epicses.forEach((epics) => {
            if (typeof epics === 'function') {
                epics(actions, store)
            }
        })
 
        return (next) => {
            return (action) => {
                emitter.emit('action', action)
                return next(action)
            }
        }
    }
}

就这样，30行代码不到，主要逻辑就是，在redux中间件中将响应的action转换成一个事件流，然后定义一个actions(source函数)，不断往后传递action，就这样就ok了，至于其他代码，都是一些扩展性功能扩展，使用的时候，我们就这样使用即可：

import { createStore, applyMiddleware } from "redux"
import { pipe, filter, forEach, map } from "callbag-basics"
import createCallbagMiddleware from "redux-callbag"
 
const epics = (actions, store) => {
  actions
    |> filter(typeOf("ADD_SOMETHING"))
    |> forEach(({ payload }) => {
        console.log("log:" + payload)
     })
 
  actions
     |> filter(typeOf("ADD_TODO"))
     |> forEach(({ payload }) => {
         setTimeout(() => {
           store.dispatch(addSomething(payload + "  23333333"))
         })
      })
 
}
 
const todos = (state = [], action)=> {
    switch (action.type) {
        case "ADD_TODO":
            return state.concat([action.payload])
        case "REMOVE_TODO":
            return []
        case "ADD_SOMETHING":
            return state.concat([action.payload])
        default:
            return state
    }
}
 
const addTodo = (payload)=> {
    return {
        type: "ADD_TODO",
        payload
    }
}
 
const addSomething = (payload)=> {
    return {
        type: "ADD_SOMETHING",
        payload
    }
}
 
const removeTodo = ()=> {
    return {
        type: "REMOVE_TODO"
    }
}
 
const typeOf = _type => {
    return ({ type }) => {
        return type === _type
    }
}
 
const store = createStore(
    todos,
    ["Hello world"],
    applyMiddleware(
        createCallbagMiddleware(epics)
    )
)
 
store.dispatch(addTodo("Hello redux"))
store.dispatch(addSomething("This will not add numbers"))
 
store.subscribe(() => {
    console.log(store.getState())
})
 
 
//output
 
log:This will not add numbers
log:Hello redux  23333333
[ 'Hello world',
  'Hello redux',
  'This will not add numbers',
  'Hello redux  23333333' ]

脱离了业务的一切技术都不能算做成功技术，那我们就以一个具体的业务问题来验证redux-callbag是否可用吧，就以redux-saga最典型的业务例子举例：

需求是这样的：

获取用户ID—>获取起飞时间与航班ID—>根据航班ID获取航班信息

—>根据起飞时间获取天气预报

在redux-callbag中可以这样写

import { createStore, applyMiddleware } from "redux"
import { pipe, filter, fromPromise,flatMap } from "callbag-basics"
import createCallbagMiddleware from "redux-callbag"
import subscribe from 'callbag-subscribe'
import fetch from 'mfetch'
import Case "case"
 
const typeOf = _type => {
    return ({ type }) => {
        return type === _type
    }
}
 
const createActions = (names)=>names.reduce((buf,name)=>{
    const upperName = Case.upper(name,'_')
    buf.constants = buf.constants || {}
    buf.constants[name] = upperName
    buf[name] = (payload)=>{
        return {
            type:upperName,
            payload
        }
    }
    return buf
},{})
 
const $actions = createActions([
    'fetchUser',
    'fetchUserSuccess',
    'fetchUserFailed',
    'fetchDeparture',
    'fetchDepartureSuccess',
    'fetchDepartureFailed',
    'fetchFlight',
    'fetchFilghtSuccess',
    'fetchFilghtFailed',
    'fetchForecast',
    'fetchForecastSuccess',
    'fetchForecastFailed',
    'fetchDashboardSuccess',
    'fetchDashboardFailed'
])
 
 
 
const epics = (actions, store) => {
  actions
    |> filter(typeOf($actions.constants.fetchUser))
    |> flatMap(()=>fromPromise(fetch('/user').then(res=>res.json())))
    |> subscribe({
        next(res){
           store.dispatch($actions.fetchUserSuccess(res.user))
           store.dispatch($actions.fetchDeparture(res.user))
        },
        error(error){
           store.dispatch($actions.fetchUserFailed(error))
        }
    })
 
 
 
  actions
     |> filter(typeOf($actions.constants.fetchDeparture))
     |> flatMap((user)=>fromPromise(fetch('/departure',{data:user}).then(res=>res.json())))
     |> subscribe({
        next(res){
           store.dispatch($actions.fetchDepartureSuccess(res.departure))
           store.dispatch($actions.fetchFlight(res.departure))
           store.dispatch($actions.fetchForecast(res.departure))
        },
        error(error){
           store.dispatch($actions.fetchDepartureFailed(error))
        }
    })
 
   actions
     |> filter(typeOf($actions.constants.fetchFlight))
     |> flatMap((departure)=>fromPromise(fetch('/flight',{data:departure}).then(res=>res.json())))
     |> subscribe({
        next(res){
           store.dispatch($actions.fetchFlightSuccess(res.flight))
        },
        error(error){
           store.dispatch($actions.fetchFlightFailed(error))
        }
    })
 
   actions
     |> filter(typeOf($actions.constants.fetchForecast))
     |> flatMap((departure)=>fromPromise(fetch('/forecast',{data:departure}).then(res=>res.json())))
     |> subscribe({
        next(res){
           store.dispatch($actions.fetchForecastSuccess(res.flight))
        },
        error(error){
           store.dispatch($actions.fetchForecastFailed(error))
        }
    })
 
}
 
const store = createStore(
    todos,
    initState,
    applyMiddleware(
        createCallbagMiddleware(epics)
    )
)

上面代码中，除了epics是核心业务逻辑，其他都是一些基本实现，所以这样看下来，其实用redux-callbag，几行代码就能实现redux-saga相同的效果，并且可读性，易维护性都实属上乘，那么，还等什么？赶紧用起来吧！

redux-callbag 欢迎您的享用！

总结

引用 \@工业聚 大神的一句话，callbag 让你从 callback -> generator -> promise -> async/await 的演进中，再回到 callback。这是一种「看山是山，看山不是山，看山还是山」的境界。是的，callbag就是这么给力！

最后总结一下我对响应式编程的体会，任何一种数据，只要它会随着时间推移而改变，我们都有可能去订阅它的变化，在一个订阅管道里去做响应式的操作，这样我们的操作将变得如此清晰明了，不管是错综复杂的依赖逻辑关系还是副作用强烈的异步组合关系，在响应式编程里都不是问题

饥人谷一直致力于培养有灵魂的编程者，打造专业有爱的国内前端技术圈子。如造梦师一般帮助近千名不甘寂寞的追梦人把编程梦变为现实，他们以饥人谷为起点，足迹遍布包括facebook、阿里巴巴、百度、网易、京东、今日头条、大众美团、饿了么、ofo在内的国内外大小企业。 了解培训课程：加微信 xiedaimala03，官网:https://jirengu.com

本文作者：饥人谷方应杭老师