全面拥抱 Reactivity: RxJS, RSocket & Svelte

在 Reactive 方面，有两个非常知名的工程师，他们是来自 Netfix 的 RxJava 作者Ben Christensen 和 RxJS 作者 Ben Lesh，两人在 Reactive 方面都做出了非常多的贡献。在 2019 年初 Reactive 社区有两个非常令人振奋的消息，RSocket 步入了大众视野，RSocket 是二进制异步化通讯协议，完全兼容 Reactive 语义，提供了多种通讯模型，而且是对等通讯；此外就是 Svelte 3.0 发布，Svelte 作者 Rich Harris 做了知名的 “Rethinking Reactivity” 的演讲。Svelte 通过编译器的方式让Reactive 使用更简单，没有 Virtual DOM 等性能损失，代码更小性能更高。当然在 Reactive 社区，RxJS 还是最核心的底层框架，始终发挥着重大作用。这篇文章，我们就讨论一下如何基于 RxJS 衔接 RSocket 和 Svelte， 在前端开发中全面拥抱 Reactivity，那么就从 Svelte 开始吧。

你可以通过 RxJS 的 interval 进行定时状态更新，同时可以实现非常复杂的逻辑：

<script>
  import { interval } from "rxjs";
  import { map, take,startWith} from "rxjs/operators";


  const counter = interval(1000).pipe(
    map(i => i + 1),
    startWith(0),
    take(10)
  );
</script>


<h2>Count to 10</h2>


{$counter}

RxJS BehaviorSubject对象

RxJS 的 BehaviorSubject 可以在 Svelte 中直接使用，和 Svelte 的状态管理发挥着同样的作用。

<script>
   const store1 = new BehaviorSubject(0);
   store1.set = store1.next;
</script>


<button on:click={()=>{store1.next(1)}}>increment 1</button>
<button on:click={()=>{$store1 = 2}}>increment 2</button>


{$store1}

RxJS的fromFetch

结合 RxJS 的 fromFetch ，可以非常容易地和后端进行交互，完全是响应式的。

<script>
   const  data = fromFetch('https://httpbin.org/ip', {
    selector: response => response.json()
  });
</script>


<h2>Your IP: </h2>


{$data.origin}

当然你可以可以结合 RxJS 的 WebSocket 特性完成和 WebSocket 的交互。详细细节可以参考  https://rxjs-dev.firebaseapp.com/api/webSocket/webSocket。

RSocket

Svelte 和 RxJS 交互解决 UI 和基本的后端通讯是没有问题的，如 HTTP REST API 和 WebSocket 等，那么为何还需要 RSocket？让我们先看一下 RSocket 的通讯模型：

• request/response: 这也是典型的 HTTP 请求模型，也适用于 RPC 场景，当然这个通讯是异步化的，也适用 Promise 模型。

• request/stream: 流式数据请求，如 Pub/Sub 模型，可以非常方便地后端的数据流，如来自 Kafka 消息等。

• fireAndForget: 不需要返回确认的场景，如数据采集后提交后端的场景，这样速度更快。

• Channel：双向通讯，如 IM 聊天场景，可以同时实现消息的发送和接收。

标准的请求响应，如 HTTP REST API，可以使用 request/response；数据采集，我们可以使用 fireAndForget 做到性能极致；如果是消息订阅或者流式数据处理，你可以使用 request/stream；如果你要在 web 页面中添加即时消息(IM)聊天场景，使用 Channel 即可。一句话，借助于 RSocket 这一协议，可以让你实现各种通讯场景的需求。RSocket 完全是基于 Reactive 语义的，这样和 RxJS 和 Svelte 可以无缝衔接，不需要额外的转换操作。

此外 RSocket 还支持对等通讯，也就是通讯的双方或者多方，同时可以为 client 或者 server。我们都知道 Svelte Store 都是针对组件通讯的，如果 Svelte Store 和RSocket 整合，则可以实现不同页面之间的通讯，页面之间的协作场景就可以非常容易完成。

目前 RSocket 对 JavaScript 支持主要包括三个方面：RSocket Browser，在浏览器中连接后端 RSocket 服务；RSocket Node.js 可以快速创建后端 RSocket 服务；RSocket Deno 可以创建基于 Deno 的后端 RSocket 服务。

关于 Svelte 和 RSocket 通讯，可以参考 https://github.com/linux-china/svelte-rsocket-demo，详细的结构图如下：

RxJS 起着衔接 Svelte 和 RSocket 的功能，主要是 RxJS 本身的强大功能。所以接下来我们介绍一下即将发布的 RxJS 7.0 的新特性和功能。

RxJS 7.0 采用最新的 TS 版本(当前为 4.0.x)，这样可以使用 TypeScript 最新的特性，这样代码就整洁很多，当然可靠性和稳定性也提高很多。

RxJS toPromise 的调整

RxJS 的 Observable 是可以转换为 Promise 对象的，但是 API 让一些人有些模糊，主要的原因是 Observable 是流式的数据，所以在 RxJS 7.0 中使用 firstValueFrom, lastValueFrom 来替换 toPromise 函数，代码如下：

import {firstValueFrom, lastValueFrom} from "rxjs";


let result = await firstValueFrom(observable);

基于 AsyncIterable 构建 Observable

AsyncIterable 在异步化越来越重要， 如熟知的 for await...of 语句，就是在异步可迭代对象上创建一个迭代循环。在 RxJS 7.0中，我们可以基于 AsyncIterable 创建 Observable 对象，然后就可以利用 Observable 强大的功能来处理异步可以迭代对象列表。

from(iterable).subscribe(x => console.log(x))

当然如果你想将Observable转换为AsyncIterables，可以参考Ben Lesh的  https://github.com/benlesh/rxjs-for-await  项目。

fromFetch 函数提示

fromFetch 做了提升，添加了一个 selector 函数，这样可以支持直接从 response 提取对应的信息，如转换为文本或者 json 对象等，之前需要进行 map 转换的，现在方便非常多。

fromFetch("http://httpbin.org/ip", {selector: response => response.json()});

RxJS 和 Deno

Deno 作为一个新的 JS 运行引擎，因其的安全性、内置 TypeScript 支持和分布式的 module 加载机制，越来越受到 JavaScript 开发者的关注。RxJS 是基于 TypeScript 编写的，所以 Deno 的原生 TypeScript 支持没有问题。

此外 Deno 的 API 设计遵循 Web APIs 规范，如 fetch 函数、WebSocket 类，这就让 RxJS 的 fromFetch 和 WebSocket 可以在 Deno 使用，可以直接在 Deno 中使用 RxJS 网络通讯相关的功能，此外 RSocket 也包括对 Deno 的支持，也可以使用 RSocket 进行网络通讯。

在 Deno 中使用 RxJS fromFetch 样例如下：

import {firstValueFrom} from "https://esm.sh/[email protected]?no-check";
import {fromFetch} from "https://esm.sh/[email protected]/fetch?no-check";


const ip = await firstValueFrom(fromFetch("http://httpbin.org/ip", {selector: response => response.json()}));
console.log(ip.origin);

RxJS 7.0 其他

如内存优化，使用内存更小，添加更多 operator 方法。当然大家也不用太担心兼容性的问题，目前 RxJS 7.0 只是增加一些 API，只会将一些 API 调整为废弃 (Deprecated) 状态，还不会删除掉，可能在 RxJS 8 中会进行删除，所以 RxJS 7.0 的兼容性不用太担心。

当然 RxJS 7.0 还有更多更新，详细信息请参考：https://github.com/ReactiveX/rxjs/blob/master/CHANGELOG.md