【JavaScript】论一个低配版Web实时通信库是如何实现的之二（ EventSource篇)-白红宇

【JavaScript】论一个低配版Web实时通信库是如何实现的之二（ EventSource篇)

阅读量：5015 次

发布时间：2019-06-12

本文共 4577 字，大约阅读时间需要 15 分钟。

前情提要

「话说上回说到！那WebSocket大侠,巧借http之内力，破了敌阵的双工鸳鸯锁，终于突出重围。

然而玄难未了，此时web森林中飞出一只银头红缨枪，划破夜

"莫非!?" websocket大侠喃喃念道，"恐怖如斯，你莫不是就是那个手使单向追魂枪的。。。"

"正是在下！"，那人厉声喝道。只见那胸前的纹章铭刻着几个洋文——

读作"EventSource"！」

上一篇文章请看这里：

引论

simple-socket是我写的一个"低配版"的Web实时通信工具（），在参考了相关源码和资料的基础上，实现了前后端实时互通的基本功能，选用了WebSocket ->server-sent-event -> AJAX轮询这三种方式做降级兼容，分为simple-socket-client和simple-socket-server两套代码。

我的上一篇文章讲了，所以今天来聊一聊event-source这块的

论一个低配版Web实时通信库是如何实现的（ WebSocket篇)

github仓库地址

https://github.com/penghuwan/simple-socket

npm命令

npm i simple-socket-serve   （服务端npm包）npm i simple-socket-client   (客户端npm包)

EventSource的前端代码

EventSource的前端API主要有这么四个

创建es对象：var es = new EventSource(url)

es两端连接事件打开的回调：es.onopen = function () { }

监听服务端发送事件： es.addEventListener("XXX", function (e) { // }

监听服务端的message事件es.onmessage = function; 相当于es.addEventListener("message"，function);

业务代码如下

（1）前端从服务端接收消息

前端通过监听服务端message事件，接收消息，并解析event和data，然后通过emitter.emit(event, data)触发事件，从而调用socket.on设置的监听回调

function Client() {  this.ws = null  this.es = null;   // EventSource对象  init.call(this);  // 设置this.type并初始化相关对象例如es或ws  listen.call(this);  // ...}function listen() {   // 保存this  var self  = this;  switch (this.type) {    // 当type为eventsource时，执行以下代码，this.type根据能力检测设置    case 'eventsource':      // 监听触发connect事件,把client对象自身传入当作socket      this.es.onopen = function () {        emitter.emit('connect', self);      };      // 监听服务端传来的message事件      this.es.addEventListener("message", function (e) {        var payload = JSON.parse(e.data);;        var event = payload.event;        var data = payload.data;        emitter.emit(event, data);      }, false);      break;     // ...  }}

（2）前端发送消息给服务端

由于event-source是单向的，只能从服务端从前端发送消息，而不能从前端发送消息给服务端。这和websocket显著不同

不过别担心，因为我们不是还有AJAX嘛！

对于前端发送消息的情况我们可以发一个post请求过去，同时借助/eventsource这个路径，告诉服务端这是一个SSE请求

$.ajax({,  type: 'POST',  url: `http://${url}/eventsource`,  data: { event, data },  success: function () {  }});

EventSource的服务端代码

好像这波就没了吧，OK，我们接下来走下路。

server-sent-event的服务端握手流程

server-sent-event(或event-source)，需要借助流(stream)的方式去实现通信。

Stream 是一个抽象接口，Node 中有很多对象实现了这个接口。例如，对http 服务器的request/response 对象就是一个 Stream。

它可以分为四种类型：

Readable - 可读操作。

Writable - 可写操作。

Duplex - 可读可写操作.

Transform - 操作被写入数据，然后读出结果。

服务器每次接收的Response是一个Writable，它可以被写入数据，将一个流写入另一个流可以通过调用pipe方法。

所以我们需要创建一个stream的实例，然后通过调用stream.pipe(Response)将流写入响应中，这样就可以被前端es.addEventListener添加的回调给接收到了。

但问题在于。。。Stream是个抽象接口，Node.js没有给Stream提供构造函数

不过没关系，我们可以这样做：

- 使用call方法继承stream父函数
- 使用util.inherits继承stream的原型
- 重写_read和_write方法（否则会报错）

// 因为我们的流需要写和读，所以使用双工的stream.Duplex构造function EventStream() { stream.Duplex.call(this);  // 构造函数继承}util.inherits(EventStream, stream.Duplex); // 原型继承// 重写_read和_write方法EventStream.prototype._read = function () { }EventStream.prototype._write = function () { }

握手代码逻辑

创建stream实例，调用pipe方法输送给Response, 同时stream我们保存在socket对象中，在向前端发送数据时候会使用

将Content-Type字段设置为'text/event-stream'，同时Connection设置为'keep-alive'

将状态码设为200（否则前端onopen方法不会触发）

_handleEShandShake(ctx, socket) {  // 前面定义好的类似stream的类  const eventStream = new EventStream();  // 设置eventStream  socket.setEventStream(eventStream);  // 握手成功后触发onConnection方法,TODO  // 设置符合Event-Source要求的首部  ctx.set({    'Content-Type': 'text/event-stream',    'Cache-Control': 'no-cache',    'Connection': 'keep-alive',  });  // 将Stream赋给body,Koa底层会判断Stream类型并调用pipe方法流入response  ctx.body = eventStream;  // 设置表示请求成功，否则前端onopen方法不会触发    ctx.status = 200;  // 触发connect方法，传递socket对象  this.emit('connect', socket);}

Event-Source服务端向前端发送消息。

这里要先说下event-source的报文结构了，由四种字段组成

event：事件名，对应前端es.addEventLisener设置的事件名

data：数据，为字符串

id: 消息标识符，可以缺省

retry：表示重新连接的时间间隔

这四个字段两两之间用\n分开，而最后一个字段值需要用\n\n做结尾

例如：`event:message\n data: XXX \n\n`

话不多说，看代码

class Socket extends events.EventEmitter {  constructor(socketId) {    super();  }  // 设置  setEventStream(eventStream) {    this.eventStream = eventStream;  }  // 自定义的emit,触发的是前端的on  emit(event, data) {    const dataStr = JSON.stringify({event,data})    if (this.transport === 'eventsource') {      if (!this.eventStream) { throw new Error('eventStream不存在,无法emit') };      // 向stream中写入数据，只要stream尚未关闭      // 数据就会传给前端的onmessage方法或addEventListener('message',fuc)方法      this.eventStream.push(`event:message\ndata:${dataStr}\n\n`);    }   }}