ReactiveX流式编程—从xstream讲起

ReactiveX流式编程

ReactiveX来自微软，它是一种针对异步数据流的编程。简单来讲，它将一切数据，包括HTTP请求，DOM事件或者普通数据等包装成流的形式，而后用强大丰富的操做符对流进行处理，使你能以同步编程的方式处理异步数据，并组合不一样的操做符来轻松优雅的实现你所须要的功能。

为何从xstream讲起

xstream的做者也是rxjs的深度用户，可是做者基于一些实践中考虑而开发这个库，做者的解释：WHY WE BUILT XSTREAM

1. xstream只有26个核心操做符和工厂函数
   
2. 只支持热模式流
   
3. 只有stream、listener、producer 三个概念，比较好理解
   
4. 压缩后只有30kb大小，平常开发能够轻松集成代替部分繁琐逻辑
   

xstream简单上手

import xs from 'xstream'

// Tick every second incremental numbers,
// only pass even numbers, then map them to their square,
// and stop after 5 seconds has passed

var stream = xs.periodic(1000)
  .filter(i => i % 2 === 0)
  .map(i => i * i)
  .endWhen(xs.periodic(5000).take(1))

// So far, the stream is idle.
// As soon as it gets its first listener, it starts executing.

stream.addListener({
  next: i => console.log(i),
  error: err => console.error(err),
  complete: () => console.log('completed'),
})
复制代码

核心概念

Stream

流表明从事件发生、处理、监听的一条管道，每一个stream都有不少operator相似：map, filter, fold, take；
每次调用operator都返回一个新的stream；
通常说来stream中的数据是由producer生产的，可是你能够调用shamefullySend*系列函数手动发射事件，可是这种方法是反reactive的，做者强烈不推荐使用；按照个人理解，这些方法在多个stream联合工做的，用来mock某些流的数据时候会比较有用

Listener

监听者 ，stream的出口，消费管道最终产物；包含有3个方法

1. next：stream里每次有管道里产生的数据到流入到这个next方法里接收
   
2. error：stream数据流转中有异常状况时调用
3. complete：生产者调用了stop方法后调用

var listener = {
  next: (value) => {
    console.log('The Stream gave me a value: ', value);
  },
  error: (err) => {
    console.error('The Stream gave me an error: ', err);
  },
  complete: () => {
    console.log('The Stream told me it is done.');
  },
}
复制代码

Producer

生产者，stream的入口，用来持续产生流的输入

var producer = {
  start: function (listener) {
    this.id = setInterval(() => listener.next('yo'), 1000)
  },

  stop: function () {
    clearInterval(this.id)
  },

  id: 0,
}

// This fellow delivers a 'yo' next event every 1 second
var stream = xs.create(producer)
复制代码

MemoryStream

记忆流 普通stream的记忆版：它会记住最后一次发送给listener的next方法的数据，这样后来addListener添加的监听者能收到记住的这个数据； 这个特性是颇有用的，可以用来保存应用运行过程当中的一些临时状态。

Stream的构造-Factories

create

标准的经过producer构造, create(producer)

createWithMemory

标准的经过producer构造memorystream, createWithMemory(producer)

from

从Array|PromiseLike|Observable建立一个stream

of

从字面量建立一个stream，这样建立的stream会马上发射全部的参数，并触发completed

fromPromise

从promise建立一个stream

merge

合并两个stream成为一个stream，合并的后的数据按照本来的时间线继续输出(以下图)

combine

这个单纯用文字不太好解释，请看下图(借用的rxjs里的combineLatest图，功能是相似的)

另外，rxjs中还有个一个相似的zip操做符(xstream中不存在，本身实现)，看下图仔细体会和xstream的combine的不一样

经常使用的操做符-Operators

map

mapTo

filter

take

drop

图片借用的rx里的skip，是同样的效果

fold

图片借用的rx里的scan，是同样的效果

flatten

这个是操做符就有点复杂了，涉及到了分流的状况，主要功能是将主stream里返回的支流直接打平，输出支流里的数据；整个xstream标准operators(extra下有扩展的)里只有这个操做符有涉及到分流的处理，弹珠(Marble)图以下

这里解释一下，为何b输出以后，主流程走到第二个tick，开始输出第二个支流，这是第一个支流的后续输出都会被废弃；

实践一个TODO List

流式思考

假如如今须要咱们写一个简单的todolist：有一个 input 和一个 button 当我在input输入内容以后，点击 button 就往todolist集合里添加一条数据，每条todo行前面有个 checkbox 用来勾选todo的完成状态，每条todo行后面有一个 del 按钮，用来删除这条todo

ok，让咱们开始以前先用 流 式的方式思考一下这个问题， 流 式的方式是基于时间线的演进系统动态变化的一个抽象，那么基于此咱们能够很简单抽闲出 3 条时间线：

基于此，能够很容易写出3条stream的代码以下：

// 工具函数，方便的建立dom事件流
import fromEvent from 'xstream/extra/fromEvent';

// 从添加按钮建立的stream
const addTodoBtn$ = fromEvent(addBtnEl, 'click').map(() => inputEl.value).filter(v => v && v !== '');

