在 AngularJS 中使用 RxJS 实现 Websocket 实时数据交互

前言

在现代 Web 应用中，实时数据交互已经成为了许多场景的必要需求，而 Websocket 作为一种流行的实现方式，可以使用它来处理实时数据交互的需求。而使用 AngularJS 和 RxJS 的组合，我们可以快速、简单地实现 Websocket 实时数据交互。

RxJS 简介

RxJS 是 ReactiveX 的 JavaScript 版本，它是一种编写异步和基于事件的代码的工具。RxJS 提供了丰富的函数式编程工具，使得处理并发、异步事件和数据流变得非常简单。

RxJS 通过 Observable 对象封装和处理不同的事件和数据流。Observable 可以组合和转化，这样我们就可以编写链式函数式编程代码。

创建 Websocket 连接

首先，我们需要使用 RxJS 创建一个 Websocket 的连接。可以使用 Rx.Observable.fromEvent 创建一个 Observable 对象，来监听 Websocket 的 'open' 事件，从而创建 Websocket 连接：

import { Observable } from 'rxjs/Observable';

const websocket: WebSocket = new WebSocket('ws://localhost:8080');

const openEventStream$ = Observable.fromEvent(websocket, 'open');
openEventStream$
  .subscribe(() => {
    console.log('WebSocket connection open');
  });

在这个例子中，使用 Rx.Observable.fromEvent 方法来创建一个 Observable 对象，监听 Websocket 的打开事件。然后使用 subscribe 方法，将回调函数绑定到 Observable 对象上，以便在 Websocket 连接打开时执行该回调函数。

发送和接收消息

一旦建立了 Websocket 连接，我们就可以发送和接收消息了。我们可以使用 Websocket 的 send 方法和 RxJS 的 fromEvent 方法来实现该功能。

这里顺带说一下，当我们使用 send 方法发送消息时，消息会异步地发送到 Websocket，这意味着我们需要等待 Websocket 的响应，以确定消息是否已成功发送到对端。假如我们需要一系列的异步操作完成后再执行下一步操作，这时我们可以使用 RxJS 的 merge 方法来合并多个 Observable 对象，实现并行执行多个异步操作。

const message = 'Hello World!';
websocket.send(message);

const messageEvent$ = Observable.fromEvent(websocket, 'message')
  .map((event: MessageEvent) => event.data);
messageEvent$
  .subscribe((data: any) => {
    console.log('Received message:', data);
  });

在这个例子中，我们首先使用 Websocket 的 send 方法发送一个消息。然后使用 RxJS 的 fromEvent 方法来创建一个 Observable 对象，监听从 Websocket 接收到的消息。这里通过 map 操作符将 MessageEvent 对象转化为消息内容，最后将接收到的消息打印到控制台。

处理异常和关闭连接

在使用 Websocket 进行实时数据交互时，经常要处理连接关闭或发生异常的情况。这时我们可以使用 RxJS 的 catchError 和 finalize 方法，来处理 Websocket 关闭或出现异常的情况。

const openEventStream$ = Observable.fromEvent(websocket, 'open');

const closeEventStream$ = Observable.fromEvent(websocket, 'close')
  .flatMap(() => Observable.throw(new Error('Websocket connection closed')));

const errorEventStream$ = Observable.fromEvent(websocket, 'error')
  .flatMap((error: any) => Observable.throw(new Error(`Websocket encountered error: ${error.message}`)));

const messageEvent$ = Observable.fromEvent(websocket, 'message')
  .map((event: MessageEvent) => event.data);

openEventStream$
  .subscribe(() => {
    console.log('WebSocket connection open');
  });

merge(closeEventStream$, errorEventStream$)
  .pipe(
    catchError((error: any) => {
      console.error(error);
      return of(null);
    }),
    finalize(() => {
      console.log('WebSocket connection closed');
    }),
  )
  .subscribe(() => {
    // do something on connection closed or error
  });

在这个例子中，我们使用 fromEvent 方法监听 Websocket 的 'close' 和 'error' 事件，并使用 flatMap 和 throw 方法来创建一个错误 Observable 对象。在 Observable 对象最后的 subscribe 中，使用 catchError 方法来处理抛出的错误，最后使用 finalize 方法在 Websocket 关闭时执行清理任务。

总结

使用 AngularJS 和 RxJS 的组合可以非常方便地实现 Websocket 的实时数据交互。同时 RxJS 提供了丰富的函数式编程工具，可以大大简化异步编程和数据流处理的复杂度，从而使得代码更加简单、易于维护。

完整示例代码请见 Github。

致谢

本文的示例代码参考了 Angular 4 使用Rx.js实现WebSocket实时信息推送。

来源：JavaScript中文网 ，转载请注明来源 本文地址：https://www.javascriptcn.com/post/65a1dca1add4f0e0ff9f7ed3