// 从删除按钮触发的stream
const delTodoBtn$ = fromEvent(document.body, 'click').map((e: Event) => e.target).filter((target: HTMLElement) => target.classList.contains('delTodo')).map((target: HTMLElement) => parseInt(target.dataset.index));

// 从标记完成选项触发的stream
const toggleTodoInput$ = fromEvent(document.body, 'change').map((e: Event) => e.target).filter((target: HTMLElement) => target.classList.contains('toggleTodo')).map((target: HTMLInputElement) => ({ checked: target.checked, index: parseInt(target.dataset.index) }));
复制代码

好了，如今咱们有了3条stream：

• addTodoBtn$在时间线上持续收集 button 的点击，并判断input框里是否有输入有效内容，若是有的话就将输入的内容做为stream的数据发射出去
• delTodoBtn$在时间线上持续收集 del 的点击，并将绑定的data-index数值发射出去
• toggleTodoInput$在时间线上持续收集 checkbox 的点击，并将当前checkbox的选中状态和data-index一块儿发射出去

保存状态

如今咱们有了3条stream，那么该如何将这些stream与dom的操做对应起来呢？同时还有另一个问题：传统的开发过程当中，咱们须要有一个外部变量相似state这样用来保存每次操做后最新的todolist数据集合(反作用)； 可是ReactiveX提倡的方式就是要消除反作用，咱们须要一点儿技巧来处理这个情况；

这里咱们思考一下整个操做分两部分：增量数据、减量数据、更新数据， 而减量数据和更新数据都是基于增量数据源的基础上操做的；那么咱们须要定义一个增量数据源，而且须要能持续保持这个数据源最后的数据状态。

让咱们翻翻上面的operators，看看有哪一个操做符是能够用来持久保存局部变量的？ 没错，就是 fold 操做符：

// 数据来源
const startUp$ = addTodoBtn$.fold((todos: Todo[], inputValue) => {
  const todo:Todo = { text: inputValue, completed: false };
  todos.push(todo);
  return todos;
}, []);// 初始化空数据列表
复制代码

因为这里的todolist的增量来源只有 button 一个(若是有多个，能够看看 combine 操做符，这里不展开)；

分支流和flatten应用

有了增量数据源，那么咱们在增量数据源的每一个tick上分出一个减量、更新的 支流 (参看上面的flatten操做符)，这样支流执行的时候拿到的都是数据源的最新数据；

// 监听数据来源，并触发删除的stream
const delTodos$ = startUp$.map(todos => delTodoBtn$.map(index => {
  console.log(index);
  todos.splice(index, 1);
  return todos;
})).flatten();

// 监听数据来源，并触发选中的stream
const toggleTodos$ = startUp$.map(todos => toggleTodoInput$.map(({ checked, index }) => {
  console.log(checked, index);
  if (todos[index]) {
    todos[index].completed = checked;
  }
  return todos;
})).flatten();
复制代码

组装stream

ok，如今咱们的数据来源是多个，输出的都是todolist最新的数据集合状态，让咱们把这些stream管道组装起来：

// 组合起来
const todos$ = xs.merge(startUp$, delTodos$, toggleTodos$);

todos$.addListener({
  next: function(todos: Todo[]) {
    console.log(todos);
    renderTodos(todos);
  },
  error: function(e) {
    console.error(e);
  },
  complete: function() {

  }
})

复制代码

反作用

在定义完清晰的stream后，咱们的实际业务代码就是这么"简单"，因为stream的出口一直都是最新的todolist集合咱们实现了相似react的全量渲染；哈哈，实际上这里还有个不怎么简单的反作用方法：

const inputEl = document.getElementById('input') as HTMLInputElement;
const addBtnEl = document.getElementById('addBtn') as HTMLButtonElement;
const todoListEl = document.getElementById('lists');
const initTodos = [];
const renderTodos = function(todos: Todo[]) {
  if (!todos) {
    return;
  }
  todoListEl.innerHTML = '';
  const fragement = document.createDocumentFragment();
  todos.forEach((todo, index) => {
    const liEL = document.createElement('li');
    if (todo.completed) {
      liEL.className = 'completed';
    }
    liEL.innerHTML = `<input type='checkbox' class="toggleTodo" name="toggleTodo" data-index="${index}" ${todo.completed ? 'checked' : null} /> <span>${todo.text}</span> <a href='javascript:void(0);' class='delTodo' data-index="${index}">x</a>`;
    fragement.appendChild(liEL);
  });
  todoListEl.appendChild(fragement);
}

复制代码

最终效果

总结

经过实际例子能够看到这里有一大段不怎么优雅的反作用方法，用来操做dom元素，因为咱们基于全量渲染的思想，并无使用传统的同步增删改dom的方式，不然反作用代码会更多； 同时因为没有vdom的加持，此段渲染代码纯粹只能用来demo展现一下；  至于更优雅的反作用处理和vdom能力，能够期待我后续关于cycle.js的介绍😜

专栏其余文章

• 【中后台应用】从表单抽象到表单中台
• 关于Terminal你不知道的事儿
• 如何实现一个Interval Hook

FE One

关注咱们的公众号FE One，会不按期分享JS函数式编程、深刻Reaction、Rxjs、工程化、WebGL、中后台构建等前端知识